cornsnake.util_git

  1"""
  2Functions for interacting with a Git repository. It includes functions for executing Git commands, checking out branches, handling commits, and managing Git configuration settings.
  3
  4[Documentation](http://docs.mrseanryan.cornsnake.s3-website-eu-west-1.amazonaws.com/cornsnake/util_git.html)
  5"""
  6
  7import os
  8
  9from . import config
 10from . import util_date
 11from . import util_file
 12from . import util_list
 13from . import util_print
 14from . import util_log
 15from . import util_proc
 16
 17COMMIT_PARTS_SEPARTOR = "_||_"
 18
 19logger = util_log.getLogger(__name__)
 20
 21
 22def _parse_result(result):
 23    """
 24    Function to parse the result obtained from executing Git commands.
 25
 26    Args:
 27    result (str): The result string to be parsed.
 28
 29    Returns:
 30    list: A list of cleaned parts obtained by splitting the result string.
 31    """
 32    parts = result.split(COMMIT_PARTS_SEPARTOR)
 33    parts_cleaned = []
 34    for part in parts:
 35        parts_cleaned.append(part.replace('"', ""))
 36    return parts_cleaned
 37
 38
 39def get_branches_with_commits_in_origin_not_local(repo_dir, branches):
 40    """
 41    For all the given branches, find any missing commits: commits that are on origin but not this copy of the repository.
 42    """
 43    if not has_git_remote_urls(repo_dir):
 44        util_print.print_warning(
 45            f"Cannot check for missing commits: the local repo {repo_dir} has no remote origin"
 46        )
 47        return []
 48    branches_with_missing_commits = []
 49    fetch(repo_dir)
 50    for branch in branches:
 51        checkout_branch(branch, repo_dir)
 52        result = execute_command("rev-list", [f"HEAD..origin/{branch}"], repo_dir)
 53        if result.strip():
 54            branches_with_missing_commits.append(branch)
 55    return branches_with_missing_commits
 56
 57
 58def get_current_branch(local_repo_location):
 59    """
 60    Function to get the current branch of the Git repository.
 61
 62    Args:
 63    local_repo_location (str): The local repository location.
 64
 65    Returns:
 66    str: The name of the current branch.
 67    """
 68    return execute_command(
 69        "rev-parse", ["--abbrev-ref", "HEAD"], local_repo_location
 70    ).strip()
 71
 72
 73def get_last_commit_id(path_to_repo_dir, file_only=None):
 74    """Get the last commit ID based on the given path to the repository directory and an optional file parameter.
 75    Args:
 76    path_to_repo_dir (str): The path to the repository directory.
 77    file_only (str): Optional parameter to specify a specific file.
 78    Returns:
 79    str: The last commit ID."""
 80    # git log -n 1 --pretty=format:"%H_|\_%ad_|\_%s" --date=short BUILD.plz
 81    git_args = [
 82        "log",
 83        "-n",
 84        "1",
 85        f'--pretty=format:"%H{COMMIT_PARTS_SEPARTOR}%ad{COMMIT_PARTS_SEPARTOR}%s"',
 86        "--date=short",
 87    ]
 88    if file_only is not None:
 89        git_args += [file_only]
 90    result = util_proc.run_process_and_get_output(
 91        config.PATH_TO_GIT, git_args, path_to_repo_dir
 92    )
 93    if config.IS_VERBOSE:
 94        print("  " + result)
 95    parsed = _parse_result(result)
 96    commit = parsed[0]
 97    return commit
 98
 99
100def has_git_remote_urls(repo_dir):
101    """
102    Does the given git repository directory have any remote origin URLs.
103    """
104    result = execute_command("remote", ["-v"], repo_dir)
105    return len(result.strip()) > 0
106
107
108def is_git_repo(repo_dir):
109    """
110    Is the given directory a git repository (contains .git folder)
111    """
112    path_to_git_config = os.path.join(repo_dir, ".git", "config")
113    return util_file.does_file_exist(path_to_git_config)
114
115
116def list_git_remote_urls(repo_dir):
117    """
118    List any remote origin URLs of the given git repository directory.
119    """
120    if not has_git_remote_urls(repo_dir):
121        return []
122
123    remote_urls = []
124    # git remote -v
125    result = execute_command("remote", ["-v"], repo_dir)
126    lines = result.split("\n")
127
128    for line in lines:
129        if not line.strip():
130            continue
131        # example:
132        # origin  https://git.api.mendix.com/b7a238f5-07a8-4008-83a7-b98590f40969.git/ (fetch)
133        parts = line.split()
134        if len(parts) == 3:
135            remote_urls.append(parts[1])
136        else:
137            raise RuntimeError(f"Cannot parse output of git remote: {line}")
138
139    return remote_urls
140
141
142def get_git_remote_push_url(repo_dir):
143    """
144    Get the remote origin *push* URL of the given git repository directory.
145    """
146    try:
147        # git remote get-url --push origin
148        return execute_command(
149            "remote", ["get-url", "--push", "origin"], repo_dir
150        ).rstrip()
151    except Exception as e:
152        util_log.log_exception(e)
153        return ""
154
155
156def execute_command(command, git_args, working_dir):
157    """Execute a Git command with specified arguments in the given working directory.
158    Args:
159    command (str): The Git command to execute.
160    git_args (list): List of arguments for the Git command.
161    working_dir (str): The working directory to execute the command in.
162    Returns:
163    str: The output of the command."""
164    git_args_with_command = [command] + git_args
165    return util_proc.run_process_and_get_output(
166        config.PATH_TO_GIT, git_args_with_command, working_dir
167    )
168
169
170# execute git command with too many parameters - so we chunk
171def execute_command_in_chunks(command, git_args, git_extra_args_to_chunk, working_dir):
172    """Execute a Git command with arguments provided in chunks to avoid exceeding the parameter limit.
173    Args:
174    command (str): The Git command to execute.
175    git_args (list): List of arguments for the Git command.
176    git_extra_args_to_chunk (list): List of additional arguments to chunk and execute.
177    working_dir (str): The working directory to execute the command in."""
178    CHUNK_SIZE = 20
179    chunks = util_list.chunk(git_extra_args_to_chunk, CHUNK_SIZE)
180    for chunk in chunks:
181        execute_command(command, git_args + chunk, working_dir)
182
183
184def fetch(working_dir):
185    """Fetch latest commits from a Git repository using the specified working directory.
186
187    Args:
188    working_dir (str): The working directory of the repository."""
189    execute_command("fetch", [], working_dir)
190
191
192def fetch_notes(working_dir):
193    """Fetch notes from a Git repository using the specified working directory.
194    Args:
195    working_dir (str): The working directory of the repository."""
196    execute_command(
197        "fetch", ["origin", "refs/notes/*:refs/notes/*", "--force"], working_dir
198    )
199
200
201def checkout_branch(branch, path_to_repo_dir):
202    """Check out the specified branch in the Git repository at the given path.
203    Args:
204    branch (str): The branch to check out.
205    path_to_repo_dir (str): The path to the repository directory.
206    Returns:
207    str: The result of the checkout command."""
208    result = execute_command("checkout", [branch], path_to_repo_dir)
209    return result
210
211
212def checkout_at_start_of_date(local_repo_location, branch, start_date):
213    """Check out the branch at the start of the specified date in the local repository location.
214    Args:
215    local_repo_location (str): The local repository location.
216    branch (str): The branch to check out.
217    start_date (str): The start date for checking out the branch."""
218    # git rev-list -n 1 --first-parent --before="2023-12-22 00:00" main
219    last_commit = execute_command(
220        "rev-list",
221        ["-n", "1", "--first-parent", f'--before="{start_date} 00:00"', branch],
222        local_repo_location,
223    )
224    last_commit = last_commit.strip()
225    if last_commit:
226        checkout_branch(last_commit, local_repo_location, True)
227        return True
228    return False
229
230
231def checkout_head(local_repo_location, branch):
232    """Check out the HEAD of the specified branch in the local repository location.
233    Args:
234    local_repo_location (str): The local repository location.
235    branch (str): The branch to check out."""
236    checkout_branch(branch, local_repo_location)
237
238
239def check_has_no_changes(path_to_local_repo):
240    """Check if the local repository at the specified path has any changes.
241    Args:
242    path_to_local_repo (str): The path to the local repository directory."""
243    # should return empty, if no changes
244    result = execute_command("status", ["--porcelain"], path_to_local_repo)
245    if len(result) > 0:
246        message = f"Local repository at {path_to_local_repo} has changes. Please ensure the local repository is clean and has no outstanding changes."
247        raise RuntimeError(message)
248
249
250def _prepare_local_clone(path_to_repo_dir, temp_git_fixer_dir, is_mirror):
251    """Prepare a local clone of the repository with the specified parameters.
252    Args:
253    path_to_repo_dir (str): The path to the repository directory.
254    temp_git_fixer_dir (str): The temporary directory for fixing Git issues.
255    is_mirror (bool): A flag indicating whether the clone is a mirror.
256    Returns:
257    str: The path to the prepared local clone directory."""
258    local_clone_dir = os.path.join(
259        temp_git_fixer_dir, "lm"
260    )  # lm = local_mirror (keeping path short)
261    util_file.ensure_dir_exists(local_clone_dir)
262    args = [path_to_repo_dir]
263    if is_mirror:
264        args = ["--mirror"] + args
265    execute_command("clone", args, local_clone_dir)
266
267    local_repo_name = util_file.get_last_part_of_path(path_to_repo_dir)
268    if is_mirror:
269        local_repo_name += ".git"
270    else:
271        if local_repo_name.endswith(".git"):
272            local_repo_name = local_repo_name[:-4]
273    return os.path.join(local_clone_dir, local_repo_name)
274
275
276def prepare_local_full_clone(path_to_repo_dir, temp_git_fixer_dir):
277    """Prepare a full local clone of the repository using the specified paths.
278    Args:
279    path_to_repo_dir (str): The path to the repository directory.
280    temp_git_fixer_dir (str): The temporary directory for fixing Git issues.
281    Returns:
282    str: The path to the prepared full local clone directory."""
283    return _prepare_local_clone(path_to_repo_dir, temp_git_fixer_dir, False)
284
285
286def prepare_local_mirror_clone(path_to_repo_dir, temp_git_fixer_dir):
287    """Prepare a mirror local clone of the repository using the specified paths.
288    Args:
289    path_to_repo_dir (str): The path to the repository directory.
290    temp_git_fixer_dir (str): The temporary directory for fixing Git issues.
291    Returns:
292    str: The path to the prepared mirror local clone directory."""
293    return _prepare_local_clone(path_to_repo_dir, temp_git_fixer_dir, True)
294
295
296def gc_expire_reflog(repo_dir):
297    """Expire the reflog in the repository to aid in garbage collection.
298    Args:
299    repo_dir (str): The repository directory path."""
300    # Expires the reflog, which helps a following 'gc prune' to remove more garbage
301    execute_command("reflog", ["expire", "--expire=now", "--all"], repo_dir)
302
303
304def gc_prune_aggressive(mirror_clone_dir):
305    """Aggressively prune the repository to remove garbage using the mirror clone directory.
306    Args:
307    mirror_clone_dir (str): The mirror clone directory path.
308    Returns:
309    str: The result of the pruning operation."""
310    result = execute_command("gc", ["--prune=now", "--aggressive"], mirror_clone_dir)
311    return result
312
313
314def gc_prune_now(repo_dir):
315    """Prune the repository immediately to remove unnecessary objects.
316    Args:
317    repo_dir (str): The repository directory path."""
318    execute_command("gc", ["--prune=now"], repo_dir)
319
320
321def gc_prune_safe(repo_dir):
322    """Safely prune the repository to optimize object storage.
323    Args:
324    repo_dir (str): The repository directory path."""
325    execute_command("gc", [], repo_dir)
326
327
328def remove_origin_and_remote_branches(repo_dir):
329    """Remove the origin and remote branches from the repository.
330    Args:
331    repo_dir (str): The repository directory path."""
332    execute_command("remote", ["remove", "origin"], repo_dir)
333
334
335def remove_tag(repo_dir, tag):
336    """Remove the specified tag from the repository.
337    Args:
338    repo_dir (str): The repository directory path.
339    tag (str): The tag to be removed."""
340    execute_command("tag", ["-d", tag], repo_dir)
341
342
343def get_commits_after_date(repo_dir, branches, start_date):
344    """Get all commits after the specified date in the given branches of the repository.
345    Args:
346    repo_dir (str): The repository directory path.
347    branches (list): List of branches to check commits from.
348    start_date (str): The start date for filtering commits.
349    Returns:
350    list: A list of commits after the specified date."""
351    commits = []
352    for branch in branches:
353        checkout_branch(branch, repo_dir)
354
355        # git log --pretty=format:"%H" --after="2023-01-31"
356        day_before_cutoff = util_date.add_day_to_date(start_date, -1)
357        result = execute_command(
358            "log", ['--pretty=format:"%H"', f'--after="{day_before_cutoff}"'], repo_dir
359        )
360        commits += result.split("\n")
361    cleaned_commits = []
362    for commit in commits:
363        cleaned = commit.replace('"', "").strip()
364        cleaned_commits.append(cleaned)
365    return cleaned_commits
366
367
368def get_git_text_editor_or_none(local_repo_location):
369    """Get the default Git text editor or return None if not found.
370    Args:
371    local_repo_location (str): The local repository location.
372    Returns:
373    tuple: A tuple containing the path to the text editor program and its arguments, or (None, None) if not found."""
374    path_to_text_editor = get_config("core.editor", local_repo_location)
375    if path_to_text_editor is None or len(path_to_text_editor) == 0:
376        return (None, None)
377    error_message = f"Cannot parse default git text editor ('core.editor' = '{path_to_text_editor}')"
378    # Unfortunately, value can be like '"program" command'. To be able to execute 'program', we need to split that out:
379    program = path_to_text_editor
380    args = []
381    try:
382        if '"' in path_to_text_editor:
383            if path_to_text_editor.count('"') == 2:
384                parts = path_to_text_editor.split(
385                    '"'
386                )  # '"path to exe" cmd' -> ['', 'path to exe', ' cmd']
387                parts = util_list.strip_strings(parts)
388                program = parts[1]
389                args = parts[2:]
390    except Exception as e:
391        util_log.log_exception(e)
392        # Intentionally NOT passing exception onwards
393    if os.path.exists(program):
394        return (program, args)
395    util_print.print_custom(error_message)
396    return (None, None)
397
398
399def get_all_tags(path_to_local_repo):
400    """Get all tags from the specified local repository.
401    Args:
402    path_to_local_repo (str): The path to the local repository directory.
403    Returns:
404    list: A list of all tags in the repository."""
405    tags = []
406    result = execute_command("tag", [], path_to_local_repo)
407    raw_tags = result.split("\n")
408    for raw_tag in raw_tags:
409        raw_tag = raw_tag.strip()
410        if raw_tag:
411            tags.append(raw_tag)
412    return tags
413
414
415def delete_all_tags(path_to_local_repo):
416    """Delete all tags from the specified local repository.
417    Args:
418    path_to_local_repo (str): The path to the local repository directory."""
419    tags = get_all_tags(path_to_local_repo)
420    execute_command_in_chunks(
421        "push", ["--delete", "origin", "--quiet"], tags, path_to_local_repo
422    )
423
424
425def get_all_origin_branches(path_to_local_repo):
426    """Get all origin branches from the specified local repository.
427    Args:
428    path_to_local_repo (str): The path to the local repository directory.
429    Returns:
430    list: A list of all origin branches in the repository."""
431    branches = []
432    result = execute_command("branch", ["-r"], path_to_local_repo)
433    raw_branches = result.split("\n")
434    for raw_branch in raw_branches:
435        if "origin" not in raw_branch:
436            continue
437        if "->" in raw_branch:
438            continue
439        raw_branch = raw_branch.strip()
440        raw_branch = raw_branch.removeprefix("origin/")
441        if raw_branch:
442            branches.append(raw_branch)
443    return branches
444
445
446def delete_branches_except(branches_to_keep, path_to_local_repo):
447    """Delete branches in the local repository except for the specified ones to keep.
448    Args:
449    branches_to_keep (list): List of branches to retain.
450    path_to_local_repo (str): The path to the local repository directory."""
451    # original bash:
452    #   git branch -r | grep -Eo 'origin/.*' | grep -v 'origin/main$' | sed 's/origin\///' | xargs git push origin -v --delete
453    # OR this one seems more reliable:
454    #   git branch -r | grep -Eo 'origin/.*' | grep -v 'origin/main$' | sed 's/origin\///' | xargs git push -d origin
455    branches = get_all_origin_branches(path_to_local_repo)
456    branches_to_delete = util_list.except_for(branches, branches_to_keep)
457
458    execute_command_in_chunks(
459        "push", ["-d", "origin"], branches_to_delete, path_to_local_repo
460    )
461
462
463def get_config(key, path_to_local_repo):
464    """Get the Git configuration setting value for the specified key in the local repository.
465    Args:
466    key (str): The configuration key to retrieve.
467    path_to_local_repo (str): The path to the local repository directory.
468    Returns:
469    str: The value of the configuration setting for the key."""
470    result = ""
471    try:
472        result = execute_command("config", ["--get", key], path_to_local_repo, False)
473    except RuntimeError as re:
474        # If the config key is missing, then git returns exit code 1 and empty stderr
475        if len(re.args) == 2 and re.args[1] == 1:
476            if config.IS_VERBOSE:
477                util_print.print_custom(
478                    f"[ok] git config has no value for setting '{key}'"
479                )
480            return ""
481        raise
482    return result
483
484
485def log_git_config(path_to_local_repo):
486    """Log interesting Git configuration settings that can impact text file handling.
487    Args:
488    path_to_local_repo (str): The path to the local repository directory."""
489    interesting_settings = [
490        "i18n.filesencoding",
491        "core.ignorecase",  # Probably best set to false.
492        "core.autocrlf",  # Caused encoding exception on validation (which we handle now). Probably best set to false - line-endings.
493        "diff.astextplain.textconv",
494    ]
495
496    logger.info("git config - text related settings")
497    for setting in interesting_settings:
498        value = get_config(setting, path_to_local_repo)
499        if len(value) > 0:
500            logger.info(f"  {setting}={value}")
501        else:
502            logger.info(f"  {setting} is not set [ok]")
503
504
505settings_best_set_to_false = [
506    "core.ignorecase",  # Probably best set to false.
507    "core.autocrlf",  # Caused encoding exception on validation (which we handle now). Probably best set to false - line-endings.
508]
509
510
511def is_git_ignored(directory, filename):
512    """Check if a file is ignored by Git in the specified directory.
513    Args:
514    directory (str): The directory containing the file.
515    filename (str): The name of the file to check.
516    Returns:
517    bool: True if the file is ignored, False otherwise."""
518    try:
519        execute_command("check-ignore", [filename], directory, False)
520    except RuntimeError as re:
521        if len(re.args) == 2:
522            exit_code = re.args[1]
523            if exit_code == 0:
524                # 0 - One or more of the provided paths is ignored.
525                return True
526            elif exit_code == 1:
527                # 1 - None of the provided paths are ignored.
528                return False
529            # 128 - A fatal error was encountered.
530            # any other exit code
531            raise
532    # 0 - One or more of the provided paths is ignored.
533    return True