path: root/tests/test_wrapper.py

# Copyright (C) 2015 The Android Open Source Project
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Unittests for the wrapper.py module."""

import io
import os
import re
import subprocess
import sys
from unittest import mock

import pytest
import utils_for_test

import main
import wrapper


@pytest.fixture(autouse=True)
def reset_wrapper() -> None:
    """Reset the wrapper module every time."""
    wrapper.Wrapper.cache_clear()


@pytest.fixture
def repo_wrapper() -> wrapper.Wrapper:
    """Fixture for the wrapper module."""
    return wrapper.Wrapper()


class GitCheckout:
    """Class to hold git checkout info for tests."""

    def __init__(self, git_dir, rev_list):
        self.git_dir = git_dir
        self.rev_list = rev_list


@pytest.fixture(scope="module")
def git_checkout(tmp_path_factory) -> GitCheckout:
    """Fixture for tests that use a real/small git checkout.

    Create a repo to operate on, but do it once per-test-run.
    """
    tempdir = tmp_path_factory.mktemp("repo-rev-tests")
    run_git = wrapper.Wrapper().run_git

    remote = os.path.join(tempdir, "remote")
    os.mkdir(remote)

    utils_for_test.init_git_tree(remote)
    run_git("commit", "--allow-empty", "-minit", cwd=remote)
    run_git("branch", "stable", cwd=remote)
    run_git("tag", "v1.0", cwd=remote)
    run_git("commit", "--allow-empty", "-m2nd commit", cwd=remote)
    rev_list = run_git("rev-list", "HEAD", cwd=remote).stdout.splitlines()

    run_git("init", cwd=tempdir)
    run_git(
        "fetch",
        remote,
        "+refs/heads/*:refs/remotes/origin/*",
        cwd=tempdir,
    )
    yield GitCheckout(tempdir, rev_list)


class TestRepoWrapper:
    """Tests helper functions in the repo wrapper"""

    def test_version(self, repo_wrapper: wrapper.Wrapper) -> None:
        """Make sure _Version works."""
        with pytest.raises(SystemExit) as e:
            with mock.patch("sys.stdout", new_callable=io.StringIO) as stdout:
                with mock.patch(
                    "sys.stderr", new_callable=io.StringIO
                ) as stderr:
                    repo_wrapper._Version()
        assert e.value.code == 0
        assert stderr.getvalue() == ""
        assert "repo launcher version" in stdout.getvalue()

    def test_python_constraints(self, repo_wrapper: wrapper.Wrapper) -> None:
        """The launcher should never require newer than main.py."""
        assert (
            main.MIN_PYTHON_VERSION_HARD >= repo_wrapper.MIN_PYTHON_VERSION_HARD
        )
        assert (
            main.MIN_PYTHON_VERSION_SOFT >= repo_wrapper.MIN_PYTHON_VERSION_SOFT
        )
        # Make sure the versions are themselves in sync.
        assert (
            repo_wrapper.MIN_PYTHON_VERSION_SOFT
            >= repo_wrapper.MIN_PYTHON_VERSION_HARD
        )

    def test_repo_script_is_executable(self) -> None:
        """The repo launcher script should be executable."""
        repo_path = utils_for_test.THIS_DIR.parent / "repo"
        assert os.access(repo_path, os.X_OK), f"{repo_path} is not executable"

    def test_init_parser(self, repo_wrapper: wrapper.Wrapper) -> None:
        """Make sure 'init' GetParser works."""
        parser = repo_wrapper.GetParser()
        opts, args = parser.parse_args([])
        assert args == []
        assert opts.manifest_url is None


class TestSetGitTrace2ParentSid:
    """Check SetGitTrace2ParentSid behavior."""

    KEY = "GIT_TRACE2_PARENT_SID"
    VALID_FORMAT = re.compile(r"^repo-[0-9]{8}T[0-9]{6}Z-P[0-9a-f]{8}$")

    def test_first_set(self, repo_wrapper: wrapper.Wrapper) -> None:
        """Test env var not yet set."""
        env = {}
        repo_wrapper.SetGitTrace2ParentSid(env)
        assert self.KEY in env
        value = env[self.KEY]
        assert self.VALID_FORMAT.match(value)

    def test_append(self, repo_wrapper: wrapper.Wrapper) -> None:
        """Test env var is appended."""
        env = {self.KEY: "pfx"}
        repo_wrapper.SetGitTrace2ParentSid(env)
        assert self.KEY in env
        value = env[self.KEY]
        assert value.startswith("pfx/")
        assert self.VALID_FORMAT.match(value[4:])

    def test_global_context(self, repo_wrapper: wrapper.Wrapper) -> None:
        """Check os.environ gets updated by default."""
        os.environ.pop(self.KEY, None)
        repo_wrapper.SetGitTrace2ParentSid()
        assert self.KEY in os.environ
        value = os.environ[self.KEY]
        assert self.VALID_FORMAT.match(value)


class TestRunCommand:
    """Check run_command behavior."""

    def test_capture(self, repo_wrapper: wrapper.Wrapper) -> None:
        """Check capture_output handling."""
        ret = repo_wrapper.run_command(["echo", "hi"], capture_output=True)
        # echo command appends OS specific linesep, but on Windows + Git Bash
        # we get UNIX ending, so we allow both.
        assert ret.stdout in ["hi" + os.linesep, "hi\n"]

    def test_check(self, repo_wrapper: wrapper.Wrapper) -> None:
        """Check check handling."""
        repo_wrapper.run_command(["true"], check=False)
        repo_wrapper.run_command(["true"], check=True)
        repo_wrapper.run_command(["false"], check=False)
        with pytest.raises(subprocess.CalledProcessError):
            repo_wrapper.run_command(["false"], check=True)


class TestRunGit:
    """Check run_git behavior."""

    def test_capture(self, repo_wrapper: wrapper.Wrapper) -> None:
        """Check capture_output handling."""
        ret = repo_wrapper.run_git("--version")
        assert "git" in ret.stdout

    def test_check(self, repo_wrapper: wrapper.Wrapper) -> None:
        """Check check handling."""
        with pytest.raises(repo_wrapper.CloneFailure):
            repo_wrapper.run_git("--version-asdfasdf")
        repo_wrapper.run_git("--version-asdfasdf", check=False)


class TestParseGitVersion:
    """Check ParseGitVersion behavior."""

    def test_autoload(self, repo_wrapper: wrapper.Wrapper) -> None:
        """Check we can load the version from the live git."""
        assert repo_wrapper.ParseGitVersion() is not None

    def test_bad_ver(self, repo_wrapper: wrapper.Wrapper) -> None:
        """Check handling of bad git versions."""
        assert repo_wrapper.ParseGitVersion(ver_str="asdf") is None

    def test_normal_ver(self, repo_wrapper: wrapper.Wrapper) -> None:
        """Check handling of normal git versions."""
        ret = repo_wrapper.ParseGitVersion(ver_str="git version 2.25.1")
        assert ret.major == 2
        assert ret.minor == 25
        assert ret.micro == 1
        assert ret.full == "2.25.1"

    def test_extended_ver(self, repo_wrapper: wrapper.Wrapper) -> None:
        """Check handling of extended distro git versions."""
        ret = repo_wrapper.ParseGitVersion(
            ver_str="git version 1.30.50.696.g5e7596f4ac-goog"
        )
        assert ret.major == 1
        assert ret.minor == 30
        assert ret.micro == 50
        assert ret.full == "1.30.50.696.g5e7596f4ac-goog"


class TestCheckGitVersion:
    """Check _CheckGitVersion behavior."""

    def test_unknown(self, repo_wrapper: wrapper.Wrapper) -> None:
        """Unknown versions should abort."""
        with mock.patch.object(
            repo_wrapper, "ParseGitVersion", return_value=None
        ):
            with pytest.raises(repo_wrapper.CloneFailure):
                repo_wrapper._CheckGitVersion()

    def test_old(self, repo_wrapper: wrapper.Wrapper) -> None:
        """Old versions should abort."""
        with mock.patch.object(
            repo_wrapper,
            "ParseGitVersion",
            return_value=repo_wrapper.GitVersion(1, 0, 0, "1.0.0"),
        ):
            with pytest.raises(repo_wrapper.CloneFailure):
                repo_wrapper._CheckGitVersion()

    def test_new(self, repo_wrapper: wrapper.Wrapper) -> None:
        """Newer versions should run fine."""
        with mock.patch.object(
            repo_wrapper,
            "ParseGitVersion",
            return_value=repo_wrapper.GitVersion(100, 0, 0, "100.0.0"),
        ):
            repo_wrapper._CheckGitVersion()


class TestRequirements:
    """Check Requirements handling."""

    def test_missing_file(self, repo_wrapper: wrapper.Wrapper) -> None:
        """Don't crash if the file is missing (old version)."""
        assert (
            repo_wrapper.Requirements.from_dir(utils_for_test.THIS_DIR) is None
        )
        assert (
            repo_wrapper.Requirements.from_file(
                utils_for_test.THIS_DIR / "xxxxxxxxxxxxxxxxxxxxxxxx"
            )
            is None
        )

    def test_corrupt_data(self, repo_wrapper: wrapper.Wrapper) -> None:
        """If the file can't be parsed, don't blow up."""
        assert repo_wrapper.Requirements.from_file(__file__) is None
        assert repo_wrapper.Requirements.from_data(b"x") is None

    def test_valid_data(self, repo_wrapper: wrapper.Wrapper) -> None:
        """Make sure we can parse the file we ship."""
        assert repo_wrapper.Requirements.from_data(b"{}") is not None
        rootdir = utils_for_test.THIS_DIR.parent
        assert repo_wrapper.Requirements.from_dir(rootdir) is not None
        assert (
            repo_wrapper.Requirements.from_file(rootdir / "requirements.json")
            is not None
        )

    def test_format_ver(self, repo_wrapper: wrapper.Wrapper) -> None:
        """Check format_ver can format."""
        assert repo_wrapper.Requirements._format_ver((1, 2, 3)) == "1.2.3"
        assert repo_wrapper.Requirements._format_ver([1]) == "1"

    def test_assert_all_unknown(self, repo_wrapper: wrapper.Wrapper) -> None:
        """Check assert_all works with incompatible file."""
        reqs = repo_wrapper.Requirements({})
        reqs.assert_all()

    def test_assert_all_new_repo(self, repo_wrapper: wrapper.Wrapper) -> None:
        """Check assert_all accepts new enough repo."""
        reqs = repo_wrapper.Requirements({"repo": {"hard": [1, 0]}})
        reqs.assert_all()

    def test_assert_all_old_repo(self, repo_wrapper: wrapper.Wrapper) -> None:
        """Check assert_all rejects old repo."""
        reqs = repo_wrapper.Requirements({"repo": {"hard": [99999, 0]}})
        with pytest.raises(SystemExit):
            reqs.assert_all()

    def test_assert_all_new_python(self, repo_wrapper: wrapper.Wrapper) -> None:
        """Check assert_all accepts new enough python."""
        reqs = repo_wrapper.Requirements({"python": {"hard": sys.version_info}})
        reqs.assert_all()

    def test_assert_all_old_python(self, repo_wrapper: wrapper.Wrapper) -> None:
        """Check assert_all rejects old python."""
        reqs = repo_wrapper.Requirements({"python": {"hard": [99999, 0]}})
        with pytest.raises(SystemExit):
            reqs.assert_all()

    def test_assert_ver_unknown(self, repo_wrapper: wrapper.Wrapper) -> None:
        """Check assert_ver works with incompatible file."""
        reqs = repo_wrapper.Requirements({})
        reqs.assert_ver("xxx", (1, 0))

    def test_assert_ver_new(self, repo_wrapper: wrapper.Wrapper) -> None:
        """Check assert_ver allows new enough versions."""
        reqs = repo_wrapper.Requirements(
            {"git": {"hard": [1, 0], "soft": [2, 0]}}
        )
        reqs.assert_ver("git", (1, 0))
        reqs.assert_ver("git", (1, 5))
        reqs.assert_ver("git", (2, 0))
        reqs.assert_ver("git", (2, 5))

    def test_assert_ver_old(self, repo_wrapper: wrapper.Wrapper) -> None:
        """Check assert_ver rejects old versions."""
        reqs = repo_wrapper.Requirements(
            {"git": {"hard": [1, 0], "soft": [2, 0]}}
        )
        with pytest.raises(SystemExit):
            reqs.assert_ver("git", (0, 5))


class TestNeedSetupGnuPG:
    """Check NeedSetupGnuPG behavior."""

    def test_missing_dir(self, tmp_path, repo_wrapper: wrapper.Wrapper) -> None:
        """The ~/.repoconfig tree doesn't exist yet."""
        repo_wrapper.home_dot_repo = str(tmp_path / "foo")
        assert repo_wrapper.NeedSetupGnuPG()

    def test_missing_keyring(
        self, tmp_path, repo_wrapper: wrapper.Wrapper
    ) -> None:
        """The keyring-version file doesn't exist yet."""
        repo_wrapper.home_dot_repo = str(tmp_path)
        assert repo_wrapper.NeedSetupGnuPG()

    def test_empty_keyring(
        self, tmp_path, repo_wrapper: wrapper.Wrapper
    ) -> None:
        """The keyring-version file exists, but is empty."""
        repo_wrapper.home_dot_repo = str(tmp_path)
        (tmp_path / "keyring-version").write_text("")
        assert repo_wrapper.NeedSetupGnuPG()

    def test_old_keyring(self, tmp_path, repo_wrapper: wrapper.Wrapper) -> None:
        """The keyring-version file exists, but it's old."""
        repo_wrapper.home_dot_repo = str(tmp_path)
        (tmp_path / "keyring-version").write_text("1.0\n")
        assert repo_wrapper.NeedSetupGnuPG()

    def test_new_keyring(self, tmp_path, repo_wrapper: wrapper.Wrapper) -> None:
        """The keyring-version file exists, and is up-to-date."""
        repo_wrapper.home_dot_repo = str(tmp_path)
        (tmp_path / "keyring-version").write_text("1000.0\n")
        assert not repo_wrapper.NeedSetupGnuPG()


class TestSetupGnuPG:
    """Check SetupGnuPG behavior."""

    def test_full(self, tmp_path, repo_wrapper: wrapper.Wrapper) -> None:
        """Make sure it works completely."""
        repo_wrapper.home_dot_repo = str(tmp_path)
        repo_wrapper.gpg_dir = str(tmp_path / "gnupg")
        assert repo_wrapper.SetupGnuPG(True)
        data = (tmp_path / "keyring-version").read_text()
        assert (
            ".".join(str(x) for x in repo_wrapper.KEYRING_VERSION)
            == data.strip()
        )


class TestVerifyRev:
    """Check verify_rev behavior."""

    def test_verify_passes(self, repo_wrapper: wrapper.Wrapper) -> None:
        """Check when we have a valid signed tag."""
        desc_result = subprocess.CompletedProcess([], 0, "v1.0\n", "")
        gpg_result = subprocess.CompletedProcess([], 0, "", "")
        with mock.patch.object(
            repo_wrapper, "run_git", side_effect=(desc_result, gpg_result)
        ):
            ret = repo_wrapper.verify_rev(
                "/", "refs/heads/stable", "1234", True
            )
            assert ret == "v1.0^0"

    def test_unsigned_commit(self, repo_wrapper: wrapper.Wrapper) -> None:
        """Check we fall back to signed tag when we have an unsigned commit."""
        desc_result = subprocess.CompletedProcess([], 0, "v1.0-10-g1234\n", "")
        gpg_result = subprocess.CompletedProcess([], 0, "", "")
        with mock.patch.object(
            repo_wrapper, "run_git", side_effect=(desc_result, gpg_result)
        ):
            ret = repo_wrapper.verify_rev(
                "/", "refs/heads/stable", "1234", True
            )
            assert ret == "v1.0^0"

    def test_verify_fails(self, repo_wrapper: wrapper.Wrapper) -> None:
        """Check we fall back to signed tag when we have an unsigned commit."""
        desc_result = subprocess.CompletedProcess([], 0, "v1.0-10-g1234\n", "")
        gpg_result = RuntimeError
        with mock.patch.object(
            repo_wrapper, "run_git", side_effect=(desc_result, gpg_result)
        ):
            with pytest.raises(RuntimeError):
                repo_wrapper.verify_rev("/", "refs/heads/stable", "1234", True)


class TestResolveRepoRev:
    """Check resolve_repo_rev behavior."""

    def test_explicit_branch(
        self,
        repo_wrapper: wrapper.Wrapper,
        git_checkout: GitCheckout,
    ) -> None:
        """Check refs/heads/branch argument."""
        rrev, lrev = repo_wrapper.resolve_repo_rev(
            git_checkout.git_dir, "refs/heads/stable"
        )
        assert rrev == "refs/heads/stable"
        assert lrev == git_checkout.rev_list[1]

        with pytest.raises(repo_wrapper.CloneFailure):
            repo_wrapper.resolve_repo_rev(
                git_checkout.git_dir, "refs/heads/unknown"
            )

    def test_explicit_tag(
        self,
        repo_wrapper: wrapper.Wrapper,
        git_checkout: GitCheckout,
    ) -> None:
        """Check refs/tags/tag argument."""
        rrev, lrev = repo_wrapper.resolve_repo_rev(
            git_checkout.git_dir, "refs/tags/v1.0"
        )
        assert rrev == "refs/tags/v1.0"
        assert lrev == git_checkout.rev_list[1]

        with pytest.raises(repo_wrapper.CloneFailure):
            repo_wrapper.resolve_repo_rev(
                git_checkout.git_dir, "refs/tags/unknown"
            )

    def test_branch_name(
        self,
        repo_wrapper: wrapper.Wrapper,
        git_checkout: GitCheckout,
    ) -> None:
        """Check branch argument."""
        rrev, lrev = repo_wrapper.resolve_repo_rev(
            git_checkout.git_dir, "stable"
        )
        assert rrev == "refs/heads/stable"
        assert lrev == git_checkout.rev_list[1]

        rrev, lrev = repo_wrapper.resolve_repo_rev(git_checkout.git_dir, "main")
        assert rrev == "refs/heads/main"
        assert lrev == git_checkout.rev_list[0]

    def test_tag_name(
        self,
        repo_wrapper: wrapper.Wrapper,
        git_checkout: GitCheckout,
    ) -> None:
        """Check tag argument."""
        rrev, lrev = repo_wrapper.resolve_repo_rev(git_checkout.git_dir, "v1.0")
        assert rrev == "refs/tags/v1.0"
        assert lrev == git_checkout.rev_list[1]

    def test_full_commit(
        self,
        repo_wrapper: wrapper.Wrapper,
        git_checkout: GitCheckout,
    ) -> None:
        """Check specific commit argument."""
        commit = git_checkout.rev_list[0]
        rrev, lrev = repo_wrapper.resolve_repo_rev(git_checkout.git_dir, commit)
        assert rrev == commit
        assert lrev == commit

    def test_partial_commit(
        self,
        repo_wrapper: wrapper.Wrapper,
        git_checkout: GitCheckout,
    ) -> None:
        """Check specific (partial) commit argument."""
        commit = git_checkout.rev_list[0][0:20]
        rrev, lrev = repo_wrapper.resolve_repo_rev(git_checkout.git_dir, commit)
        assert rrev == git_checkout.rev_list[0]
        assert lrev == git_checkout.rev_list[0]

    def test_unknown(
        self,
        repo_wrapper: wrapper.Wrapper,
        git_checkout: GitCheckout,
    ) -> None:
        """Check unknown ref/commit argument."""
        with pytest.raises(repo_wrapper.CloneFailure):
            repo_wrapper.resolve_repo_rev(git_checkout.git_dir, "boooooooya")


class TestCheckRepoVerify:
    """Check check_repo_verify behavior."""

    def test_no_verify(self, repo_wrapper: wrapper.Wrapper) -> None:
        """Always fail with --no-repo-verify."""
        assert not repo_wrapper.check_repo_verify(False)

    def test_gpg_initialized(
        self,
        repo_wrapper: wrapper.Wrapper,
    ) -> None:
        """Should pass if gpg is setup already."""
        with mock.patch.object(
            repo_wrapper, "NeedSetupGnuPG", return_value=False
        ):
            assert repo_wrapper.check_repo_verify(True)

    def test_need_gpg_setup(
        self,
        repo_wrapper: wrapper.Wrapper,
    ) -> None:
        """Should pass/fail based on gpg setup."""
        with mock.patch.object(
            repo_wrapper, "NeedSetupGnuPG", return_value=True
        ):
            with mock.patch.object(repo_wrapper, "SetupGnuPG") as m:
                m.return_value = True
                assert repo_wrapper.check_repo_verify(True)

                m.return_value = False
                assert not repo_wrapper.check_repo_verify(True)


class TestCheckRepoRev:
    """Check check_repo_rev behavior."""

    def test_verify_works(
        self,
        repo_wrapper: wrapper.Wrapper,
        git_checkout: GitCheckout,
    ) -> None:
        """Should pass when verification passes."""
        with mock.patch.object(
            repo_wrapper, "check_repo_verify", return_value=True
        ):
            with mock.patch.object(
                repo_wrapper, "verify_rev", return_value="12345"
            ):
                rrev, lrev = repo_wrapper.check_repo_rev(
                    git_checkout.git_dir, "stable"
                )
        assert rrev == "refs/heads/stable"
        assert lrev == "12345"

    def test_verify_fails(
        self,
        repo_wrapper: wrapper.Wrapper,
        git_checkout: GitCheckout,
    ) -> None:
        """Should fail when verification fails."""
        with mock.patch.object(
            repo_wrapper, "check_repo_verify", return_value=True
        ):
            with mock.patch.object(
                repo_wrapper, "verify_rev", side_effect=RuntimeError
            ):
                with pytest.raises(RuntimeError):
                    repo_wrapper.check_repo_rev(git_checkout.git_dir, "stable")

    def test_verify_ignore(
        self,
        repo_wrapper: wrapper.Wrapper,
        git_checkout: GitCheckout,
    ) -> None:
        """Should pass when verification is disabled."""
        with mock.patch.object(
            repo_wrapper, "verify_rev", side_effect=RuntimeError
        ):
            rrev, lrev = repo_wrapper.check_repo_rev(
                git_checkout.git_dir, "stable", repo_verify=False
            )
        assert rrev == "refs/heads/stable"
        assert lrev == git_checkout.rev_list[1]