versioning_systems.py

import logging
import subprocess
import typing as t
from pathlib import Path
from shutil import which

from codecov_cli.fallbacks import FallbackFieldEnum
from codecov_cli.helpers.git import parse_git_service, parse_slug

logger = logging.getLogger("codecovcli")


class VersioningSystemInterface(object):
    def __repr__(self) -> str:
        return str(type(self))

    def get_fallback_value(self, fallback_field: FallbackFieldEnum) -> t.Optional[str]:
        pass

    def get_network_root(self) -> t.Optional[Path]:
        pass

    def list_relevant_files(self, directory: t.Optional[Path] = None) -> t.List[str]:
        pass


def get_versioning_system() -> t.Optional[VersioningSystemInterface]:
    for klass in [GitVersioningSystem, NoVersioningSystem]:
        if klass.is_available():
            logger.debug(f"versioning system found: {klass}")
            return klass()


class GitVersioningSystem(VersioningSystemInterface):
    @classmethod
    def is_available(cls):
        if which("git") is not None:
            p = subprocess.run(
                ["git", "rev-parse", "--show-toplevel"], capture_output=True
            )
            if p.stdout:
                return True
        return False

    def get_fallback_value(self, fallback_field: FallbackFieldEnum):
        if fallback_field == FallbackFieldEnum.commit_sha:
            # here we will get the commit SHA of the latest commit
            # that is NOT a merge commit
            p = subprocess.run(
                # List current commit parent's SHA
                ["git", "rev-parse", "HEAD^@"],
                capture_output=True,
            )
            parents_hash = p.stdout.decode().strip().splitlines()
            if len(parents_hash) == 2:
                # IFF the current commit is a merge commit it will have 2 parents
                # We return the 2nd one - The commit that came from the branch merged into ours
                return parents_hash[1]
            # At this point we know the current commit is not a merge commit
            # so we get it's SHA and return that
            p = subprocess.run(["git", "log", "-1", "--format=%H"], capture_output=True)
            if p.stdout:
                return p.stdout.decode().strip()

        if fallback_field == FallbackFieldEnum.branch:
            p = subprocess.run(
                ["git", "rev-parse", "--abbrev-ref", "HEAD"], capture_output=True
            )
            if p.stdout:
                branch_name = p.stdout.decode().strip()
                # branch_name will be 'HEAD' if we are in 'detached HEAD' state
                return branch_name if branch_name != "HEAD" else None

        if fallback_field == FallbackFieldEnum.slug:
            # if there are multiple remotes, we will prioritize using the one called 'origin' if it exists, else we will use the first one in 'git remote' list

            p = subprocess.run(["git", "remote"], capture_output=True)

            if not p.stdout:
                return None

            remotes = p.stdout.decode().strip().splitlines()

            remote_name = "origin" if "origin" in remotes else remotes[0]

            p = subprocess.run(
                ["git", "ls-remote", "--get-url", remote_name], capture_output=True
            )
            if not p.stdout:
                return None

            remote_url = p.stdout.decode().strip()

            return parse_slug(remote_url)

        if fallback_field == FallbackFieldEnum.git_service:
            # if there are multiple remotes, we will prioritize using the one called 'origin' if it exists, else we will use the first one in 'git remote' list

            p = subprocess.run(["git", "remote"], capture_output=True)
            if not p.stdout:
                return None

            remotes = p.stdout.decode().strip().splitlines()
            remote_name = "origin" if "origin" in remotes else remotes[0]
            p = subprocess.run(
                ["git", "ls-remote", "--get-url", remote_name], capture_output=True
            )
            if not p.stdout:
                return None

            remote_url = p.stdout.decode().strip()
            return parse_git_service(remote_url)

        return None

    def get_network_root(self):
        p = subprocess.run(["git", "rev-parse", "--show-toplevel"], capture_output=True)
        if p.stdout:
            return Path(p.stdout.decode().rstrip())
        return None

    def list_relevant_files(self, root_folder: t.Optional[Path] = None) -> t.List[str]:
        dir_to_use = root_folder or self.get_network_root()
        if dir_to_use is None:
            raise ValueError("Can't determine root folder")

        res = subprocess.run(
            ["git", "-C", str(dir_to_use), "ls-files"], capture_output=True
        )

        return [
            (
                filename[1:-1]
                if filename.startswith('"') and filename.endswith('"')
                else filename
            )
            for filename in res.stdout.decode("unicode_escape").strip().split("\n")
        ]


class NoVersioningSystem(VersioningSystemInterface):
    @classmethod
    def is_available(cls):
        return True

    def get_network_root(self):
        return Path.cwd()