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