| # Copyright 2018 The ChromiumOS Authors |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| """Unit tests for chromite.lib.repo and helpers for testing that module.""" |
| |
| import logging |
| import os |
| |
| from chromite.lib import constants |
| from chromite.lib import cros_build_lib |
| from chromite.lib import cros_test_lib |
| from chromite.lib import git |
| from chromite.lib import osutils |
| from chromite.lib import repo_util |
| |
| |
| def RepoInitSideEffects(*_args, **kwargs) -> None: |
| """Mimic side effects of `repo init` by creating .repo dir.""" |
| os.mkdir(os.path.join(kwargs["cwd"], ".repo")) |
| |
| |
| def CopySideEffects(dest) -> None: |
| """Mimic side effects of Repository.Copy by creating .repo dir.""" |
| os.mkdir(os.path.join(dest, ".repo")) |
| |
| |
| def RepoCmdPath(repo_root): |
| """Return the path to the repo command to use for the given repo root.""" |
| return os.path.join(repo_root, ".repo", "repo", "repo") |
| |
| |
| class RepositoryTest(cros_test_lib.RunCommandTempDirTestCase): |
| """Tests for repo_util.Repository.""" |
| |
| MANIFEST_URL = "https://example.com/manifest.xml" |
| |
| def setUp(self) -> None: |
| self.empty_root = os.path.join(self.tempdir, "empty") |
| self.empty_root_subdir = os.path.join(self.empty_root, "sub", "dir") |
| os.makedirs(self.empty_root_subdir) |
| self.repo_root = os.path.join(self.tempdir, "root") |
| self.repo_root_subdir = os.path.join(self.repo_root, "sub", "dir") |
| os.makedirs(self.repo_root_subdir) |
| self.repo_dir = os.path.join(self.repo_root, ".repo") |
| os.makedirs(self.repo_dir) |
| |
| def testInit(self) -> None: |
| """Test Repository.__init__.""" |
| repo = repo_util.Repository(self.repo_root) |
| self.assertTrue(os.path.samefile(repo.root, self.repo_root)) |
| |
| def testInitNoRepoDir(self) -> None: |
| """Test Repository.__init__ fails if not in repo.""" |
| with self.assertRaises(repo_util.NotInRepoError): |
| repo_util.Repository(self.empty_root) |
| |
| def testInitializeSimple(self) -> None: |
| """Test Repository.Initialize simple call.""" |
| expected_cmd = ["repo", "init", "--manifest-url", self.MANIFEST_URL] |
| self.rc.AddCmdResult(expected_cmd, side_effect=RepoInitSideEffects) |
| repo = repo_util.Repository.Initialize( |
| self.empty_root, self.MANIFEST_URL |
| ) |
| self.assertCommandCalled(expected_cmd, cwd=self.empty_root) |
| self.assertTrue(os.path.samefile(repo.root, self.empty_root)) |
| |
| def testInitializeComplex(self) -> None: |
| """Test Repository.Initialize complex call.""" |
| expected_cmd = [ |
| "repo", |
| "init", |
| "--manifest-url", |
| "http://manifest.xyz/manifest", |
| "--manifest-branch", |
| "test-branch", |
| "--manifest-name", |
| "other.xml", |
| "--mirror", |
| "--reference", |
| "/repo/reference", |
| "--depth", |
| "99", |
| "--groups", |
| "abba,queen", |
| "--repo-url", |
| "https://repo.xyz/repo", |
| "--repo-branch", |
| "repo-branch", |
| ] |
| self.rc.AddCmdResult(expected_cmd, side_effect=RepoInitSideEffects) |
| repo = repo_util.Repository.Initialize( |
| self.empty_root, |
| "http://manifest.xyz/manifest", |
| manifest_branch="test-branch", |
| manifest_name="other.xml", |
| mirror=True, |
| reference="/repo/reference", |
| depth=99, |
| groups="abba,queen", |
| repo_url="https://repo.xyz/repo", |
| repo_branch="repo-branch", |
| ) |
| self.assertCommandCalled(expected_cmd, cwd=self.empty_root) |
| self.assertTrue(os.path.samefile(repo.root, self.empty_root)) |
| |
| def testInitializeExistingRepoDir(self) -> None: |
| """Test Repository.Initialize fails in existing repo dir.""" |
| with self.assertRaisesRegex(repo_util.Error, "cannot init in existing"): |
| repo_util.Repository.Initialize(self.repo_root, self.MANIFEST_URL) |
| |
| def testInitializeExistingRepoSubdir(self) -> None: |
| """Test Repository.Initialize fails in existing repo subdir.""" |
| with self.assertRaisesRegex(repo_util.Error, "cannot init in existing"): |
| repo_util.Repository.Initialize( |
| self.repo_root_subdir, self.MANIFEST_URL |
| ) |
| |
| def testInitializeFailCleanup(self) -> None: |
| """Test Repository.Initialize failure deletes .repo.""" |
| expected_cmd = ["repo", "init", "--manifest-url", self.MANIFEST_URL] |
| self.rc.AddCmdResult( |
| expected_cmd, returncode=99, side_effect=RepoInitSideEffects |
| ) |
| rmdir = self.PatchObject(osutils, "RmDir") |
| with self.assertRaises(cros_build_lib.RunCommandError): |
| repo_util.Repository.Initialize(self.empty_root, self.MANIFEST_URL) |
| repo_dir = os.path.join(self.empty_root, ".repo") |
| rmdir.assert_any_call(repo_dir, ignore_missing=True) |
| |
| def testFind(self) -> None: |
| """Test Repository.Find finds repo from subdir.""" |
| repo = repo_util.Repository.Find(self.repo_root_subdir) |
| self.assertEqual(repo.root, self.repo_root) |
| |
| def testFindNothing(self) -> None: |
| """Test Repository.Find finds nothing from non-repo dir.""" |
| self.assertIsNone(repo_util.Repository.Find(self.empty_root_subdir)) |
| |
| def testMustFind(self) -> None: |
| """Test Repository.MustFind finds repo from subdir.""" |
| repo = repo_util.Repository.MustFind(self.repo_root_subdir) |
| self.assertEqual(repo.root, self.repo_root) |
| |
| def testMustFindNothing(self) -> None: |
| """Test Repository.MustFind fails from non-repo dir.""" |
| with self.assertRaises(repo_util.NotInRepoError): |
| repo_util.Repository.MustFind(self.empty_root_subdir) |
| |
| |
| class RepositoryCommandMethodTest(cros_test_lib.RunCommandTempDirTestCase): |
| """Tests for repo_util.Repository command methods.""" |
| |
| # Testing _Run: pylint: disable=protected-access |
| |
| def setUp(self) -> None: |
| self.root = os.path.join(self.tempdir, "root") |
| self.repo_dir = os.path.join(self.root, ".repo") |
| self.subdir = os.path.join(self.root, "sub", "dir") |
| os.makedirs(self.repo_dir) |
| os.makedirs(self.subdir) |
| self.repo = repo_util.Repository(self.root) |
| |
| def AssertRepoCalled(self, repo_args, **kwargs) -> None: |
| kwargs.setdefault("cwd", self.root) |
| kwargs.setdefault("capture_output", False) |
| kwargs.setdefault("debug_level", logging.DEBUG) |
| kwargs.setdefault("encoding", "utf-8") |
| self.assertCommandCalled( |
| ["python3", RepoCmdPath(self.root)] + repo_args, **kwargs |
| ) |
| |
| def AddRepoResult(self, repo_args, **kwargs) -> None: |
| self.rc.AddCmdResult( |
| ["python3", RepoCmdPath(self.root)] + repo_args, **kwargs |
| ) |
| |
| def testAssertRepoCalled(self) -> None: |
| """Test that AddRepoResult test helper works.""" |
| self.repo._Run(["subcmd", "arg1"]) |
| with self.assertRaises(AssertionError): |
| self.AssertRepoCalled(["subcmd", "arg2"]) |
| with self.assertRaises(AssertionError): |
| self.AssertRepoCalled(["subcmd"]) |
| with self.assertRaises(AssertionError): |
| self.AssertRepoCalled(["subcmd", "arg1"], cwd="other_dir") |
| self.AssertRepoCalled(["subcmd", "arg1"]) |
| |
| def testRun(self) -> None: |
| """Test Repository._Run repo_cmd.""" |
| self.repo._Run(["subcmd", "arg"]) |
| self.AssertRepoCalled(["subcmd", "arg"]) |
| |
| def testRunSubDirCwd(self) -> None: |
| """Test Repository._Run cwd.""" |
| self.repo._Run(["subcmd"], cwd=self.subdir) |
| self.AssertRepoCalled(["subcmd"], cwd=self.subdir) |
| |
| def testRunBadCwd(self) -> None: |
| """Test Repository._Run fails on cwd outside of repo.""" |
| with self.assertRaises(repo_util.NotInRepoError): |
| self.repo._Run(["subcmd"], cwd=self.tempdir) |
| |
| def testSyncSimple(self) -> None: |
| """Test Repository.Sync simple call.""" |
| self.repo.Sync() |
| self.AssertRepoCalled(["sync"]) |
| |
| def testSyncComplex(self) -> None: |
| """Test Repository.Sync complex call.""" |
| manifest_path = os.path.join(self.tempdir, "other", "manifest.xml") |
| osutils.Touch(manifest_path, makedirs=True) |
| self.repo.Sync( |
| projects=["p1", "p2"], |
| local_only=True, |
| current_branch=True, |
| jobs=9, |
| manifest_path=manifest_path, |
| cwd=self.subdir, |
| ) |
| self.AssertRepoCalled( |
| [ |
| "sync", |
| "p1", |
| "p2", |
| "--local-only", |
| "--current-branch", |
| "--jobs", |
| "9", |
| "--manifest-name", |
| "../../../other/manifest.xml", |
| ], |
| cwd=self.subdir, |
| ) |
| |
| def testStartBranchSimple(self) -> None: |
| """Test Repository.StartBranch simple call.""" |
| self.repo.StartBranch("my-branch") |
| self.AssertRepoCalled(["start", "my-branch", "--all"]) |
| |
| def testStartBranchComplex(self) -> None: |
| """Test Repository.StartBranch complex call.""" |
| self.repo.StartBranch( |
| "my-branch", projects=["foo", "bar"], cwd=self.subdir |
| ) |
| self.AssertRepoCalled( |
| ["start", "my-branch", "foo", "bar"], cwd=self.subdir |
| ) |
| |
| def testListSimple(self) -> None: |
| """Test Repository.List simple call.""" |
| stdout = "src/project : my/project\nsrc/ugly : path : other/project\n" |
| self.AddRepoResult(["list"], stdout=stdout) |
| projects = self.repo.List() |
| self.AssertRepoCalled(["list"], capture_output=True) |
| self.assertListEqual( |
| projects, |
| [ |
| repo_util.ProjectInfo(name="my/project", path="src/project"), |
| repo_util.ProjectInfo( |
| name="other/project", path="src/ugly : path" |
| ), |
| ], |
| ) |
| |
| def testListComplex(self) -> None: |
| """Test Repository.List complex call.""" |
| self.repo.List(["project1", "project2"], cwd=self.subdir) |
| self.AssertRepoCalled( |
| ["list", "project1", "project2"], |
| cwd=self.subdir, |
| capture_output=True, |
| ) |
| |
| def testListProjectNotFound(self) -> None: |
| """Test Repository.List fails when given a nonexistant project.""" |
| self.AddRepoResult( |
| ["list", "foobar"], |
| returncode=1, |
| stderr="error: project foobar not found", |
| ) |
| with self.assertRaises(repo_util.ProjectNotFoundError): |
| self.repo.List(["foobar"]) |
| |
| def testManifest(self) -> None: |
| """Test Repository.Manifest.""" |
| stdout = "<manifest></manifest>" |
| self.AddRepoResult(["manifest"], stdout=stdout) |
| manifest = self.repo.Manifest() |
| self.assertIsNotNone(manifest) |
| |
| def testCopy(self) -> None: |
| """Test Repository.Copy.""" |
| copy_root = os.path.join(self.tempdir, "copy") |
| os.mkdir(copy_root) |
| |
| def mkdirDestRepo(*_args, **_kwargs) -> None: |
| os.mkdir(os.path.join(copy_root, ".repo")) |
| |
| stdout = "src/p1 : p1\nother : other/project\n" |
| self.AddRepoResult(["list"], stdout=stdout, side_effect=mkdirDestRepo) |
| copy = self.repo.Copy(copy_root) |
| self.assertEqual(copy.root, copy_root) |
| kwargs = dict( |
| debug_level=logging.DEBUG, |
| capture_output=True, |
| encoding="utf-8", |
| extra_env={"LC_MESSAGES": "C"}, |
| cwd=self.root, |
| ) |
| self.assertCommandCalled( |
| [ |
| "cp", |
| "--archive", |
| "--link", |
| "--parents", |
| ".repo/project-objects/p1.git/objects", |
| copy_root, |
| ], |
| **kwargs, |
| ) |
| self.assertCommandCalled( |
| [ |
| "cp", |
| "--archive", |
| "--link", |
| "--parents", |
| ".repo/project-objects/other/project.git/objects", |
| copy_root, |
| ], |
| **kwargs, |
| ) |
| self.assertCommandCalled( |
| [ |
| "cp", |
| "--archive", |
| "--no-clobber", |
| ".repo", |
| copy_root, |
| ], |
| **kwargs, |
| ) |
| |
| |
| @cros_test_lib.pytestmark_network_test |
| class RepositoryIntegrationTest(cros_test_lib.TempDirTestCase): |
| """Tests for repo_util.Repository that actually call `repo`. |
| |
| Note that these test methods are *not* independent: they must run in |
| definition order. |
| """ |
| |
| MANIFEST_URL = constants.EXTERNAL_GOB_URL + "/chromiumos/manifest.git" |
| PROJECT = "chromiumos/chromite" |
| PROJECT_DIR = "chromite" |
| |
| root = None |
| repo = None |
| copy = None |
| project_path = None |
| |
| tests = [] |
| |
| def runTest(self) -> None: |
| for test in self.tests: |
| test(self) |
| |
| @tests.append |
| def testInitialize(self) -> None: |
| """Test Repository.Initialize creates a .repo dir.""" |
| self.root = os.path.join(self.tempdir, "root") |
| os.mkdir(self.root) |
| # Try to use the current repo as a --reference. |
| reference = repo_util.Repository.Find(constants.SOURCE_ROOT) |
| self.repo = repo_util.Repository.Initialize( |
| self.root, |
| manifest_url=self.MANIFEST_URL, |
| reference=reference, |
| depth=1, |
| ) |
| self.assertExists(os.path.join(self.root, ".repo")) |
| |
| @tests.append |
| def testSync(self) -> None: |
| """Test Repository.Sync creates a project checkout dir.""" |
| self.repo.Sync([self.PROJECT], current_branch=True) |
| self.project_path = os.path.join(self.root, self.PROJECT_DIR) |
| self.assertExists(self.project_path) |
| |
| @tests.append |
| def testMustFind(self) -> None: |
| """Test Repository.MustFind finds the Repository from a subdir.""" |
| repo = repo_util.Repository.MustFind(self.project_path) |
| self.assertEqual(repo.root, self.root) |
| |
| @tests.append |
| def testList(self) -> None: |
| """Test Repository.List returns correct ProjectInfo.""" |
| projects = self.repo.List([self.PROJECT]) |
| self.assertEqual( |
| projects, |
| [repo_util.ProjectInfo(name=self.PROJECT, path=self.PROJECT_DIR)], |
| ) |
| |
| @tests.append |
| def testStartBranch(self) -> None: |
| """Test Repository.StartBranch creates a git branch.""" |
| self.repo.StartBranch("my-branch") |
| project_branch = git.GetCurrentBranch(self.project_path) |
| self.assertEqual(project_branch, "my-branch") |
| |
| @tests.append |
| def testManifest(self) -> None: |
| """Test Repository.Manifest includes project.""" |
| manifest = self.repo.Manifest() |
| project = manifest.GetUniqueProject(self.PROJECT) |
| self.assertEqual(project.path, self.PROJECT_DIR) |
| |
| @tests.append |
| def testCopy(self) -> None: |
| """Test Repository.Copy creates a working copy.""" |
| copy_root = os.path.join(self.tempdir, "copy") |
| os.mkdir(copy_root) |
| copy = self.repo.Copy(copy_root) |
| self.assertEqual(copy.root, copy_root) |
| copy.Sync([self.PROJECT], local_only=True) |
| self.assertExists(os.path.join(copy_root, self.PROJECT_DIR)) |
| # Check for hardlinking; any object file will do. |
| objects_path = repo_util.PROJECT_OBJECTS_PATH_FORMAT % self.PROJECT |
| for dirname, _, files in os.walk(os.path.join(self.root, objects_path)): |
| if files: |
| object_file = os.path.join(dirname, files[0]) |
| break |
| else: |
| self.fail("no object file found") |
| self.assertTrue( |
| os.path.samefile( |
| os.path.join(self.root, object_file), |
| os.path.join(copy_root, object_file), |
| ) |
| ) |
