# Copyright (C) 2022 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 subcmds/sync.py module.""" import json import os from pathlib import Path import shutil import tempfile import time import unittest from unittest import mock import pytest import command from error import GitError from error import RepoExitError import manifest_xml from project import SyncNetworkHalfResult from subcmds import sync @pytest.mark.parametrize( "cli_args, expected", [ ([], None), (["--recurse-submodules"], True), (["--no-recurse-submodules"], False), (["--fetch-submodules"], True), (["--no-fetch-submodules"], False), ], ) def test_recurse_submodules_option(cli_args, expected): """The submodule flags preserve an unset manifest-driven state.""" cmd = sync.Sync() opts, _ = cmd.OptionParser.parse_args(cli_args) assert opts.recurse_submodules is expected @pytest.mark.parametrize( "old_flag, new_flag", [ ("--fetch-submodules", "--recurse-submodules"), ("--no-fetch-submodules", "--no-recurse-submodules"), ], ) def test_recurse_submodules_option_deprecation(old_flag, new_flag): """The old submodule flags warn and direct users to their replacements.""" cmd = sync.Sync() with mock.patch.object(sync.logger, "warning") as warning: cmd.OptionParser.parse_args([old_flag]) warning.assert_called_once_with( "%s is deprecated; use %s instead", old_flag, new_flag ) assert old_flag not in cmd.OptionParser.format_help() @pytest.mark.parametrize( "use_superproject, cli_args, result", [ (True, ["--current-branch"], True), (True, ["--no-current-branch"], True), (True, [], True), (False, ["--current-branch"], True), (False, ["--no-current-branch"], False), (False, [], None), ], ) def test_get_current_branch_only(use_superproject, cli_args, result): """Test Sync._GetCurrentBranchOnly logic. Sync._GetCurrentBranchOnly should return True if a superproject is requested, and otherwise the value of the current_branch_only option. """ cmd = sync.Sync() opts, _ = cmd.OptionParser.parse_args(cli_args) with mock.patch( "git_superproject.UseSuperproject", return_value=use_superproject ): assert cmd._GetCurrentBranchOnly(opts, cmd.manifest) == result @pytest.mark.parametrize( "cli_args, expected_groups", [ ([], None), (["-g", "groupA"], "groupA"), (["--groups=groupB,groupC"], "groupB,groupC"), ], ) def test_groups_option_parsing(cli_args, expected_groups): """Test --groups / -g option parsing.""" cmd = sync.Sync() opts, _ = cmd.OptionParser.parse_args(cli_args) assert opts.groups == expected_groups def _create_manifest_with_groups(topdir: Path) -> manifest_xml.XmlManifest: """Create a test XmlManifest with projects assigned to various groups.""" repodir = topdir / ".repo" manifest_dir = repodir / "manifests" manifest_file = repodir / manifest_xml.MANIFEST_FILE_NAME repodir.mkdir(exist_ok=True) manifest_dir.mkdir(exist_ok=True) gitdir = repodir / "manifests.git" gitdir.mkdir(exist_ok=True) (gitdir / "config").write_text( """[remote "origin"] url = https://localhost:0/manifest """, encoding="utf-8", ) manifest_file.write_text( """ """, encoding="utf-8", ) for p in [ "proj_g1", "proj_g2", "proj_g1_g2", "proj_default", "proj_notdefault", ]: (repodir / "projects" / f"{p}.git").mkdir(parents=True, exist_ok=True) return manifest_xml.XmlManifest(str(repodir), str(manifest_file)) @pytest.mark.parametrize( "cli_args, expected_projects", [ (["-g", "group1"], ["proj_g1", "proj_g1_g2"]), (["-g", "group2"], ["proj_g2", "proj_g1_g2"]), (["-g", "group1,group2"], ["proj_g1", "proj_g1_g2", "proj_g2"]), (["-g", "default,-group1"], ["proj_default", "proj_g2"]), ([], ["proj_default", "proj_g1", "proj_g1_g2", "proj_g2"]), ], ) def test_sync_groups_manifest_filtering( tmp_path: Path, cli_args, expected_projects ): """Test that repo sync -g selects only matching projects.""" manifest = _create_manifest_with_groups(tmp_path) cmd = sync.Sync() cmd.manifest = manifest opts, args = cmd.OptionParser.parse_args(cli_args) projects = cmd.GetProjects(args, groups=opts.groups, missing_ok=True) project_names = sorted([p.name for p in projects]) assert project_names == sorted(expected_projects) def test_sync_update_projects_revision_id_respects_groups(tmp_path: Path): """Test that _UpdateProjectsRevisionId filters projects using opt.groups.""" manifest = _create_manifest_with_groups(tmp_path) cmd = sync.Sync() cmd.manifest = manifest superproject = mock.MagicMock() superproject.UpdateProjectsRevisionId.return_value = mock.MagicMock( manifest_path=None ) manifest._superproject = superproject opts, args = cmd.OptionParser.parse_args(["-g", "group1"]) opts.verbose = False opts.fetch_submodules = False opts.this_manifest_only = True opts.local_only = False with mock.patch.object( cmd, "GetProjects", wraps=cmd.GetProjects ) as spy_get_projects: with mock.patch.object(cmd, "ManifestList", return_value=[manifest]): cmd._UpdateProjectsRevisionId(opts, args, {}, manifest) spy_get_projects.assert_called_once() _, kwargs = spy_get_projects.call_args assert kwargs.get("groups") == "group1" # Used to patch os.cpu_count() for reliable results. OS_CPU_COUNT = 24 @pytest.mark.parametrize( "argv, jobs_manifest, jobs, jobs_net, jobs_check", [ # No user or manifest settings. ([], None, OS_CPU_COUNT, 1, command.DEFAULT_LOCAL_JOBS), # No user settings, so manifest settings control. ([], 3, 3, 3, 3), # User settings, but no manifest. (["--jobs=4"], None, 4, 4, 4), (["--jobs=4", "--jobs-network=5"], None, 4, 5, 4), (["--jobs=4", "--jobs-checkout=6"], None, 4, 4, 6), (["--jobs=4", "--jobs-network=5", "--jobs-checkout=6"], None, 4, 5, 6), ( ["--jobs-network=5"], None, OS_CPU_COUNT, 5, command.DEFAULT_LOCAL_JOBS, ), (["--jobs-checkout=6"], None, OS_CPU_COUNT, 1, 6), (["--jobs-network=5", "--jobs-checkout=6"], None, OS_CPU_COUNT, 5, 6), # User settings with manifest settings. (["--jobs=4"], 3, 4, 4, 4), (["--jobs=4", "--jobs-network=5"], 3, 4, 5, 4), (["--jobs=4", "--jobs-checkout=6"], 3, 4, 4, 6), (["--jobs=4", "--jobs-network=5", "--jobs-checkout=6"], 3, 4, 5, 6), (["--jobs-network=5"], 3, 3, 5, 3), (["--jobs-checkout=6"], 3, 3, 3, 6), (["--jobs-network=5", "--jobs-checkout=6"], 3, 3, 5, 6), # Settings that exceed rlimits get capped. (["--jobs=1000000"], None, 83, 83, 83), ([], 1000000, 83, 83, 83), ], ) def test_cli_jobs(argv, jobs_manifest, jobs, jobs_net, jobs_check): """Tests --jobs option behavior.""" mp = mock.MagicMock() mp.manifest.default.sync_j = jobs_manifest mp.manifest.default.sync_j_max = None cmd = sync.Sync() opts, args = cmd.OptionParser.parse_args(argv) cmd.ValidateOptions(opts, args) with mock.patch.object(sync, "_rlimit_nofile", return_value=(256, 256)): with mock.patch.object(os, "cpu_count", return_value=OS_CPU_COUNT): cmd._ValidateOptionsWithManifest(opts, mp) assert opts.jobs == jobs assert opts.jobs_network == jobs_net assert opts.jobs_checkout == jobs_check @pytest.mark.parametrize( "argv, jobs_manifest, jobs_manifest_max, jobs, jobs_net, jobs_check", [ (["--jobs=10"], None, 5, 5, 5, 5), (["--jobs=10", "--jobs-network=10"], None, 5, 5, 5, 5), (["--jobs=10", "--jobs-checkout=10"], None, 5, 5, 5, 5), ], ) def test_cli_jobs_sync_j_max( argv, jobs_manifest, jobs_manifest_max, jobs, jobs_net, jobs_check ): """Tests --jobs option behavior with sync-j-max.""" mp = mock.MagicMock() mp.manifest.default.sync_j = jobs_manifest mp.manifest.default.sync_j_max = jobs_manifest_max cmd = sync.Sync() opts, args = cmd.OptionParser.parse_args(argv) cmd.ValidateOptions(opts, args) with mock.patch.object(sync, "_rlimit_nofile", return_value=(256, 256)): with mock.patch.object(os, "cpu_count", return_value=OS_CPU_COUNT): cmd._ValidateOptionsWithManifest(opts, mp) assert opts.jobs == jobs assert opts.jobs_network == jobs_net assert opts.jobs_checkout == jobs_check class LocalSyncState(unittest.TestCase): """Tests for LocalSyncState.""" _TIME = 10 def setUp(self): """Common setup.""" self.topdir = tempfile.mkdtemp("LocalSyncState") self.repodir = os.path.join(self.topdir, ".repo") os.makedirs(self.repodir) self.manifest = mock.MagicMock( topdir=self.topdir, repodir=self.repodir, repoProject=mock.MagicMock(relpath=".repo/repo"), ) self.state = self._new_state() def tearDown(self): """Common teardown.""" shutil.rmtree(self.topdir) def _new_state(self, time=_TIME): with mock.patch("time.time", return_value=time): return sync.LocalSyncState(self.manifest) def test_set(self): """Times are set.""" p = mock.MagicMock(relpath="projA") self.state.SetFetchTime(p) self.state.SetCheckoutTime(p) self.assertEqual(self.state.GetFetchTime(p), self._TIME) self.assertEqual(self.state.GetCheckoutTime(p), self._TIME) def test_update(self): """Times are updated.""" with open(self.state._path, "w") as f: f.write( """ { "projB": { "last_fetch": 5, "last_checkout": 7 } } """ ) # Initialize state to read from the new file. self.state = self._new_state() projA = mock.MagicMock(relpath="projA") projB = mock.MagicMock(relpath="projB") self.assertEqual(self.state.GetFetchTime(projA), None) self.assertEqual(self.state.GetFetchTime(projB), 5) self.assertEqual(self.state.GetCheckoutTime(projB), 7) self.state.SetFetchTime(projA) self.state.SetFetchTime(projB) self.assertEqual(self.state.GetFetchTime(projA), self._TIME) self.assertEqual(self.state.GetFetchTime(projB), self._TIME) self.assertEqual(self.state.GetCheckoutTime(projB), 7) def test_save_to_file(self): """Data is saved under repodir.""" p = mock.MagicMock(relpath="projA") self.state.SetFetchTime(p) self.state.Save() self.assertEqual( os.listdir(self.repodir), [".repo_localsyncstate.json"] ) def test_partial_sync(self): """Partial sync state is detected.""" with open(self.state._path, "w") as f: f.write( """ { "projA": { "last_fetch": 5, "last_checkout": 5 }, "projB": { "last_fetch": 5, "last_checkout": 5 } } """ ) # Initialize state to read from the new file. self.state = self._new_state() projB = mock.MagicMock(relpath="projB") self.assertEqual(self.state.IsPartiallySynced(), False) self.state.SetFetchTime(projB) self.state.SetCheckoutTime(projB) self.assertEqual(self.state.IsPartiallySynced(), True) def test_ignore_repo_project(self): """Sync data for repo project is ignored when checking partial sync.""" p = mock.MagicMock(relpath="projA") self.state.SetFetchTime(p) self.state.SetCheckoutTime(p) self.state.SetFetchTime(self.manifest.repoProject) self.state.Save() self.assertEqual(self.state.IsPartiallySynced(), False) self.state = self._new_state(self._TIME + 1) self.state.SetFetchTime(self.manifest.repoProject) self.assertEqual( self.state.GetFetchTime(self.manifest.repoProject), self._TIME + 1 ) self.assertEqual(self.state.GetFetchTime(p), self._TIME) self.assertEqual(self.state.IsPartiallySynced(), False) def test_nonexistent_project(self): """Unsaved projects don't have data.""" p = mock.MagicMock(relpath="projC") self.assertEqual(self.state.GetFetchTime(p), None) self.assertEqual(self.state.GetCheckoutTime(p), None) def test_prune_removed_projects(self): """Removed projects are pruned.""" with open(self.state._path, "w") as f: f.write( """ { "projA": { "last_fetch": 5 }, "projB": { "last_fetch": 7 } } """ ) def mock_exists(path): if "projA" in path: return False return True projA = mock.MagicMock(relpath="projA") projB = mock.MagicMock(relpath="projB") self.state = self._new_state() self.assertEqual(self.state.GetFetchTime(projA), 5) self.assertEqual(self.state.GetFetchTime(projB), 7) with mock.patch("os.path.exists", side_effect=mock_exists): self.state.PruneRemovedProjects() self.assertIsNone(self.state.GetFetchTime(projA)) self.state = self._new_state() self.assertIsNone(self.state.GetFetchTime(projA)) self.assertEqual(self.state.GetFetchTime(projB), 7) def test_prune_removed_and_symlinked_projects(self): """Removed projects that still exists on disk as symlink are pruned.""" with open(self.state._path, "w") as f: f.write( """ { "projA": { "last_fetch": 5 }, "projB": { "last_fetch": 7 } } """ ) def mock_exists(path): return True def mock_islink(path): if "projB" in path: return True return False projA = mock.MagicMock(relpath="projA") projB = mock.MagicMock(relpath="projB") self.state = self._new_state() self.assertEqual(self.state.GetFetchTime(projA), 5) self.assertEqual(self.state.GetFetchTime(projB), 7) with mock.patch("os.path.exists", side_effect=mock_exists): with mock.patch("os.path.islink", side_effect=mock_islink): self.state.PruneRemovedProjects() self.assertIsNone(self.state.GetFetchTime(projB)) self.state = self._new_state() self.assertIsNone(self.state.GetFetchTime(projB)) self.assertEqual(self.state.GetFetchTime(projA), 5) class FakeProject: def __init__(self, relpath, name=None, objdir=None): self.relpath = relpath self.name = name or relpath self.objdir = objdir or relpath self.worktree = relpath self.parent = None self.use_git_worktrees = False self.UseAlternates = False self.manifest = mock.MagicMock() self.manifest.GetProjectsWithName.return_value = [self] self.config = mock.MagicMock() self.EnableRepositoryExtension = mock.MagicMock() def RelPath(self, local=None): return self.relpath def __str__(self): return f"project: {self.relpath}" def __repr__(self): return str(self) class SafeCheckoutOrder(unittest.TestCase): def test_no_nested(self): p_f = FakeProject("f") p_foo = FakeProject("foo") out = sync._SafeCheckoutOrder([p_f, p_foo]) self.assertEqual(out, [[p_f, p_foo]]) def test_basic_nested(self): p_foo = p_foo = FakeProject("foo") p_foo_bar = FakeProject("foo/bar") out = sync._SafeCheckoutOrder([p_foo, p_foo_bar]) self.assertEqual(out, [[p_foo], [p_foo_bar]]) def test_complex_nested(self): p_foo = FakeProject("foo") p_foobar = FakeProject("foobar") p_foo_dash_bar = FakeProject("foo-bar") p_foo_bar = FakeProject("foo/bar") p_foo_bar_baz_baq = FakeProject("foo/bar/baz/baq") p_bar = FakeProject("bar") out = sync._SafeCheckoutOrder( [ p_foo_bar_baz_baq, p_foo, p_foobar, p_foo_dash_bar, p_foo_bar, p_bar, ] ) self.assertEqual( out, [ [p_bar, p_foo, p_foo_dash_bar, p_foobar], [p_foo_bar], [p_foo_bar_baz_baq], ], ) def test_sibling_submodules_with_shared_parent_are_serialized(self): parent = mock.Mock(worktree="/worktree/parent") other_parent = mock.Mock(worktree="/worktree/other") p_parent = FakeProject("parent") p_other = FakeProject("other") p_parent_sub1 = FakeProject("parent/sub1") p_parent_sub1.parent = parent p_parent_sub2 = FakeProject("parent/sub2") p_parent_sub2.parent = parent p_other_sub = FakeProject("other/sub") p_other_sub.parent = other_parent out = sync._SafeCheckoutOrder( [p_parent_sub2, p_other_sub, p_parent, p_parent_sub1, p_other] ) self.assertEqual( out, [ [p_other, p_parent], [p_other_sub, p_parent_sub1], [p_parent_sub2], ], ) def test_nested_submodules_respect_delayed_parent_level(self): parent = mock.Mock(worktree="/worktree/parent") sub1 = mock.Mock(worktree="/worktree/parent/sub1") sub2 = mock.Mock(worktree="/worktree/parent/sub2") p_parent = FakeProject("parent") p_parent_sub1 = FakeProject("parent/sub1") p_parent_sub1.parent = parent p_parent_sub1_nested = FakeProject("parent/sub1/nested") p_parent_sub1_nested.parent = sub1 p_parent_sub2 = FakeProject("parent/sub2") p_parent_sub2.parent = parent p_parent_sub2_nested = FakeProject("parent/sub2/nested") p_parent_sub2_nested.parent = sub2 out = sync._SafeCheckoutOrder( [ p_parent_sub2_nested, p_parent_sub2, p_parent_sub1_nested, p_parent, p_parent_sub1, ] ) self.assertEqual( out, [ [p_parent], [p_parent_sub1], [p_parent_sub1_nested, p_parent_sub2], [p_parent_sub2_nested], ], ) class Chunksize(unittest.TestCase): """Tests for _chunksize.""" def test_single_project(self): """Single project.""" self.assertEqual(sync._chunksize(1, 1), 1) def test_low_project_count(self): """Multiple projects, low number of projects to sync.""" self.assertEqual(sync._chunksize(10, 1), 10) self.assertEqual(sync._chunksize(10, 2), 5) self.assertEqual(sync._chunksize(10, 4), 2) self.assertEqual(sync._chunksize(10, 8), 1) self.assertEqual(sync._chunksize(10, 16), 1) def test_high_project_count(self): """Multiple projects, high number of projects to sync.""" self.assertEqual(sync._chunksize(2800, 1), 32) self.assertEqual(sync._chunksize(2800, 16), 32) self.assertEqual(sync._chunksize(2800, 32), 32) self.assertEqual(sync._chunksize(2800, 64), 32) self.assertEqual(sync._chunksize(2800, 128), 21) class GetPreciousObjectsState(unittest.TestCase): """Tests for _GetPreciousObjectsState.""" def setUp(self): """Common setup.""" self.cmd = sync.Sync() self.project = p = mock.MagicMock( use_git_worktrees=False, UseAlternates=False ) p.manifest.GetProjectsWithName.return_value = [p] self.opt = mock.Mock(spec_set=["this_manifest_only"]) self.opt.this_manifest_only = False def test_worktrees(self): """False for worktrees.""" self.project.use_git_worktrees = True self.assertFalse( self.cmd._GetPreciousObjectsState(self.project, self.opt) ) def test_not_shared(self): """Singleton project.""" self.assertFalse( self.cmd._GetPreciousObjectsState(self.project, self.opt) ) def test_shared(self): """Shared project.""" self.project.manifest.GetProjectsWithName.return_value = [ self.project, self.project, ] self.assertTrue( self.cmd._GetPreciousObjectsState(self.project, self.opt) ) def test_shared_with_alternates(self): """Shared project, with alternates.""" self.project.manifest.GetProjectsWithName.return_value = [ self.project, self.project, ] self.project.UseAlternates = True self.assertFalse( self.cmd._GetPreciousObjectsState(self.project, self.opt) ) def test_not_found(self): """Project not found in manifest.""" self.project.manifest.GetProjectsWithName.return_value = [] self.assertFalse( self.cmd._GetPreciousObjectsState(self.project, self.opt) ) class KeyboardInterruptTest(unittest.TestCase): """Tests for KeyboardInterrupt handling in Sync operations.""" def setUp(self): self.project = mock.MagicMock(name="project") self.project.name = "project" self.project.relpath = "proj" self.project.manifest.IsArchive = False self.opt = mock.Mock() self.opt.quiet = True self.opt.verbose = False self.opt.tags = False self.sync_dict = {} self.get_parallel_context_mock = { "projects": [self.project], "sync_dict": self.sync_dict, "ssh_proxy": None, } @mock.patch("subcmds.sync.Sync.is_multiprocessing_active") def test_fetch_one_keyboard_interrupt_main_process(self, mock_is_active): """Test that _FetchOne re-raises KeyboardInterrupt if not worker.""" mock_is_active.return_value = False self.project.Sync_NetworkHalf.side_effect = KeyboardInterrupt() with mock.patch.object( sync.Sync, "get_parallel_context", return_value=self.get_parallel_context_mock, ): with self.assertRaises(KeyboardInterrupt): sync.Sync._FetchOne(self.opt, 0) @mock.patch("subcmds.sync.Sync.is_multiprocessing_active") def test_fetch_one_keyboard_interrupt_worker_process(self, mock_is_active): """Test that _FetchOne suppresses KeyboardInterrupt in workers.""" mock_is_active.return_value = True self.project.Sync_NetworkHalf.side_effect = KeyboardInterrupt() with mock.patch.object( sync.Sync, "get_parallel_context", return_value=self.get_parallel_context_mock, ): result = sync.Sync._FetchOne(self.opt, 0) self.assertFalse(result.success) class CheckForBloatedProjects(unittest.TestCase): """Tests for Sync._CheckForBloatedProjects.""" def setUp(self): self.cmd = sync.Sync() self.opt = mock.Mock() self.opt.quiet = True self.opt.jobs = 1 self.project = mock.MagicMock(clone_depth="1") self.project.name = "project" self.project.Exists = True self.project.worktree = "worktree" self.project.stateless_prune_needed = False self.cmd.git_event_log = mock.MagicMock() self.cmd._bloated_projects = [] @mock.patch("subcmds.sync.git_require") def test_git_version_unsupported(self, mock_git_require): """Test that it returns early if git version is unsupported.""" mock_git_require.return_value = False self.cmd._CheckForBloatedProjects([self.project], self.opt) self.assertFalse(self.cmd.git_event_log.ErrorEvent.called) @mock.patch("subcmds.sync.git_require") def test_no_projects(self, mock_git_require): """Test that it returns early if no projects have clone_depth.""" mock_git_require.return_value = True self.project.clone_depth = None self.cmd._CheckForBloatedProjects([self.project], self.opt) self.assertFalse(self.cmd.git_event_log.ErrorEvent.called) @mock.patch("subcmds.sync.git_require") @mock.patch("subcmds.sync.Progress") def test_bloated_project_found(self, mock_progress, mock_git_require): """Test that it adds project to _bloated_projects.""" mock_git_require.return_value = True self.cmd.get_parallel_context = mock.Mock( return_value={"projects": [self.project]} ) def mock_execute_in_parallel( jobs, func, work_items, callback, **kwargs ): callback(None, mock.Mock(), ["project"]) return True self.cmd.ExecuteInParallel = mock_execute_in_parallel with mock.patch.object(self.cmd, "ParallelContext"): self.cmd._CheckForBloatedProjects([self.project], self.opt) self.assertEqual(self.cmd._bloated_projects, ["project"]) @mock.patch("subcmds.sync.git_require") @mock.patch("subcmds.sync.Progress") def test_stateless_prune_excluded(self, mock_progress, mock_git_require): """Test that projects pruned for stateless sync are excluded.""" mock_git_require.return_value = True self.project.stateless_prune_needed = True self.cmd.ExecuteInParallel = mock.Mock() with mock.patch.object(self.cmd, "ParallelContext"): self.cmd._CheckForBloatedProjects([self.project], self.opt) self.assertFalse(self.cmd.ExecuteInParallel.called) self.assertEqual(self.cmd._bloated_projects, []) class GCProjectsTest(unittest.TestCase): """Tests for Sync._GCProjects.""" def setUp(self): self.cmd = sync.Sync() self.opt = mock.Mock() self.opt.quiet = True self.opt.auto_gc = True self.opt.jobs = 1 self.project = mock.MagicMock() self.project.name = "project" self.project.objdir = "objdir" self.project.gitdir = "gitdir" self.project.bare_git = mock.MagicMock() self.project.bare_git._project = self.project self.cmd.git_event_log = mock.MagicMock() @mock.patch("subcmds.sync.Progress") def test_GCProjects_skip_gc(self, mock_progress): """Test that it skips GC if opt.auto_gc is False.""" self.opt.auto_gc = False with mock.patch.object( sync.Sync, "_SetPreciousObjectsState" ) as mock_set_state: self.cmd._GCProjects([self.project], self.opt, None) mock_set_state.assert_called_once_with(self.project, self.opt) self.assertFalse(self.project.bare_git.gc.called) @mock.patch("subcmds.sync.Progress") def test_GCProjects_sequential(self, mock_progress): """Test sequential GC (jobs < 2).""" with mock.patch.object(sync.Sync, "_SetPreciousObjectsState"): self.cmd._GCProjects([self.project], self.opt, None) self.project.bare_git.gc.assert_called_once_with( "--auto", config={"gc.autoDetach": "false"} ) # Verify that gc.autoDetach was not permanently set in config. for call in self.project.config.SetString.call_args_list: self.assertNotEqual(call.args[0], "gc.autoDetach") @mock.patch("subcmds.sync.Progress") def test_GCProjects_parallel(self, mock_progress): """Test parallel GC (jobs >= 2).""" self.opt.jobs = 2 with mock.patch.object(sync.Sync, "_SetPreciousObjectsState"): with mock.patch("subcmds.sync._threading.Thread") as mock_thread: mock_t = mock.MagicMock() mock_thread.return_value = mock_t err_event = mock.Mock() err_event.is_set.return_value = False self.cmd._GCProjects([self.project], self.opt, err_event) self.assertTrue(mock_thread.called) class SyncCommand(unittest.TestCase): """Tests for cmd.Execute.""" def setUp(self): """Common setup.""" self.repodir = tempfile.mkdtemp(".repo") self.manifest = manifest = mock.MagicMock( repodir=self.repodir, ) git_event_log = mock.MagicMock(ErrorEvent=mock.Mock(return_value=None)) self.outer_client = outer_client = mock.MagicMock() outer_client.manifest.IsArchive = True manifest.manifestProject.worktree = "worktree_path/" manifest.repoProject.LastFetch = time.time() self.sync_network_half_error = None self.sync_local_half_error = None self.cmd = sync.Sync( manifest=manifest, outer_client=outer_client, git_event_log=git_event_log, ) def Sync_NetworkHalf(*args, **kwargs): return SyncNetworkHalfResult(True, self.sync_network_half_error) def Sync_LocalHalf(*args, **kwargs): if self.sync_local_half_error: raise self.sync_local_half_error self.project = p = mock.MagicMock( use_git_worktrees=False, UseAlternates=False, name="project", Sync_NetworkHalf=Sync_NetworkHalf, Sync_LocalHalf=Sync_LocalHalf, RelPath=mock.Mock(return_value="rel_path"), ) p.manifest.GetProjectsWithName.return_value = [p] mock.patch.object( sync, "_PostRepoFetch", return_value=None, ).start() mock.patch.object( self.cmd, "GetProjects", return_value=[self.project] ).start() opt, _ = self.cmd.OptionParser.parse_args([]) opt.clone_bundle = False opt.jobs = 4 opt.quiet = True opt.use_superproject = False opt.current_branch_only = True opt.optimized_fetch = True opt.retry_fetches = 1 opt.prune = False opt.auto_gc = False opt.repo_verify = False self.opt = opt def tearDown(self): mock.patch.stopall() def test_command_exit_error(self): """Ensure unsuccessful commands raise expected errors.""" self.sync_network_half_error = GitError( "sync_network_half_error error", project=self.project ) self.sync_local_half_error = GitError( "sync_local_half_error", project=self.project ) with self.assertRaises(RepoExitError) as e: self.cmd.Execute(self.opt, []) self.assertIn(self.sync_local_half_error, e.aggregate_errors) self.assertIn(self.sync_network_half_error, e.aggregate_errors) def test_groups_passed_to_get_projects(self): """Ensure Execute passes opt.groups to GetProjects.""" self.opt.groups = "my_group" self.opt.mp_update = False with mock.patch.object(self.cmd, "_UpdateRepoProject"): with mock.patch.object(self.cmd, "_ValidateOptionsWithManifest"): with mock.patch.object(self.cmd, "_SyncInterleaved"): with mock.patch.object(self.cmd, "_RunPostSyncHook"): self.cmd.Execute(self.opt, []) self.cmd.GetProjects.assert_called() _, kwargs = self.cmd.GetProjects.call_args self.assertEqual(kwargs.get("groups"), "my_group") class SyncUpdateRepoProject(unittest.TestCase): """Tests for Sync._UpdateRepoProject.""" def setUp(self): """Common setup.""" self.repodir = tempfile.mkdtemp(".repo") self.manifest = manifest = mock.MagicMock(repodir=self.repodir) # Create a repoProject with a mock Sync_NetworkHalf. repoProject = mock.MagicMock(name="repo") repoProject.Sync_NetworkHalf = mock.Mock( return_value=SyncNetworkHalfResult(True, None) ) manifest.repoProject = repoProject manifest.IsArchive = False manifest.CloneFilter = None manifest.PartialCloneExclude = None manifest.CloneFilterForDepth = None git_event_log = mock.MagicMock(ErrorEvent=mock.Mock(return_value=None)) self.cmd = sync.Sync(manifest=manifest, git_event_log=git_event_log) opt, _ = self.cmd.OptionParser.parse_args([]) opt.local_only = False opt.repo_verify = False opt.verbose = False opt.quiet = True opt.force_sync = False opt.clone_bundle = False opt.tags = False opt.optimized_fetch = False opt.retry_fetches = 0 opt.prune = False self.opt = opt self.errors = [] mock.patch.object(sync.Sync, "_GetCurrentBranchOnly").start() def tearDown(self): shutil.rmtree(self.repodir) mock.patch.stopall() def test_fetches_when_stale(self): """Test it fetches when the repo project is stale.""" self.manifest.repoProject.LastFetch = time.time() - ( sync._ONE_DAY_S + 1 ) with mock.patch.object(sync, "_PostRepoFetch") as mock_post_fetch: self.cmd._UpdateRepoProject(self.opt, self.manifest, self.errors) self.manifest.repoProject.Sync_NetworkHalf.assert_called_once() mock_post_fetch.assert_called_once() self.assertEqual(self.errors, []) def test_skips_when_fresh(self): """Test it skips fetch when repo project is fresh.""" self.manifest.repoProject.LastFetch = time.time() with mock.patch.object(sync, "_PostRepoFetch") as mock_post_fetch: self.cmd._UpdateRepoProject(self.opt, self.manifest, self.errors) self.manifest.repoProject.Sync_NetworkHalf.assert_not_called() mock_post_fetch.assert_not_called() def test_skips_local_only(self): """Test it does nothing with --local-only.""" self.opt.local_only = True self.manifest.repoProject.LastFetch = time.time() - ( sync._ONE_DAY_S + 1 ) with mock.patch.object(sync, "_PostRepoFetch") as mock_post_fetch: self.cmd._UpdateRepoProject(self.opt, self.manifest, self.errors) self.manifest.repoProject.Sync_NetworkHalf.assert_not_called() mock_post_fetch.assert_not_called() def test_post_repo_fetch_skipped_on_env_var(self): """Test _PostRepoFetch is skipped when REPO_SKIP_SELF_UPDATE is set.""" self.manifest.repoProject.LastFetch = time.time() with mock.patch.dict(os.environ, {"REPO_SKIP_SELF_UPDATE": "1"}): with mock.patch.object(sync, "_PostRepoFetch") as mock_post_fetch: self.cmd._UpdateRepoProject( self.opt, self.manifest, self.errors ) mock_post_fetch.assert_not_called() def test_fetch_failure_is_handled(self): """Test that a fetch failure is recorded and doesn't crash.""" self.manifest.repoProject.LastFetch = time.time() - ( sync._ONE_DAY_S + 1 ) fetch_error = GitError("Fetch failed") self.manifest.repoProject.Sync_NetworkHalf.return_value = ( SyncNetworkHalfResult(False, fetch_error) ) with mock.patch.object(sync, "_PostRepoFetch") as mock_post_fetch: self.cmd._UpdateRepoProject(self.opt, self.manifest, self.errors) self.manifest.repoProject.Sync_NetworkHalf.assert_called_once() mock_post_fetch.assert_not_called() self.assertEqual(self.errors, [fetch_error]) class InterleavedSyncTest(unittest.TestCase): """Tests for interleaved sync.""" def setUp(self): """Set up a sync command with mocks.""" self.repodir = tempfile.mkdtemp(".repo") self.manifest = mock.MagicMock(repodir=self.repodir) self.manifest.repoProject.LastFetch = time.time() self.manifest.repoProject.worktree = self.repodir self.manifest.manifestProject.worktree = self.repodir self.manifest.IsArchive = False self.manifest.CloneBundle = False self.manifest.default.sync_j = 1 self.outer_client = mock.MagicMock() self.outer_client.manifest.IsArchive = False self.cmd = sync.Sync( manifest=self.manifest, outer_client=self.outer_client ) self.cmd.outer_manifest = self.manifest # Mock projects. self.projA = FakeProject("projA", objdir="objA") self.projB = FakeProject("projB", objdir="objB") self.projA_sub = FakeProject( "projA/sub", name="projA_sub", objdir="objA_sub" ) self.projC = FakeProject("projC", objdir="objC") # Mock methods that are not part of the core interleaved sync logic. mock.patch.object(self.cmd, "_UpdateAllManifestProjects").start() mock.patch.object(self.cmd, "_UpdateProjectsRevisionId").start() mock.patch.object(self.cmd, "_ValidateOptionsWithManifest").start() mock.patch.object(sync, "_PostRepoUpgrade").start() mock.patch.object(sync, "_PostRepoFetch").start() # Mock parallel context for worker tests. self.parallel_context_patcher = mock.patch( "subcmds.sync.Sync.get_parallel_context" ) self.mock_get_parallel_context = self.parallel_context_patcher.start() self.sync_dict = {} self.mock_context = { "projects": [], "sync_dict": self.sync_dict, } self.mock_get_parallel_context.return_value = self.mock_context # Mock _GetCurrentBranchOnly for worker tests. mock.patch.object(sync.Sync, "_GetCurrentBranchOnly").start() self.cmd._fetch_times = mock.Mock() self.cmd._local_sync_state = mock.Mock() def tearDown(self): """Clean up resources.""" shutil.rmtree(self.repodir) mock.patch.stopall() def test_interleaved_fail_fast(self): """Test that --fail-fast is respected in interleaved mode.""" opt, args = self.cmd.OptionParser.parse_args( ["--interleaved", "--fail-fast", "-j2"] ) opt.quiet = True # With projA/sub, _SafeCheckoutOrder creates two batches: # 1. [projA, projB] # 2. [projA/sub] # We want to fail on the first batch and ensure the second isn't run. all_projects = [self.projA, self.projB, self.projA_sub] mock.patch.object( self.cmd, "GetProjects", return_value=all_projects ).start() # Mock ExecuteInParallel to simulate a failed run on the first batch of # projects. execute_mock = mock.patch.object( self.cmd, "ExecuteInParallel", return_value=False ).start() with self.assertRaises(sync.SyncFailFastError): self.cmd._SyncInterleaved( opt, args, [], self.manifest, self.manifest.manifestProject, all_projects, {}, ) execute_mock.assert_called_once() def test_interleaved_shared_objdir_serial(self): """Test that projects with shared objdir are processed serially.""" opt, args = self.cmd.OptionParser.parse_args(["--interleaved", "-j4"]) opt.quiet = True # Setup projects with a shared objdir. self.projA.objdir = "common_objdir" self.projC.objdir = "common_objdir" all_projects = [self.projA, self.projB, self.projC] mock.patch.object( self.cmd, "GetProjects", return_value=all_projects ).start() def execute_side_effect(jobs, target, work_items, **kwargs): # The callback is a partial object. The first arg is the set we # need to update to avoid the stall detection. synced_relpaths_set = kwargs["callback"].args[0] projects_in_pass = self.cmd.get_parallel_context()["projects"] for item in work_items: for project_idx in item: synced_relpaths_set.add( projects_in_pass[project_idx].relpath ) return True execute_mock = mock.patch.object( self.cmd, "ExecuteInParallel", side_effect=execute_side_effect ).start() self.cmd._SyncInterleaved( opt, args, [], self.manifest, self.manifest.manifestProject, all_projects, {}, ) execute_mock.assert_called_once() jobs_arg, _, work_items = execute_mock.call_args.args self.assertEqual(jobs_arg, 2) work_items_sets = {frozenset(item) for item in work_items} expected_sets = {frozenset([0, 2]), frozenset([1])} self.assertEqual(work_items_sets, expected_sets) def _get_opts(self, args=None): """Helper to get default options for worker tests.""" if args is None: args = ["--interleaved"] opt, _ = self.cmd.OptionParser.parse_args(args) # Set defaults for options used by the worker. opt.quiet = True opt.verbose = False opt.force_sync = False opt.clone_bundle = False opt.tags = False opt.optimized_fetch = False opt.retry_fetches = 0 opt.prune = False opt.detach_head = False opt.force_checkout = False opt.rebase = False return opt def test_worker_successful_sync(self): """Test _SyncProjectList with a successful fetch and checkout.""" opt = self._get_opts() project = self.projA project.Sync_NetworkHalf = mock.Mock( return_value=SyncNetworkHalfResult(error=None, remote_fetched=True) ) project.Sync_LocalHalf = mock.Mock() project.manifest.manifestProject.config = mock.MagicMock() self.mock_context["projects"] = [project] with mock.patch("subcmds.sync.SyncBuffer") as mock_sync_buffer: mock_sync_buf_instance = mock.MagicMock() mock_sync_buf_instance.Finish.return_value = True mock_sync_buf_instance.errors = [] mock_sync_buffer.return_value = mock_sync_buf_instance result_obj = self.cmd._SyncProjectList(opt, [0]) self.assertEqual(len(result_obj.results), 1) result = result_obj.results[0] self.assertTrue(result.fetch_success) self.assertTrue(result.checkout_success) self.assertEqual(result.fetch_errors, []) self.assertEqual(result.checkout_errors, []) project.Sync_NetworkHalf.assert_called_once() project.Sync_LocalHalf.assert_called_once() def test_worker_fetch_fails(self): """Test _SyncProjectList with a failed fetch.""" opt = self._get_opts() project = self.projA fetch_error = GitError("Fetch failed") project.Sync_NetworkHalf = mock.Mock( return_value=SyncNetworkHalfResult( error=fetch_error, remote_fetched=False ) ) project.Sync_LocalHalf = mock.Mock() self.mock_context["projects"] = [project] result_obj = self.cmd._SyncProjectList(opt, [0]) result = result_obj.results[0] self.assertFalse(result.fetch_success) self.assertFalse(result.checkout_success) self.assertEqual(result.fetch_errors, [fetch_error]) self.assertEqual(result.checkout_errors, []) project.Sync_NetworkHalf.assert_called_once() project.Sync_LocalHalf.assert_not_called() def test_worker_no_worktree(self): """Test interleaved sync does not checkout with no worktree.""" opt = self._get_opts() project = self.projA project.worktree = None project.Sync_NetworkHalf = mock.Mock( return_value=SyncNetworkHalfResult(error=None, remote_fetched=True) ) project.Sync_LocalHalf = mock.Mock() self.mock_context["projects"] = [project] result_obj = self.cmd._SyncProjectList(opt, [0]) result = result_obj.results[0] self.assertTrue(result.fetch_success) self.assertTrue(result.checkout_success) project.Sync_NetworkHalf.assert_called_once() project.Sync_LocalHalf.assert_not_called() def test_worker_fetch_fails_exception(self): """Test _SyncProjectList with an exception during fetch.""" opt = self._get_opts() project = self.projA fetch_error = GitError("Fetch failed") project.Sync_NetworkHalf = mock.Mock(side_effect=fetch_error) project.Sync_LocalHalf = mock.Mock() self.mock_context["projects"] = [project] result_obj = self.cmd._SyncProjectList(opt, [0]) result = result_obj.results[0] self.assertFalse(result.fetch_success) self.assertFalse(result.checkout_success) self.assertEqual(result.fetch_errors, [fetch_error]) project.Sync_NetworkHalf.assert_called_once() project.Sync_LocalHalf.assert_not_called() def test_worker_checkout_fails(self): """Test _SyncProjectList with an exception during checkout.""" opt = self._get_opts() project = self.projA project.Sync_NetworkHalf = mock.Mock( return_value=SyncNetworkHalfResult(error=None, remote_fetched=True) ) checkout_error = GitError("Checkout failed") project.Sync_LocalHalf = mock.Mock(side_effect=checkout_error) project.manifest.manifestProject.config = mock.MagicMock() self.mock_context["projects"] = [project] with mock.patch("subcmds.sync.SyncBuffer"): result_obj = self.cmd._SyncProjectList(opt, [0]) result = result_obj.results[0] self.assertTrue(result.fetch_success) self.assertFalse(result.checkout_success) self.assertEqual(result.fetch_errors, []) self.assertEqual(result.checkout_errors, [checkout_error]) project.Sync_NetworkHalf.assert_called_once() project.Sync_LocalHalf.assert_called_once() def test_worker_local_only(self): """Test _SyncProjectList with --local-only.""" opt = self._get_opts(["--interleaved", "--local-only"]) project = self.projA project.Sync_NetworkHalf = mock.Mock() project.Sync_LocalHalf = mock.Mock() project.manifest.manifestProject.config = mock.MagicMock() self.mock_context["projects"] = [project] with mock.patch("subcmds.sync.SyncBuffer") as mock_sync_buffer: mock_sync_buf_instance = mock.MagicMock() mock_sync_buf_instance.Finish.return_value = True mock_sync_buf_instance.errors = [] mock_sync_buffer.return_value = mock_sync_buf_instance result_obj = self.cmd._SyncProjectList(opt, [0]) result = result_obj.results[0] self.assertTrue(result.fetch_success) self.assertTrue(result.checkout_success) project.Sync_NetworkHalf.assert_not_called() project.Sync_LocalHalf.assert_called_once() def test_worker_network_only(self): """Test _SyncProjectList with --network-only.""" opt = self._get_opts(["--interleaved", "--network-only"]) project = self.projA project.Sync_NetworkHalf = mock.Mock( return_value=SyncNetworkHalfResult(error=None, remote_fetched=True) ) project.Sync_LocalHalf = mock.Mock() self.mock_context["projects"] = [project] result_obj = self.cmd._SyncProjectList(opt, [0]) result = result_obj.results[0] self.assertTrue(result.fetch_success) self.assertTrue(result.checkout_success) project.Sync_NetworkHalf.assert_called_once() project.Sync_LocalHalf.assert_not_called() class UpdateCopyLinkfileListTest(unittest.TestCase): """Tests for Sync.UpdateCopyLinkfileList.""" def setUp(self): self.tempdirobj = tempfile.TemporaryDirectory(prefix="repo_tests") self.topdir = self.tempdirobj.name self.repodir = os.path.join(self.topdir, ".repo") os.makedirs(self.repodir) manifest = mock.MagicMock() manifest.subdir = self.repodir self.manifest = manifest git_event_log = mock.MagicMock(ErrorEvent=mock.Mock(return_value=None)) self.cmd = sync.Sync( manifest=manifest, outer_client=mock.MagicMock(), git_event_log=git_event_log, ) self.cmd.client = mock.MagicMock(topdir=self.topdir) def tearDown(self): self.tempdirobj.cleanup() def _write_copylinkfile_json(self, data: dict) -> None: path = os.path.join(self.repodir, "copy-link-files.json") with open(path, "w") as f: json.dump(data, f) def _setup_projects(self, linkfile_dests: list) -> None: project = mock.MagicMock() project.linkfiles = [mock.MagicMock(dest=d) for d in linkfile_dests] project.copyfiles = [] mock.patch.object( self.cmd, "GetProjects", return_value=[project] ).start() def test_removes_old_symlink_dest(self): """Old linkfile dests that are symlinks should be removed.""" old_dest = os.path.join(self.topdir, "old-link") os.symlink("target", old_dest) self._write_copylinkfile_json( {"linkfile": ["old-link"], "copyfile": []} ) self._setup_projects([]) self.cmd.UpdateCopyLinkfileList(self.manifest) self.assertFalse(os.path.lexists(old_dest)) def test_does_not_delete_through_new_symlink(self): """Old dests that resolve through a new symlink must not delete files. When the manifest changes from individual linkfiles inside a directory to a single directory linkfile, and _CopyAndLinkFiles has already created the symlink (interleaved mode), cleanup must not follow the symlink and delete real project files. """ project_dir = os.path.join(self.topdir, "vendor", "tools", "llms") os.makedirs(os.path.join(project_dir, "dot-llms", "rules")) os.makedirs(os.path.join(project_dir, "dot-llms", "skills")) with open( os.path.join(project_dir, "dot-llms", "rules", "basics.md"), "w" ) as f: f.write("# basics") with open( os.path.join(project_dir, "dot-llms", "skills", "repo.md"), "w" ) as f: f.write("# repo") # Simulate interleaved mode: .llms -> vendor/tools/llms/dot-llms. llms_link = os.path.join(self.topdir, ".llms") os.symlink("vendor/tools/llms/dot-llms", llms_link) self._write_copylinkfile_json( {"linkfile": [".llms/rules", ".llms/skills"], "copyfile": []} ) self._setup_projects([".llms"]) self.cmd.UpdateCopyLinkfileList(self.manifest) # Real project files must still exist. self.assertTrue( os.path.exists( os.path.join(project_dir, "dot-llms", "rules", "basics.md") ) ) self.assertTrue( os.path.exists( os.path.join(project_dir, "dot-llms", "skills", "repo.md") ), ) self.assertTrue(os.path.islink(llms_link)) def test_cleans_up_empty_parent_dirs(self): """After removing old dests, empty parent directories are removed.""" llms_dir = os.path.join(self.topdir, ".llms") os.makedirs(llms_dir) os.symlink( "../vendor/tools/llms/rules", os.path.join(llms_dir, "rules") ) os.symlink( "../vendor/tools/llms/skills", os.path.join(llms_dir, "skills"), ) self._write_copylinkfile_json( {"linkfile": [".llms/rules", ".llms/skills"], "copyfile": []} ) self._setup_projects([".llms"]) self.cmd.UpdateCopyLinkfileList(self.manifest) self.assertFalse(os.path.lexists(os.path.join(llms_dir, "rules"))) self.assertFalse(os.path.lexists(os.path.join(llms_dir, "skills"))) # Parent directory should be removed since it's now empty. self.assertFalse(os.path.exists(llms_dir)) def test_preserves_nonempty_parent_dirs(self): """Non-empty parent directories are preserved after old dest removal.""" llms_dir = os.path.join(self.topdir, ".llms") os.makedirs(llms_dir) os.symlink( "../vendor/tools/llms/rules", os.path.join(llms_dir, "rules") ) with open(os.path.join(llms_dir, "my-notes.txt"), "w") as f: f.write("user content") self._write_copylinkfile_json( {"linkfile": [".llms/rules"], "copyfile": []} ) self._setup_projects([".llms"]) self.cmd.UpdateCopyLinkfileList(self.manifest) self.assertFalse(os.path.lexists(os.path.join(llms_dir, "rules"))) self.assertTrue(os.path.exists(os.path.join(llms_dir, "my-notes.txt"))) self.assertTrue(os.path.isdir(llms_dir)) class SyncToSuperprojectRevTests(unittest.TestCase): """Tests for Sync._SyncToSuperprojectRev.""" def setUp(self): self.repodir = tempfile.mkdtemp(".repo") self.manifest = mock.MagicMock(repodir=self.repodir) self.manifest.superproject = mock.MagicMock() self.manifest.path_prefix = "" self.mp = mock.MagicMock() self.cmd = sync.Sync(manifest=self.manifest) self.cmd.outer_manifest = self.manifest self.opt = mock.Mock() self.opt.verbose = False self.opt.superproject_revision = "deadbeef" self.opt.mp_update = True self.errors = [] def tearDown(self): shutil.rmtree(self.repodir) @mock.patch("subcmds.sync.GitCommand") def test_successful_sync(self, mock_git_command): """Test successful sync to superproject rev.""" mock_superproject = self.manifest.superproject mock_superproject.Sync.return_value = mock.Mock(success=True) mock_git = mock.Mock() mock_git.Wait.return_value = 0 mock_git.stdout = "proj branch manifest_commit_hash\n" mock_git_command.return_value = mock_git with mock.patch.object( self.cmd, "_UpdateManifestProject" ) as mock_update: self.cmd._SyncToSuperprojectRev( self.opt, self.manifest, self.mp, "name", self.errors ) mock_superproject.SetRevisionId.assert_called_with("deadbeef") mock_superproject.Sync.assert_called_once() mock_git_command.assert_called_once() self.mp.SetRevision.assert_called_with("manifest_commit_hash") mock_update.assert_called_once() self.assertEqual(self.errors, []) @mock.patch("subcmds.sync.GitCommand") def test_parse_error(self, mock_git_command): """Test error when .supermanifest cannot be parsed.""" mock_superproject = self.manifest.superproject mock_superproject.Sync.return_value = mock.Mock(success=True) mock_git = mock.Mock() mock_git.Wait.return_value = 0 # Invalid format (not 3 parts) mock_git.stdout = "invalid_content\n" mock_git_command.return_value = mock_git with self.assertRaises(sync.SyncError) as e: self.cmd._SyncToSuperprojectRev( self.opt, self.manifest, self.mp, "name", self.errors ) self.assertIn("could not parse .supermanifest", str(e.exception)) @mock.patch("subcmds.sync.GitCommand") def test_read_error(self, mock_git_command): """Test error when reading .supermanifest fails.""" mock_superproject = self.manifest.superproject mock_superproject.Sync.return_value = mock.Mock(success=True) mock_git = mock.Mock() mock_git.Wait.return_value = 1 mock_git.stderr = "git error" mock_git_command.return_value = mock_git with self.assertRaises(sync.SyncError) as e: self.cmd._SyncToSuperprojectRev( self.opt, self.manifest, self.mp, "name", self.errors ) self.assertIn("failed to read .supermanifest", str(e.exception)) def test_no_superproject(self): """Test error when superproject is not defined.""" self.manifest.superproject = None with self.assertRaises(sync.SyncError) as e: self.cmd._SyncToSuperprojectRev( self.opt, self.manifest, self.mp, "name", self.errors ) self.assertIn("superproject not defined", str(e.exception)) @mock.patch("subcmds.sync.GitCommand") def test_sync_failure(self, mock_git_command): """Test error when superproject sync fails.""" mock_superproject = self.manifest.superproject mock_superproject.Sync.return_value = mock.Mock(success=False) with self.assertRaises(sync.SyncError) as e: self.cmd._SyncToSuperprojectRev( self.opt, self.manifest, self.mp, "name", self.errors ) self.assertIn("failed to sync superproject", str(e.exception)) class UpdateAllManifestProjectsTests(unittest.TestCase): """Tests for Sync._UpdateAllManifestProjects.""" def setUp(self): self.repodir = tempfile.mkdtemp(".repo") self.manifest = mock.MagicMock(repodir=self.repodir) self.manifest.superproject = mock.MagicMock() self.manifest.path_prefix = "" self.manifest.standalone_manifest_url = None self.manifest.submanifests = {} self.mp = mock.MagicMock() self.mp.manifest = self.manifest self.mp.standalone_manifest_url = None self.cmd = sync.Sync(manifest=self.manifest) self.cmd.outer_manifest = self.manifest self.opt = mock.Mock() self.opt.verbose = False self.opt.superproject_revision = None self.opt.mp_update = True self.errors = [] def tearDown(self): shutil.rmtree(self.repodir) def test_superproject_revision_outer_manifest(self): """Test that _SyncToSuperprojectRev is called for outer manifest.""" self.opt.superproject_revision = "deadbeef" with mock.patch.object( self.cmd, "_SyncToSuperprojectRev" ) as mock_sync_to_rev: self.cmd._UpdateAllManifestProjects( self.opt, self.mp, "name", self.errors ) mock_sync_to_rev.assert_called_once_with( self.opt, self.manifest, self.mp, "name", self.errors ) def test_superproject_revision_submanifest(self): """Test that _SyncToSuperprojectRev is NOT called for submanifest.""" self.opt.superproject_revision = "deadbeef" submanifest = mock.MagicMock() submanifest.path_prefix = "sub/" submanifest.standalone_manifest_url = None self.mp.manifest = submanifest with mock.patch.object( self.cmd, "_SyncToSuperprojectRev" ) as mock_sync_to_rev: with mock.patch.object( self.cmd, "_UpdateManifestProject" ) as mock_update_manifest: self.cmd._UpdateAllManifestProjects( self.opt, self.mp, "name", self.errors ) mock_sync_to_rev.assert_not_called() mock_update_manifest.assert_called_once() class TestSmartSyncSetupRemoteHelper(unittest.TestCase): """Tests for _SmartSyncSetup with remote helpers.""" def setUp(self): self.cmd = sync.Sync() self.opt = mock.MagicMock() self.opt.quiet = False self.opt.smart_sync = True self.manifest = mock.MagicMock() self.smart_sync_manifest_path = "/fake/path/to/manifest.xml" @mock.patch("shutil.which") @mock.patch("subprocess.Popen") @mock.patch("xmlrpc.client.Server") @mock.patch("subcmds.sync.PersistentTransport") def test_smart_sync_setup_with_helper( self, mock_transport_class, mock_server_class, mock_popen, mock_which ): """Test _SmartSyncSetup when a helper is present and succeeds.""" import subprocess self.manifest.manifest_server = ( "persistent-https://android-smartsync.corp.google.com/" "manifestserver" ) self.manifest.manifest_server_helper = "repo-remote-sso" mock_which.return_value = "/fake/bin/repo-remote-sso" # Mock subprocess to return a JSON with status ok and proxy address mock_process = mock.MagicMock() mock_process.communicate.return_value = ( '{"status":"ok","message":"http://127.0.0.1:999"}\n', "", ) mock_process.returncode = 0 mock_popen.return_value = mock_process # Mock XML-RPC server call mock_server = mock.MagicMock() mock_server.GetApprovedManifest.return_value = [ True, "", ] mock_server_class.return_value = mock_server # Mock manifest project branch self.cmd._GetBranch = mock.MagicMock(return_value="main") self.cmd._ReloadManifest = mock.MagicMock() # Mock open to avoid writing to disk with mock.patch("builtins.open", mock.mock_open()): manifest_name = self.cmd._SmartSyncSetup( self.opt, self.smart_sync_manifest_path, self.manifest ) # Assertions mock_which.assert_called_once_with("repo-remote-sso") mock_popen.assert_called_once_with( ["repo-remote-sso", self.manifest.manifest_server], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) # Verify transport was created with the proxy returned by helper (with # http:// prepended) mock_transport_class.assert_called_once_with( self.manifest.manifest_server, proxy="http://127.0.0.1:999" ) # Verify Server was created with the same URL, with persistent- stripped mock_server_class.assert_called_once_with( "https://android-smartsync.corp.google.com/manifestserver", transport=mock_transport_class.return_value, ) self.assertEqual(manifest_name, "manifest.xml") @mock.patch("shutil.which") @mock.patch("subprocess.Popen") def test_smart_sync_setup_helper_error(self, mock_popen, mock_which): """Test _SmartSyncSetup when helper returns an error status.""" self.manifest.manifest_server = ( "http://android-smartsync.corp.google.com/manifestserver" ) self.manifest.manifest_server_helper = "repo-remote-sso" mock_which.return_value = "/fake/bin/repo-remote-sso" # Mock subprocess to return a JSON with status error mock_process = mock.MagicMock() mock_process.communicate.return_value = ( '{"status":"error","message":"uplink-helper failed"}\n', "", ) mock_process.returncode = 0 mock_popen.return_value = mock_process with self.assertRaises(sync.SmartSyncError) as context: self.cmd._SmartSyncSetup( self.opt, self.smart_sync_manifest_path, self.manifest ) self.assertIn( "helper repo-remote-sso returned error: uplink-helper failed", str(context.exception), ) @mock.patch("shutil.which") def test_smart_sync_setup_missing_declared_helper(self, mock_which): """Test _SmartSyncSetup when helper declared in manifest is missing.""" self.manifest.manifest_server = ( "http://android-smartsync.corp.google.com/manifestserver" ) self.manifest.manifest_server_helper = "repo-remote-sso" mock_which.return_value = None with self.assertRaises(sync.SmartSyncError) as context: self.cmd._SmartSyncSetup( self.opt, self.smart_sync_manifest_path, self.manifest ) self.assertIn( "helper binary 'repo-remote-sso' declared in manifest was not " "found", str(context.exception), ) @mock.patch("shutil.which") @mock.patch("subprocess.Popen") def test_smart_sync_setup_helper_exit_code_error( self, mock_popen, mock_which ): """Test _SmartSyncSetup when helper exits with non-zero and stderr.""" self.manifest.manifest_server = ( "http://android-smartsync.corp.google.com/manifestserver" ) self.manifest.manifest_server_helper = "repo-remote-sso" mock_which.return_value = "/fake/bin/repo-remote-sso" # Mock subprocess: exit code 1, stderr, and no JSON on stdout mock_process = mock.MagicMock() mock_process.communicate.return_value = ( "", "internal binary error occurred\n", ) mock_process.returncode = 1 mock_popen.return_value = mock_process with self.assertRaises(sync.SmartSyncError) as context: self.cmd._SmartSyncSetup( self.opt, self.smart_sync_manifest_path, self.manifest ) self.assertIn( "helper repo-remote-sso exited with exit code 1. " "Stderr: internal binary error occurred", str(context.exception), ) @mock.patch("shutil.which") @mock.patch("subprocess.Popen") def test_smart_sync_setup_helper_json_decode_error_with_stderr( self, mock_popen, mock_which ): """Test _SmartSyncSetup when helper returns invalid JSON and stderr.""" self.manifest.manifest_server = ( "http://android-smartsync.corp.google.com/manifestserver" ) self.manifest.manifest_server_helper = "repo-remote-sso" mock_which.return_value = "/fake/bin/repo-remote-sso" mock_process = mock.MagicMock() mock_process.communicate.return_value = ( "not a json", "some warning messages\n", ) mock_process.returncode = 0 mock_popen.return_value = mock_process with self.assertRaises(sync.SmartSyncError) as context: self.cmd._SmartSyncSetup( self.opt, self.smart_sync_manifest_path, self.manifest ) self.assertIn( "failed to parse JSON from helper repo-remote-sso", str(context.exception), ) self.assertIn( "Stderr was: some warning messages", str(context.exception), ) @mock.patch("shutil.which") @mock.patch("subprocess.Popen") def test_smart_sync_setup_helper_error_with_stderr( self, mock_popen, mock_which ): """Test _SmartSyncSetup when helper returns error status and stderr.""" self.manifest.manifest_server = ( "http://android-smartsync.corp.google.com/manifestserver" ) self.manifest.manifest_server_helper = "repo-remote-sso" mock_which.return_value = "/fake/bin/repo-remote-sso" mock_process = mock.MagicMock() mock_process.communicate.return_value = ( '{"status":"error","message":"uplink-helper failed"}\n', "debugging logs\n", ) mock_process.returncode = 0 mock_popen.return_value = mock_process with self.assertRaises(sync.SmartSyncError) as context: self.cmd._SmartSyncSetup( self.opt, self.smart_sync_manifest_path, self.manifest ) self.assertIn( "helper repo-remote-sso returned error: uplink-helper failed", str(context.exception), ) self.assertIn( "Stderr was: debugging logs", str(context.exception), )