Source code for mafw.devtools.git

#  Copyright 2026 European Union
#  Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
#  SPDX-License-Identifier: EUPL-1.2
"""
Git operations for MAFw development tools.

This module centralizes all functions that shell out to ``git`` commands,
used by the release workflow, documentation builder, and dependency management.
"""

from __future__ import annotations

import re
import subprocess
from pathlib import Path
from typing import Any

from mafw.devtools import DevtoolsError, ensure_devtools_available

ensure_devtools_available()

from packaging.version import InvalidVersion, Version  # noqa: E402

from mafw.tools.shell_tools import CONSOLE, run_stdout  # noqa: E402
from mafw.tools.shell_tools import run as cmd  # noqa: E402

PYPROJECT_FILE = Path('pyproject.toml')
"""Path to the TOML file containing the project dependencies."""

STABLE_TAG_PATTERN = re.compile(r'^v(?P<major>\d+)\.(?P<minor>\d+)\.(?P<micro>\d+)$')
"""Regular expression used to identify stable git tags in the form ``vX.Y.Z``."""




def check_main_branch() -> None:
    """Ensure the release process is executed from the ``main`` branch.

    :raises DevtoolsError: If the current branch is not ``main``.
    """
    branch = run_stdout('git rev-parse --abbrev-ref HEAD')
    if branch != 'main':
        raise DevtoolsError('Must be on main branch.')





def ensure_clean_git() -> None:
    """Ensure the git working tree is clean (excluding untracked files).

    :raises DevtoolsError: If tracked changes are present.
    """
    status = run_stdout('git status --porcelain -uno')
    if status:
        raise DevtoolsError('Git working tree is not clean. Commit or stash changes first.')





def get_last_stable_tag() -> str | None:
    """Return the latest stable tag matching ``vX.Y.Z``.

    :return: Latest stable tag or ``None`` when no stable tags are present.
    :rtype: str | None
    """
    versions = get_git_tags()
    # Filter to only strict vX.Y.Z tags (no pre/post/dev — already handled by get_git_tags)
    stable: list[tuple[Version, str]] = []
    for v, tag in versions:
        if STABLE_TAG_PATTERN.fullmatch(tag):
            stable.append((v, tag))
    if not stable:
        return None
    return stable[-1][1]





def prevent_duplicate_tag(version: str) -> None:
    """Ensure the target release tag does not already exist.

    :param version: Target release version.
    :type version: str
    :raises DevtoolsError: If ``v<version>`` already exists.
    """
    existing_tags = run_stdout('git tag').splitlines()
    if f'v{version}' in existing_tags:
        raise DevtoolsError(f'Tag v{version} already exists.')





def commit_changes(version: str, dry_run: bool, *, include_changelog: bool = True) -> None:
    """Commit tracked release artifacts for the target version.

    :param version: Target release version.
    :type version: str
    :param dry_run: Whether command execution is disabled.
    :type dry_run: bool
    :param include_changelog: Whether ``CHANGELOG.md`` should be staged and
        committed as part of the release artifacts.
    :type include_changelog: bool
    """
    from mafw.devtools.documentation.requirements import REQUIREMENTS_GROUPS
    from mafw.devtools.release.changelog import CHANGELOG_FILE
    from mafw.devtools.release.versioning import ABOUT_FILE, NOTICE_FILE

    CONSOLE.print('Committing tracked release artifacts...')
    tracked_files: list[str] = [str(ABOUT_FILE), str(NOTICE_FILE), str(PYPROJECT_FILE), 'README.rst']
    for group in REQUIREMENTS_GROUPS:
        tracked_files.append(f'docs/source/requirements/{group}_requirements.rst')

    if include_changelog:
        tracked_files.append(str(CHANGELOG_FILE))
    cmd(['git', 'add', *tracked_files], dry_run=dry_run)
    cmd(['git', 'commit', '-m', f'chore(release): v{version}'], dry_run=dry_run)





def create_tag(version: str, dry_run: bool) -> str:
    """Create the local git tag for the target version.

    :param version: Target release version.
    :type version: str
    :param dry_run: Whether command execution is disabled.
    :type dry_run: bool
    :return: Created tag name.
    :rtype: str
    """
    tag = f'v{version}'
    CONSOLE.print(f'Creating local tag {tag}...')
    cmd(['git', 'tag', tag], dry_run=dry_run)
    return tag





def push_changes(dry_run: bool) -> None:
    """Push commits and tags to the remote repository.

    :param dry_run: Whether command execution is disabled.
    :type dry_run: bool
    """
    CONSOLE.print('Pushing commits and tags...')
    cmd(['git', 'push', 'origin-ssh', 'main'], dry_run=dry_run)
    cmd(['git', 'push', '--tags'], dry_run=dry_run)





def commit_dependency_unfreeze(dry_run: bool) -> None:
    """Commit the unfreezing of dependency upper bounds to ``main``.

    :param dry_run: Whether command execution is disabled.
    :type dry_run: bool
    """
    from mafw.devtools.documentation.requirements import REQUIREMENTS_GROUPS

    CONSOLE.print('Committing dependency unfreeze...')
    tracked_files: list[str] = [str(PYPROJECT_FILE), 'README.rst']
    for group in REQUIREMENTS_GROUPS:
        tracked_files.append(f'docs/source/requirements/{group}_requirements.rst')

    cmd(['git', 'add', *tracked_files], dry_run=dry_run)
    cmd(['git', 'commit', '-m', 'chore(dependencies): unfreeze upper bounds'], dry_run=dry_run)





def get_git_tags(min_version: str | None = None) -> list[tuple[Version, Any]]:
    """Return list of (Version, tag) tuples sorted ascending.

    :param min_version: Minimum version to consider, defaults to None
    :type min_version: str | None
    :return: List of (Version, tag) tuples
    :rtype: list[tuple[Version, Any]]
    """
    proc = cmd(['git', 'tag'], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, check=False)
    tags = proc.stdout.split()
    versions = []
    for t in tags:
        try:
            v = Version(t)
            if v.is_prerelease or v.is_devrelease:
                continue
            if min_version and v < Version(min_version):
                continue
            versions.append((v, t))
        except InvalidVersion:
            continue
    versions.sort()
    return versions





def get_current_branch() -> str:
    """Get the name of the currently checked out branch.

    :return: Name of the current branch
    :rtype: str
    """
    proc = cmd(
        ['git', 'rev-parse', '--abbrev-ref', 'HEAD'], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, check=False
    )
    return str(proc.stdout.strip())





def sort_tags_semver(tags: list[str]) -> list[str]:
    """Sort tags using semantic versioning comparison.

    :param tags: List of tag strings to sort
    :type tags: list[str]
    :return: Sorted list of tag strings
    :rtype: list[str]
    """
    from mafw.devtools.documentation.builder import parse_version_tuple

    return sorted(tags, key=lambda t: parse_version_tuple(t))





def git_rev_of(ref: str) -> str:
    """Get the git revision hash for a given reference.

    :param ref: Git reference (tag, branch, commit hash)
    :type ref: str
    :return: Git revision hash
    :rtype: str
    :raises RuntimeError: If git rev-list fails
    """
    proc = cmd(['git', 'rev-list', '-n', '1', ref], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, check=False)
    if proc.returncode != 0:
        raise RuntimeError(f'git rev-list failed for {ref}:\n{proc.stdout}')
    return str(proc.stdout.strip())





def is_ancestor(a: str, b: str) -> bool:
    """Return True if commit a is ancestor of commit b (git merge-base --is-ancestor).

    :param a: First commit reference
    :type a: str
    :param b: Second commit reference
    :type b: str
    :return: True if a is ancestor of b
    :rtype: bool
    """
    proc = cmd(
        ['git', 'merge-base', '--is-ancestor', a, b], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, check=False
    )
    return proc.returncode == 0