# 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 contextlib import json import optparse import os from pathlib import Path import shutil import tempfile import threading import time from typing import Any, Dict, Iterator, List, Optional, Tuple, Type import unittest from unittest import mock import xmlrpc.client import xmlrpc.server import pytest import command import error import git_command 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" def test_sync_update_projects_revision_id_populates_logging_data( tmp_path: Path, ) -> None: """Test that _UpdateProjectsRevisionId fills in the caller's dict.""" 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([]) opts.this_manifest_only = True opts.local_only = False superproject_logging_data = {} with mock.patch.object( cmd, "ManifestList", return_value=[manifest] ), mock.patch.object( cmd, "_ConfigureSuperproject", return_value=False ), mock.patch.object( sync.git_superproject, "UseSuperproject", return_value=True ): cmd._UpdateProjectsRevisionId( opts, args, superproject_logging_data, manifest ) assert superproject_logging_data == { "superproject": True, "haslocalmanifests": False, "hassuperprojecttag": True, "updatedrevisionid": False, } @pytest.mark.parametrize( "superproject_revision, is_outer_manifest, expected_revision", [ ("deadbeef", True, "deadbeef"), ("deadbeef", False, None), (None, True, None), ], ids=["outer_manifest", "not_outer_manifest", "no_superproject_revision"], ) def test_sync_update_projects_revision_id_pins_superproject_revision( tmp_path: Path, superproject_revision: Optional[str], is_outer_manifest: bool, expected_revision: Optional[str], ) -> None: """Test that --superproject-revision is applied to a reloaded manifest. _SyncToSuperprojectRev pins the superproject, but updating the manifest project unloads the manifest, so _UpdateProjectsRevisionId must pin the freshly loaded superproject again before using it. """ manifest = _create_manifest_with_groups(tmp_path) cmd = sync.Sync() cmd.manifest = manifest cmd.outer_manifest = manifest if is_outer_manifest else mock.MagicMock() # Simulate the state after a manifest reload: a new, unpinned superproject. superproject = mock.MagicMock() superproject.UpdateProjectsRevisionId.return_value = mock.MagicMock( manifest_path=None, fatal=False ) manifest._superproject = superproject opts, args = cmd.OptionParser.parse_args([]) opts.verbose = False opts.this_manifest_only = True opts.local_only = False opts.superproject_revision = superproject_revision with mock.patch.object( cmd, "ManifestList", return_value=[manifest] ), mock.patch.object( sync.git_superproject, "UseSuperproject", return_value=True ), mock.patch.object( sync.git_superproject, "PrintMessages", return_value=False ): cmd._UpdateProjectsRevisionId(opts, args, {}, manifest) if expected_revision is not None: superproject.SetRevisionId.assert_called_once_with(expected_revision) # The pin must be in place before the superproject is used. method_calls = [name for name, _, _ in superproject.mock_calls] assert method_calls.index("SetRevisionId") < method_calls.index( "UpdateProjectsRevisionId" ) else: superproject.SetRevisionId.assert_not_called() superproject.UpdateProjectsRevisionId.assert_called_once() def test_sync_update_projects_revision_id_logs_without_superproject_tag( tmp_path: Path, ) -> None: """Test that sync state is recorded when the manifest has no superproject. SyncAnalysisState never removes keys, so values from an earlier sync would otherwise persist in the config. """ manifest = _create_manifest_with_groups(tmp_path) cmd = sync.Sync() cmd.manifest = manifest opts, args = cmd.OptionParser.parse_args([]) opts.this_manifest_only = True opts.local_only = False superproject_logging_data = {} cmd._UpdateProjectsRevisionId( opts, args, superproject_logging_data, manifest ) assert superproject_logging_data == { "superproject": False, "haslocalmanifests": False, "hassuperprojecttag": False, } @pytest.mark.parametrize( "generate_manpages, expected_default", [ (False, "7; based on number of CPU cores"), (True, "based on number of CPU cores"), ], ids=("interactive", "manpages"), ) def test_jobs_checkout_help_default( generate_manpages: bool, expected_default: str, ) -> None: """Test checkout-jobs default help in interactive and manpage modes.""" with mock.patch.object(sync, "DEFAULT_LOCAL_JOBS", 7), mock.patch.object( command, "GENERATE_MANPAGES", generate_manpages, ): help_text = " ".join(sync.Sync().OptionParser.format_help().split()) assert f"defaults to --jobs or {expected_default}" in help_text if generate_manpages: assert "defaults to --jobs or 7" not in help_text # 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 @pytest.mark.parametrize( "argv, sync_smartsync_manifest, expected_smart_sync, expected_implicit", [ ([], False, False, False), ([], None, False, False), ([], True, True, True), (["-s"], False, True, False), (["--smart-sync"], False, True, False), (["--smart-sync"], True, True, False), (["--no-smart-sync"], True, False, False), (["--no-smart-sync"], False, False, False), (["--head"], True, False, False), (["--head"], False, False, False), (["-t", "tag123"], True, False, False), (["--smart-tag=tag123"], True, False, False), (["-m", "other.xml"], True, False, False), (["--manifest-name=other.xml"], True, False, False), (["-l"], True, False, False), (["--local-only"], True, False, False), (["--nmu"], True, True, True), (["--no-manifest-update"], True, True, True), (["--nmu"], False, False, False), (["--superproject-revision=abc"], True, False, False), (["-s", "-l"], True, True, False), (["-s", "--nmu"], True, True, False), (["-s", "--superproject-revision=abc"], True, True, False), ], ) def test_cli_smart_sync( argv: List[str], sync_smartsync_manifest: Optional[bool], expected_smart_sync: bool, expected_implicit: bool, ) -> None: """Tests --smart-sync and --no-smart-sync option behavior with manifest default. """ manifest = mock.MagicMock() manifest.default.sync_smartsync = sync_smartsync_manifest manifest.manifestProject.smart_sync = None cmd = sync.Sync(manifest=manifest) opts, args = cmd.OptionParser.parse_args(argv) cmd.ValidateOptions(opts, args) implicit = cmd._ResolveSmartSyncOption(opts, manifest) assert opts.smart_sync == expected_smart_sync assert implicit == expected_implicit @pytest.mark.parametrize( "argv, config_smart_sync, user_pref, sync_smartsync_manifest, " "expected_smart_sync, expected_implicit", [ # Config True overrides manifest False/None and user_pref False. ([], True, None, False, True, True), ([], True, None, None, True, True), ([], True, False, False, True, True), # Config False overrides manifest True and user_pref True. ([], False, None, True, False, False), ([], False, True, True, False, False), # Config None falls back to user_pref, then manifest. ([], None, True, False, True, False), ([], None, False, True, False, False), ([], None, None, True, True, True), ([], None, None, False, False, False), # CLI flags override config. (["-s"], False, None, False, True, False), (["--smart-sync"], False, None, True, True, False), (["--no-smart-sync"], True, None, True, False, False), (["--no-smart-sync"], True, None, False, False, False), (["--head"], True, None, True, False, False), (["--head"], True, None, False, False, False), # Conflicting flags suppress implicit smart sync from config. (["-t", "tag123"], True, None, False, False, False), (["-m", "other.xml"], True, None, False, False, False), (["-l"], True, None, False, False, False), (["--nmu"], True, None, False, True, True), (["--superproject-revision=abc"], True, None, False, False, False), (["-s", "-l"], True, None, False, True, False), ], ) def test_cli_smart_sync_with_config( argv: List[str], config_smart_sync: Optional[bool], user_pref: Optional[bool], sync_smartsync_manifest: Optional[bool], expected_smart_sync: bool, expected_implicit: bool, ) -> None: """Tests repo.smartsync config precedence between CLI, user, and manifest. """ manifest = mock.MagicMock() manifest.default.sync_smartsync = sync_smartsync_manifest manifest.manifestProject.smart_sync = config_smart_sync user_cfg = mock.MagicMock() user_cfg.GetBoolean.return_value = user_pref cmd = sync.Sync(manifest=manifest) opts, args = cmd.OptionParser.parse_args(argv) cmd.ValidateOptions(opts, args) with mock.patch.object(sync.RepoConfig, "ForUser", return_value=user_cfg): implicit = cmd._ResolveSmartSyncOption(opts, manifest) assert opts.smart_sync == expected_smart_sync assert implicit == expected_implicit @pytest.mark.parametrize( "argv, user_pref, sync_smartsync_manifest, expected_smart_sync, " "expected_implicit", [ # sync.smartsync=false: never smart sync unless -s/-t is given. ([], False, True, False, False), ([], False, False, False, False), (["-s"], False, True, True, False), (["--smart-sync"], False, False, True, False), (["-t", "tag123"], False, True, False, False), (["--nmu"], False, True, False, False), # sync.smartsync=true: like -s, counted as explicit, not implicit. ([], True, False, True, False), ([], True, True, True, False), (["--nmu"], True, False, True, False), (["--no-smart-sync"], True, True, False, False), (["-t", "tag123"], True, False, False, False), (["-m", "other.xml"], True, False, False, False), (["-l"], True, False, False, False), (["--superproject-revision=abc"], True, False, False, False), # sync.smartsync unset: manifest default applies as before. ([], None, True, True, True), ([], None, False, False, False), ], ) def test_user_config_smart_sync( argv: List[str], user_pref: Optional[bool], sync_smartsync_manifest: bool, expected_smart_sync: bool, expected_implicit: bool, ) -> None: """Tests the sync.smartsync user setting against CLI and manifest.""" manifest = mock.MagicMock() manifest.default.sync_smartsync = sync_smartsync_manifest user_cfg = mock.MagicMock() user_cfg.GetBoolean.return_value = user_pref cmd = sync.Sync(manifest=manifest) opts, args = cmd.OptionParser.parse_args(argv) cmd.ValidateOptions(opts, args) with mock.patch.object(sync.RepoConfig, "ForUser", return_value=user_cfg): implicit = cmd._ResolveSmartSyncOption(opts, manifest) user_cfg.GetBoolean.assert_called_once_with("sync.smartsync") assert opts.smart_sync == expected_smart_sync assert implicit == expected_implicit @pytest.mark.parametrize( "value, expected_smart_sync", [("true", True), ("false", False), ("yes", True), ("no", False)], ) def test_user_config_smart_sync_from_file( monkeypatch: pytest.MonkeyPatch, tmp_home_dir: Path, value: str, expected_smart_sync: bool, ) -> None: """Tests that sync.smartsync is read from ~/.repoconfig/config.""" # ForUser() caches its instance process-wide; make it pick up this HOME. monkeypatch.setattr(sync.RepoConfig, "_ForUser", None) monkeypatch.delenv("REPO_CONFIG_DIR", raising=False) cfg_dir = tmp_home_dir / ".repoconfig" cfg_dir.mkdir() (cfg_dir / "config").write_text(f"[sync]\n\tsmartsync = {value}\n") manifest = mock.MagicMock() # Set the manifest default to the opposite so we know the user config won. manifest.default.sync_smartsync = not expected_smart_sync cmd = sync.Sync(manifest=manifest) opts, args = cmd.OptionParser.parse_args([]) cmd.ValidateOptions(opts, args) assert cmd._ResolveSmartSyncOption(opts, manifest) is False assert opts.smart_sync == expected_smart_sync @pytest.mark.parametrize( "argv, user_pref, sync_smartsync_manifest, expected_exception", [ (["-u", "user", "-p", "pass"], None, False, sync.SmartSyncError), (["-u", "user", "-p", "pass"], None, True, None), (["-s", "-u", "user", "-p", "pass"], None, False, None), (["-t", "tag", "-u", "user", "-p", "pass"], None, False, None), ( ["--no-smart-sync", "-u", "user", "-p", "pass"], None, True, sync.SmartSyncError, ), (["-l", "-u", "user", "-p", "pass"], None, True, sync.SmartSyncError), (["--nmu", "-u", "user", "-p", "pass"], None, True, None), ( ["--nmu", "-u", "user", "-p", "pass"], None, False, sync.SmartSyncError, ), (["-s", "-l", "-u", "user", "-p", "pass"], None, True, None), (["-u", "user"], None, False, SystemExit), (["-p", "pass"], None, False, SystemExit), # -u/-p are accepted when smart sync comes from sync.smartsync=true... (["-u", "user", "-p", "pass"], True, False, None), # ...and rejected when sync.smartsync=false drops the manifest default. (["-u", "user", "-p", "pass"], False, True, sync.SmartSyncError), ], ) def test_cli_manifest_server_credentials( argv: List[str], user_pref: Optional[bool], sync_smartsync_manifest: bool, expected_exception: Optional[Type[BaseException]], ) -> None: """Tests -u and -p validation rules.""" manifest = mock.MagicMock() manifest.default.sync_smartsync = sync_smartsync_manifest user_cfg = mock.MagicMock() user_cfg.GetBoolean.return_value = user_pref cmd = sync.Sync(manifest=manifest) opts, args = cmd.OptionParser.parse_args(argv) with mock.patch.object(sync.RepoConfig, "ForUser", return_value=user_cfg): if expected_exception: with pytest.raises(expected_exception): cmd.ValidateOptions(opts, args) cmd._ResolveSmartSyncOption(opts, manifest) else: cmd.ValidateOptions(opts, args) cmd._ResolveSmartSyncOption(opts, manifest) 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=FakeProject(".repo/repo"), ) self.manifest.outer_client = self.manifest 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 = FakeProject("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 = FakeProject("projA") projB = FakeProject("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_same_relpath_projects_keep_separate_state(self) -> None: """Projects with the same relpath keep separate sync state.""" outer = FakeProject("proj") child = FakeProject("proj", path_prefix="sub") self.state.SetFetchTime(outer) self.state.Save() self.state = self._new_state(self._TIME + 1) self.state.SetFetchTime(child) self.assertEqual(self.state.GetFetchTime(outer), self._TIME) self.assertEqual( self.state.GetFetchTime(child), self._TIME + 1, ) def test_partial_sync_with_same_relpath_projects(self) -> None: """Projects with the same relpath keep independent partial sync.""" outer = FakeProject("proj") child = FakeProject("proj", path_prefix="sub") for project in (outer, child): self.state.SetFetchTime(project) self.state.SetCheckoutTime(project) self.state.Save() self.assertFalse(self.state.IsPartiallySynced()) self.state = self._new_state(self._TIME + 1) self.state.SetFetchTime(outer) self.state.SetCheckoutTime(outer) self.assertTrue(self.state.IsPartiallySynced()) def test_save_to_file(self): """Data is saved under repodir.""" p = FakeProject("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 = FakeProject("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 = FakeProject("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 = FakeProject("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 = FakeProject("projA") projB = FakeProject("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_keeps_submanifest_project(self) -> None: """Existing submanifest projects are not pruned.""" project = FakeProject("proj", path_prefix="sub") os.makedirs( os.path.join(self.topdir, "sub", "proj", ".git"), ) self.state.SetFetchTime(project) self.state.PruneRemovedProjects() self.assertEqual(self.state.GetFetchTime(project), self._TIME) def test_prune_from_submanifest_keeps_project(self) -> None: """Existing projects are not pruned during a submanifest sync.""" child_topdir = os.path.join(self.topdir, "sub") project = FakeProject("proj", path_prefix="sub") os.makedirs(os.path.join(child_topdir, "proj", ".git")) child_manifest = mock.MagicMock( topdir=child_topdir, repodir=self.repodir, outer_client=self.manifest, ) with mock.patch("time.time", return_value=self._TIME): state = sync.LocalSyncState(child_manifest) state.SetFetchTime(project) state.PruneRemovedProjects() self.assertEqual(state.GetFetchTime(project), self._TIME) 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 = FakeProject("projA") projB = FakeProject("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: str, name: Optional[str] = None, objdir: Optional[str] = None, parent: Optional["FakeProject"] = None, is_derived: bool = False, revisionId: Optional[str] = None, gitlink_path: Optional[str] = None, path_prefix: str = "", ) -> None: self.relpath = relpath self.path_prefix = path_prefix self.name = name or relpath self.objdir = objdir or relpath self.worktree = relpath self.parent = parent self.is_derived = is_derived self.revisionId = revisionId self.gitlink_path = gitlink_path self.use_git_worktrees = False self.UseAlternates = False self.UseReprojectCmd = False self.manifest = mock.MagicMock() self.manifest.GetProjectsWithName.return_value = [self] self.config = mock.MagicMock() self.EnableRepositoryExtension = mock.MagicMock() @property def Derived(self) -> bool: return self.is_derived def SetRevision( self, revisionExpr: str, revisionId: Optional[str] = None ) -> None: self.revisionExpr = revisionExpr self.revisionId = revisionId or revisionExpr def RelPath(self, local: bool = True) -> str: if local: return self.relpath return os.path.join(self.path_prefix, 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 NestedProjects(unittest.TestCase): def test_flat_manifest(self) -> None: p_foo = FakeProject("foo") p_foo_bar = FakeProject("foo-bar") self.assertEqual(sync._NestedProjects([p_foo, p_foo_bar]), []) def test_nested_paths(self) -> None: p_foo = FakeProject("foo") p_foo_bar = FakeProject("foo/bar") p_foo_bar_baz = FakeProject("foo/bar/baz") self.assertEqual( sync._NestedProjects([p_foo_bar_baz, p_foo, p_foo_bar]), [p_foo_bar, p_foo_bar_baz], ) def test_submodule_of_a_parent(self) -> None: parent = FakeProject("foo") sub = FakeProject("foo/sub", parent=parent, is_derived=True) self.assertEqual(sync._NestedProjects([parent, sub]), [sub]) class ParentFirstBatches(unittest.TestCase): def test_no_submodules(self) -> None: p_a = FakeProject("a") p_a_b = FakeProject("a/b") out = sync._ParentFirstBatches([p_a, p_a_b]) self.assertEqual(out, [[p_a, p_a_b]]) def test_submodules_follow_their_parent(self) -> None: p_a = FakeProject("a") p_a_b = FakeProject("a/b", parent=p_a, is_derived=True) p_a_b_c = FakeProject("a/b/c", parent=p_a_b, is_derived=True) out = sync._ParentFirstBatches([p_a_b_c, p_a, p_a_b]) self.assertEqual(out, [[p_a], [p_a_b], [p_a_b_c]]) class RefreshDerivedRevisions(unittest.TestCase): def _parent_with_submodules(self, **gitlinks: str) -> FakeProject: p_a = FakeProject("a") p_a.GetSubmoduleRevisions = mock.Mock(return_value=gitlinks) return p_a def _submodule(self, parent: FakeProject, path: str) -> FakeProject: return FakeProject( f"a/{path}", parent=parent, is_derived=True, gitlink_path=path ) def test_reads_each_parent_once(self) -> None: p_a = self._parent_with_submodules(b="beef1234", c="cafe1234") p_a_b = self._submodule(p_a, "b") p_a_c = self._submodule(p_a, "c") sync._RefreshDerivedRevisions([p_a, p_a_b, p_a_c]) p_a.GetSubmoduleRevisions.assert_called_once_with() self.assertEqual(p_a_b.revisionId, "beef1234") self.assertEqual(p_a_c.revisionId, "cafe1234") def test_reuses_gitlinks_read_for_an_earlier_level(self) -> None: p_a = self._parent_with_submodules(b="beef1234", c="cafe1234") p_a_b = self._submodule(p_a, "b") p_a_c = self._submodule(p_a, "c") submodule_revisions = {} sync._RefreshDerivedRevisions([p_a_b], submodule_revisions) sync._RefreshDerivedRevisions([p_a_c], submodule_revisions) p_a.GetSubmoduleRevisions.assert_called_once_with() self.assertEqual(p_a_c.revisionId, "cafe1234") def test_tells_apart_projects_with_the_same_path(self) -> None: # Paths are relative to their own (sub)manifest, so two projects can # share one. first = self._parent_with_submodules(b="beef1234") second = self._parent_with_submodules(b="cafe1234") first_sub = self._submodule(first, "b") second_sub = self._submodule(second, "b") submodule_revisions = {} sync._RefreshDerivedRevisions([first_sub], submodule_revisions) sync._RefreshDerivedRevisions([second_sub], submodule_revisions) self.assertEqual(first_sub.revisionId, "beef1234") self.assertEqual(second_sub.revisionId, "cafe1234") def test_ignores_projects_from_the_manifest(self) -> None: p_a = self._parent_with_submodules() sync._RefreshDerivedRevisions([p_a]) p_a.GetSubmoduleRevisions.assert_not_called() def test_reports_submodules_removed_from_their_parent(self) -> None: p_a = self._parent_with_submodules(b="beef1234") p_a_c = self._submodule(p_a, "c") removed = sync._RefreshDerivedRevisions([p_a, p_a_c]) self.assertEqual(removed, [p_a_c]) def test_keeps_submodules_of_an_unreadable_parent(self) -> None: p_a = FakeProject("a") p_a.GetSubmoduleRevisions = mock.Mock(return_value=None) p_a_b = FakeProject( "a/b", parent=p_a, is_derived=True, revisionId="stale", gitlink_path="b", ) removed = sync._RefreshDerivedRevisions([p_a, p_a_b]) self.assertEqual(removed, []) self.assertEqual(p_a_b.revisionId, "stale") class FetchParentFirst(unittest.TestCase): def test_submodules_are_fetched_after_their_parent(self) -> None: cmd = sync.Sync() cmd._fetch_times = mock.Mock() cmd._fetch_times.Get = mock.Mock(return_value=0) calls = [] p_a = FakeProject("a") def fake_read() -> Dict[str, str]: calls.append(("read gitlinks of", p_a.relpath)) return {"b": "beef1234"} p_a.GetSubmoduleRevisions = mock.Mock(side_effect=fake_read) p_a_b = FakeProject( "a/b", parent=p_a, is_derived=True, gitlink_path="b" ) def fake_fetch( projects: List[FakeProject], *_args: object ) -> sync._FetchResult: calls.append(("fetch", [p.relpath for p in projects])) return sync._FetchResult(True, {p.objdir for p in projects}) opt = mock.Mock(fail_fast=False) with mock.patch.object(cmd, "_Fetch", side_effect=fake_fetch): result = cmd._FetchParentFirst([p_a_b, p_a], opt, None, None, []) self.assertTrue(result.success) self.assertEqual(result.projects, {"a", "a/b"}) self.assertEqual( calls, [ ("fetch", ["a"]), ("read gitlinks of", "a"), ("fetch", ["a/b"]), ], ) class WithoutProjects(unittest.TestCase): def test_drops_the_unwanted_projects(self) -> None: p_a = FakeProject("a") p_a_b = FakeProject("a/b") self.assertEqual(sync._WithoutProjects([p_a, p_a_b], [p_a_b]), [p_a]) self.assertEqual(sync._WithoutProjects([p_a, p_a_b], []), [p_a, p_a_b]) def test_keeps_projects_with_the_same_path(self) -> None: # Paths are relative to their own (sub)manifest, so two projects can # share one. first = FakeProject("a/b") second = FakeProject("a/b") self.assertEqual( sync._WithoutProjects([first, second], [second]), [first] ) 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.opt.network_only = False self.tempdirobj = tempfile.TemporaryDirectory(prefix="repo_tests") self.addCleanup(self.tempdirobj.cleanup) self.project = mock.MagicMock(clone_depth="1") self.project.name = "project" self.project.Exists = True self.project.worktree = self.tempdirobj.name self.project.stateless_prune_needed = False self.cmd.git_event_log = mock.MagicMock() self.cmd._bloated_projects = [] def test_one_project_uses_dirty_and_head(self) -> None: """A project whose HEAD left the manifest revision is measured.""" self.project.GetDirtyAndHead.return_value = (False, "local") self.project.GetRevisionId.return_value = "manifest" self.project.bare_git.count_objects.return_value = ( "packs: 0\nsize-pack: 0\nsize-garbage: 0\n" ) with mock.patch.object( sync.Sync, "get_parallel_context", return_value={"projects": [self.project]}, ): self.assertIsNone(self.cmd._CheckOneBloatedProject(0)) self.project.GetDirtyAndHead.assert_called_once_with() self.project.bare_git.count_objects.assert_called_once_with("-v") def test_one_dirty_project_is_measured(self) -> None: """A dirty project is measured even if HEAD matches the manifest.""" self.project.GetDirtyAndHead.return_value = (True, "local") self.project.GetRevisionId.return_value = "local" self.project.bare_git.count_objects.return_value = ( "packs: 0\nsize-pack: 0\nsize-garbage: 0\n" ) with mock.patch.object( sync.Sync, "get_parallel_context", return_value={"projects": [self.project]}, ): self.assertIsNone(self.cmd._CheckOneBloatedProject(0)) self.project.bare_git.count_objects.assert_called_once_with("-v") def test_one_unborn_project_skips_bloat_check(self) -> None: """A project without a resolvable HEAD is skipped.""" self.project.GetDirtyAndHead.return_value = (True, None) with mock.patch.object( sync.Sync, "get_parallel_context", return_value={"projects": [self.project]}, ): self.assertIsNone(self.cmd._CheckOneBloatedProject(0)) self.project.GetRevisionId.assert_not_called() self.project.bare_git.count_objects.assert_not_called() @mock.patch.object(git_command, "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.object(git_command, "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.object(git_command, "git_require", return_value=True) @mock.patch("subcmds.sync.Progress") def test_network_only_skips_check( self, mock_progress: mock.Mock, mock_git_require: mock.Mock ) -> None: """--network-only doesn't read any worktree state.""" self.opt.network_only = True self.cmd.ExecuteInParallel = mock.Mock() self.cmd._CheckForBloatedProjects([self.project], self.opt) mock_progress.assert_not_called() self.cmd.ExecuteInParallel.assert_not_called() @mock.patch.object(git_command, "git_require", return_value=True) @mock.patch("subcmds.sync.Progress") def test_projects_without_worktree_excluded( self, mock_progress: mock.Mock, mock_git_require: mock.Mock ) -> None: """Projects without a checked-out worktree are never scanned.""" self.cmd.ExecuteInParallel = mock.Mock() missing = os.path.join(self.tempdirobj.name, "missing") for attr, value in ( ("worktree", missing), ("worktree", None), ("Exists", False), ): with self.subTest(attr=attr, value=value): mock_progress.reset_mock() self.cmd.ExecuteInParallel.reset_mock() with mock.patch.object(self.project, attr, value): self.cmd._CheckForBloatedProjects([self.project], self.opt) mock_progress.assert_not_called() self.cmd.ExecuteInParallel.assert_not_called() @mock.patch.object(git_command, "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.object(git_command, "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, UseReprojectCmd=False, name="project", relpath="rel_path", parent=None, 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 = error.GitError( "sync_network_half_error error", project=self.project ) self.sync_local_half_error = error.GitError( "sync_local_half_error", project=self.project ) with self.assertRaises(error.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") def _ExecuteUntilSync( self, args: List[str] ) -> Tuple[mock.MagicMock, mock.MagicMock]: """Run Execute up to the sync itself, returning the sync mocks.""" self.opt.mp_update = False with contextlib.ExitStack() as stack: for name in ( "_UpdateRepoProject", "_UpdateProjectsRevisionId", "_ValidateOptionsWithManifest", "_RunPostSyncHook", ): stack.enter_context(mock.patch.object(self.cmd, name)) phased = stack.enter_context( mock.patch.object(self.cmd, "_SyncPhased") ) interleaved = stack.enter_context( mock.patch.object(self.cmd, "_SyncInterleaved") ) self.cmd.Execute(self.opt, args) return phased, interleaved def test_reproject_cmd_allows_a_flat_manifest(self) -> None: """Ensure repo.reprojectcmd syncs a manifest without nesting.""" self.project.UseReprojectCmd = True phased, interleaved = self._ExecuteUntilSync([]) self.assertTrue(phased.called or interleaved.called) def test_reproject_cmd_rejects_nested_projects(self) -> None: """Ensure repo.reprojectcmd fails a manifest with nested projects.""" p_foo = FakeProject("foo") p_foo_bar = FakeProject("foo/bar") p_foo.UseReprojectCmd = p_foo_bar.UseReprojectCmd = True self.cmd.GetProjects.return_value = [p_foo, p_foo_bar] with self.assertRaises(error.SyncError) as e: self._ExecuteUntilSync([]) self.assertIn("foo/bar", str(e.exception)) self.assertNotIn(" - foo\n", str(e.exception)) def test_reproject_cmd_rejects_a_submodule(self) -> None: """Ensure repo.reprojectcmd fails a manifest with a submodule.""" p_foo = FakeProject("foo") p_sub = FakeProject("foo/sub", parent=p_foo, is_derived=True) p_foo.UseReprojectCmd = p_sub.UseReprojectCmd = True self.cmd.GetProjects.return_value = [p_foo, p_sub] with self.assertRaises(error.SyncError) as e: self._ExecuteUntilSync([]) self.assertIn("foo/sub", str(e.exception)) def test_reproject_cmd_checks_the_whole_manifest(self) -> None: """Ensure nesting is checked beyond the projects given as args.""" p_foo = FakeProject("foo") p_foo_bar = FakeProject("foo/bar") p_foo.UseReprojectCmd = p_foo_bar.UseReprojectCmd = True self.cmd.GetProjects.side_effect = lambda args, **kwargs: ( [p_foo_bar] if args else [p_foo, p_foo_bar] ) with self.assertRaises(error.SyncError): self._ExecuteUntilSync(["foo/bar"]) self.assertEqual(self.cmd.GetProjects.call_count, 2) _, kwargs = self.cmd.GetProjects.call_args self.assertEqual(kwargs.get("missing_ok"), True) def test_reproject_cmd_off_ignores_nested_projects(self) -> None: """Ensure nesting is only checked with repo.reprojectcmd in use.""" projects = [FakeProject("foo"), FakeProject("foo/bar")] for p in projects: p.Exists = False self.cmd.GetProjects.return_value = projects phased, interleaved = self._ExecuteUntilSync([]) self.assertTrue(phased.called or interleaved.called) def test_implicit_smart_sync_fallback(self) -> None: """Ensure implicit smart sync silently falls back to ToT.""" self.manifest.default.sync_smartsync = True self.opt.smart_sync = None mock.patch.object(self.cmd, "_UpdateAllManifestProjects").start() self.opt.verbose = False with mock.patch.object( self.cmd, "_SmartSyncSetup", side_effect=sync.SmartSyncError("unreachable"), ) as mock_setup: 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"): with mock.patch.object( sync.logger, "warning" ) as mock_warn: self.cmd.Execute(self.opt, []) mock_setup.assert_called_once() self.assertTrue(mock_setup.call_args.kwargs.get("best_effort")) self.assertFalse(self.opt.smart_sync) self.assertFalse(mock_warn.called) def test_implicit_smart_sync_fallback_verbose(self) -> None: """Ensure the fallback reason is logged when --verbose is given.""" self.manifest.default.sync_smartsync = True self.opt.smart_sync = None mock.patch.object(self.cmd, "_UpdateAllManifestProjects").start() self.opt.verbose = True with mock.patch.object( self.cmd, "_SmartSyncSetup", side_effect=sync.SmartSyncError("unreachable"), ): 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"): with mock.patch.object( sync.logger, "warning" ) as mock_warn: self.cmd.Execute(self.opt, []) self.assertFalse(self.opt.smart_sync) self.assertTrue(mock_warn.called) def test_explicit_smart_sync_error(self) -> None: """Ensure explicit smart sync raises SmartSyncError on failure.""" self.manifest.default.sync_smartsync = True self.opt.smart_sync = True self.opt.mp_update = False with mock.patch.object( self.cmd, "_SmartSyncSetup", side_effect=sync.SmartSyncError("unreachable"), ) as mock_setup: with self.assertRaises(sync.SmartSyncError): self.cmd.Execute(self.opt, []) mock_setup.assert_called_once() self.assertFalse(mock_setup.call_args.kwargs.get("best_effort")) def test_user_config_smart_sync_disabled(self) -> None: """Ensure sync.smartsync=false skips the manifest's default.""" self.manifest.default.sync_smartsync = True self.opt.smart_sync = None mock.patch.object(self.cmd, "_UpdateAllManifestProjects").start() user_cfg = mock.MagicMock() user_cfg.GetBoolean.return_value = False with mock.patch.object( sync.RepoConfig, "ForUser", return_value=user_cfg ): with mock.patch.object(self.cmd, "_SmartSyncSetup") as mock_setup: 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, []) mock_setup.assert_not_called() self.assertFalse(self.opt.smart_sync) def test_user_config_smart_sync_enabled_error(self) -> None: """Ensure sync.smartsync=true fails hard instead of falling back.""" self.manifest.default.sync_smartsync = False self.opt.smart_sync = None self.opt.mp_update = False user_cfg = mock.MagicMock() user_cfg.GetBoolean.return_value = True with mock.patch.object( sync.RepoConfig, "ForUser", return_value=user_cfg ): with mock.patch.object( self.cmd, "_SmartSyncSetup", side_effect=sync.SmartSyncError("unreachable"), ) as mock_setup: with self.assertRaises(sync.SmartSyncError): self.cmd.Execute(self.opt, []) mock_setup.assert_called_once() def test_implicit_smart_sync_unparsable_manifest(self) -> None: """Ensure an unparsable smart sync manifest falls back to ToT.""" self.manifest.default.sync_smartsync = True self.opt.smart_sync = None mock.patch.object(self.cmd, "_UpdateAllManifestProjects").start() self.opt.verbose = False with mock.patch.object( self.cmd, "_SmartSyncSetup", side_effect=error.ManifestParseError("mismatched tag"), ): 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.assertFalse(self.opt.smart_sync) # A half-applied override would pin the client to a broken manifest. self.manifest.ClearOverride.assert_called_once_with() def test_explicit_smart_sync_unparsable_manifest(self) -> None: """Ensure explicit smart sync still reports the parse error.""" self.manifest.default.sync_smartsync = True self.opt.smart_sync = True self.opt.mp_update = False with mock.patch.object( self.cmd, "_SmartSyncSetup", side_effect=error.ManifestParseError("mismatched tag"), ): with self.assertRaises(error.ManifestParseError): self.cmd.Execute(self.opt, []) def _ExecuteWithSmartSyncOverride( self, reload_error: Optional[Exception] = None, setup_error: Optional[Exception] = None, ) -> Tuple[str, mock.MagicMock, mock.MagicMock, mock.MagicMock]: """Run Execute with an existing smart sync override on disk. Returns: The override path, and the _SmartSyncSetup, _ReloadManifest and _UpdateAllManifestProjects mocks. """ worktree = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, worktree) self.manifest.manifestProject.worktree = worktree # -l reloads the superproject manifest; keep that out of the way. self.manifest.superproject = None self.opt.verbose = False override_path = os.path.join(worktree, "smart_sync_override.xml") with open(override_path, "w") as f: f.write("") with contextlib.ExitStack() as stack: for name in ( "_UpdateRepoProject", "_ValidateOptionsWithManifest", "_SyncInterleaved", "_RunPostSyncHook", ): stack.enter_context(mock.patch.object(self.cmd, name)) setup = stack.enter_context( mock.patch.object( self.cmd, "_SmartSyncSetup", side_effect=setup_error ) ) reload = stack.enter_context( mock.patch.object( self.cmd, "_ReloadManifest", side_effect=reload_error ) ) update_mps = stack.enter_context( mock.patch.object(self.cmd, "_UpdateAllManifestProjects") ) self.cmd.Execute(self.opt, []) return override_path, setup, reload, update_mps def test_smart_sync_override_reused_offline(self) -> None: """Ensure -l reuses an existing smart sync manifest. This applies whether the override came from the manifest's sync-smartsync default or from an explicit -s, so that `sync -n` followed by `sync -l` matches a single sync. """ for sync_smartsync in (True, False): for mp_update in (True, False): with self.subTest( sync_smartsync=sync_smartsync, mp_update=mp_update ): self.manifest.default.sync_smartsync = sync_smartsync self.opt.smart_sync = None self.opt.local_only = True self.opt.mp_update = mp_update path, setup, reload, update_mps = ( self._ExecuteWithSmartSyncOverride() ) setup.assert_not_called() reload.assert_called_once_with( "smart_sync_override.xml", self.manifest ) self.assertTrue(os.path.isfile(path)) if mp_update: self.assertEqual( update_mps.call_args[0][2], "smart_sync_override.xml", ) def test_no_manifest_update_keeps_default_smart_sync(self) -> None: """Ensure --nmu still runs the manifest-driven default smart sync.""" self.manifest.default.sync_smartsync = True self.opt.smart_sync = None self.opt.mp_update = False _, setup, _, update_mps = self._ExecuteWithSmartSyncOverride() setup.assert_called_once() update_mps.assert_not_called() def test_no_manifest_update_does_not_reuse_override(self) -> None: """Ensure --nmu handles an existing override like a regular sync.""" cases = ( ("manifest without sync-smartsync", False, None), ("failed default smart sync", True, sync.SmartSyncError("down")), ) for name, sync_smartsync, setup_error in cases: with self.subTest(name): self.manifest.default.sync_smartsync = sync_smartsync self.opt.smart_sync = None self.opt.mp_update = False path, _, reload, _ = self._ExecuteWithSmartSyncOverride( setup_error=setup_error ) reload.assert_not_called() self.assertFalse(os.path.isfile(path)) def test_smart_sync_override_removed_offline(self) -> None: """Ensure -l --no-smart-sync still drops the override.""" for sync_smartsync in (True, False): with self.subTest(sync_smartsync=sync_smartsync): self.manifest.default.sync_smartsync = sync_smartsync self.opt.smart_sync = False self.opt.local_only = True path, setup, reload, _ = self._ExecuteWithSmartSyncOverride() setup.assert_not_called() reload.assert_not_called() self.assertFalse(os.path.isfile(path)) def test_smart_sync_override_unparsable_offline(self) -> None: """Ensure an unloadable reused override falls back to ToT.""" self.manifest.default.sync_smartsync = True self.opt.smart_sync = None self.opt.local_only = True path, _, _, update_mps = self._ExecuteWithSmartSyncOverride( reload_error=error.ManifestParseError("mismatched tag") ) self.manifest.ClearOverride.assert_called_once_with() self.assertFalse(os.path.isfile(path)) self.assertIsNone(update_mps.call_args[0][2]) 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 = 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_refreshes_submodule_revision(self) -> None: """Test submodules are synced at the revision of the fetched parent.""" opt, args = self.cmd.OptionParser.parse_args(["--interleaved", "-j4"]) opt.quiet = True submodule = FakeProject( "projA/sub", name="projA_sub", objdir="objA_sub", parent=self.projA, is_derived=True, revisionId="stale", gitlink_path="sub", ) all_projects = [self.projA, submodule] mock.patch.object( self.cmd, "GetProjects", return_value=all_projects ).start() self.projA.GetSubmoduleRevisions = mock.Mock( return_value={"sub": "fetched"} ) synced = [] def execute_side_effect( jobs: int, target: object, work_items: List[List[int]], **kwargs: object, ) -> bool: 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: project = projects_in_pass[project_idx] synced.append((project.relpath, project.revisionId)) synced_relpaths_set.add(project.relpath) return True mock.patch.object( self.cmd, "ExecuteInParallel", side_effect=execute_side_effect ).start() self.cmd._SyncInterleaved( opt, args, [], self.manifest, self.manifest.manifestProject, all_projects, {}, ) self.assertIn(("projA/sub", "fetched"), synced) def test_interleaved_skips_removed_submodule(self) -> None: """Test submodules dropped by their parent are not checked out.""" opt, args = self.cmd.OptionParser.parse_args(["--interleaved", "-j4"]) opt.quiet = True submodule = FakeProject( "projA/sub", name="projA_sub", objdir="objA_sub", parent=self.projA, is_derived=True, revisionId="stale", gitlink_path="sub", ) # The parent no longer holds a gitlink for the submodule. self.projA.GetSubmoduleRevisions = mock.Mock(return_value={}) # The reloaded manifest no longer derives the removed submodule. mock.patch.object( self.cmd, "GetProjects", return_value=[self.projA] ).start() synced = [] def execute_side_effect( jobs: int, target: object, work_items: List[List[int]], **kwargs: object, ) -> bool: 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: project = projects_in_pass[project_idx] synced.append(project.relpath) synced_relpaths_set.add(project.relpath) return True mock.patch.object( self.cmd, "ExecuteInParallel", side_effect=execute_side_effect ).start() self.cmd._SyncInterleaved( opt, args, [], self.manifest, self.manifest.manifestProject, [self.projA, submodule], {}, ) self.assertEqual(synced, ["projA"]) def _make_syncable(self, project: FakeProject) -> FakeProject: project.Sync_NetworkHalf = mock.Mock( return_value=SyncNetworkHalfResult(error=None, remote_fetched=True) ) project.Sync_LocalHalf = mock.Mock() return project def _run_interleaved( self, opt: optparse.Values, initial_projects: List[FakeProject], reloaded_projects: List[FakeProject], ) -> None: """Run _SyncInterleaved with the real workers and callback. |initial_projects| make up the first pass, |reloaded_projects| every later one, the way reloading the manifest between passes does. """ mock.patch.object( self.cmd, "GetProjects", return_value=reloaded_projects ).start() mock.patch.object(self.cmd, "event_log").start() def execute_side_effect( jobs: int, target: object, work_items: List[List[int]], **kwargs: object, ) -> bool: results = [target(item) for item in work_items] return kwargs["callback"](None, kwargs["output"], results) mock.patch.object( self.cmd, "ExecuteInParallel", side_effect=execute_side_effect ).start() with mock.patch("subcmds.sync.SyncBuffer") as mock_sync_buffer: mock_sync_buffer.return_value.Finish.return_value = True mock_sync_buffer.return_value.errors = [] self.cmd._SyncInterleaved( opt, [], [], self.manifest, self.manifest.manifestProject, initial_projects, {}, ) def test_interleaved_syncs_same_path_projects_of_every_manifest( self, ) -> None: """Test a project is not skipped because another shares its path.""" opt = self._get_opts(["--interleaved", "-j4"]) outer = self._make_syncable( FakeProject("foo", name="outer", objdir="a") ) sub = self._make_syncable( FakeProject("foo", name="sub", objdir="b", path_prefix="sub") ) # |sub| is only discovered once the manifest is reloaded after the # first pass has synced |outer|. self._run_interleaved(opt, [outer], [outer, sub]) outer.Sync_LocalHalf.assert_called_once() sub.Sync_LocalHalf.assert_called_once() def test_interleaved_reports_failures_by_a_unique_path(self) -> None: """Test failing projects are listed by a path that is theirs alone.""" opt = self._get_opts(["--interleaved", "-j4"]) outer = self._make_syncable( FakeProject("foo", name="outer", objdir="a") ) sub = self._make_syncable( FakeProject("foo", name="sub", objdir="b", path_prefix="sub") ) sub.Sync_LocalHalf.side_effect = error.GitError("checkout failed") self.cmd.git_event_log = mock.MagicMock() with self.assertRaises(error.SyncError): self._run_interleaved(opt, [outer, sub], [outer, sub]) self.assertEqual( self.cmd._interleaved_err_checkout_results, ["sub/foo"] ) 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_reports_a_path_unique_across_manifests(self) -> None: """Test _SyncResult.relpath tells apart same-path projects.""" project = FakeProject("foo", objdir="objA", path_prefix="sub") self._make_syncable(project) self.mock_context["projects"] = [project] for this_manifest_only, expected in ((False, "sub/foo"), (True, "foo")): with self.subTest(this_manifest_only=this_manifest_only): opt = self._get_opts() opt.this_manifest_only = this_manifest_only with mock.patch("subcmds.sync.SyncBuffer") as mock_sync_buffer: mock_sync_buffer.return_value.Finish.return_value = True mock_sync_buffer.return_value.errors = [] result_obj = self.cmd._SyncProjectList(opt, [0]) self.assertEqual(result_obj.results[0].relpath, expected) def test_worker_fetch_fails(self): """Test _SyncProjectList with a failed fetch.""" opt = self._get_opts() project = self.projA fetch_error = 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 = 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 = 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.object(git_command, "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.object(git_command, "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(error.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.object(git_command, "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(error.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(error.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.object(git_command, "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(error.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.GetApprovedManifestV2.return_value = { "success": True, "manifest_xml": "", } 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("xmlrpc.client.Server") def test_smart_sync_setup_unparsable_manifest( self, mock_server_class: mock.MagicMock ) -> None: """Test _SmartSyncSetup when the server returns a bad manifest. The parse error is left alone so that an explicit smart sync keeps reporting it as-is; recovering from it is up to the caller. """ self.manifest.manifest_server = ( "http://android-smartsync.corp.google.com/manifestserver" ) self.manifest.manifest_server_helper = None mock_server = mock.MagicMock() mock_server.GetApprovedManifestV2.return_value = { "success": True, "manifest_xml": "", } mock_server_class.return_value = mock_server self.cmd._GetBranch = mock.MagicMock(return_value="main") self.cmd._ReloadManifest = mock.MagicMock( side_effect=error.ManifestParseError("mismatched tag") ) with mock.patch("builtins.open", mock.mock_open()): with self.assertRaises(error.ManifestParseError): self.cmd._SmartSyncSetup( self.opt, self.smart_sync_manifest_path, self.manifest ) @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), ) class TestSmartSyncSetupBranch(unittest.TestCase): """Tests for how _SmartSyncSetup picks the branch sent to the server.""" def setUp(self) -> None: self.cmd = sync.Sync() self.opt = mock.MagicMock() self.opt.quiet = True self.opt.smart_sync = True self.opt.smart_tag = None self.manifest = mock.MagicMock() self.manifest.manifest_server = ( "http://android-smartsync.corp.google.com/manifestserver" ) self.manifest.manifest_server_helper = None self.smart_sync_manifest_path = "/fake/path/to/manifest.xml" self.mock_server = mock.MagicMock() self.mock_server.GetApprovedManifestV2.return_value = { "success": True, "manifest_xml": "", } self.cmd._GetBranch = mock.MagicMock(return_value="main") self.cmd._ReloadManifest = mock.MagicMock() self.cmd._ResolveManifestServerTransport = mock.MagicMock( return_value=(self.manifest.manifest_server, mock.MagicMock()) ) def _RunSetup(self, env: Dict[str, str]) -> str: with mock.patch.dict(os.environ, env, clear=True), mock.patch( "xmlrpc.client.Server", return_value=self.mock_server ), mock.patch("builtins.open", mock.mock_open()): return self.cmd._SmartSyncSetup( self.opt, self.smart_sync_manifest_path, self.manifest ) def test_default_branch(self) -> None: """Without SYNC_BRANCH the manifest branch is used.""" self._RunSetup({}) self.cmd._GetBranch.assert_called_once_with( self.manifest.manifestProject ) self.mock_server.GetApprovedManifestV2.assert_called_once_with( {"branch": "main"} ) def test_sync_branch(self) -> None: """SYNC_BRANCH replaces the manifest branch.""" self._RunSetup({"SYNC_BRANCH": "git_main-abfs"}) self.cmd._GetBranch.assert_not_called() self.mock_server.GetApprovedManifestV2.assert_called_once_with( {"branch": "git_main-abfs"} ) def test_sync_branch_with_target(self) -> None: """SYNC_BRANCH and SYNC_TARGET are both forwarded.""" self._RunSetup( { "SYNC_BRANCH": "git_main-abfs", "SYNC_TARGET": "cf_x86_64_phone-trunk_staging-eng", } ) self.mock_server.GetApprovedManifestV2.assert_called_once_with( { "branch": "git_main-abfs", "target": "cf_x86_64_phone-trunk_staging-eng", } ) def test_empty_sync_branch(self) -> None: """An empty SYNC_BRANCH falls back to the manifest branch.""" self._RunSetup({"SYNC_BRANCH": ""}) self.cmd._GetBranch.assert_called_once_with( self.manifest.manifestProject ) self.mock_server.GetApprovedManifestV2.assert_called_once_with( {"branch": "main"} ) class GetApprovedManifestTests(unittest.TestCase): """Tests for _GetApprovedManifest.""" def test_request_with_target_and_best_effort(self) -> None: server = mock.MagicMock() server.GetApprovedManifestV2.return_value = { "success": True, "manifest_xml": "", } result = sync._GetApprovedManifest(server, "main", "tgt", True) self.assertEqual(result, (True, "")) server.GetApprovedManifestV2.assert_called_once_with( {"branch": "main", "target": "tgt", "best_effort": True} ) server.GetApprovedManifest.assert_not_called() def test_request_omits_unset_options(self) -> None: server = mock.MagicMock() server.GetApprovedManifestV2.return_value = { "success": True, "manifest_xml": "", } sync._GetApprovedManifest(server, "main", None, False) server.GetApprovedManifestV2.assert_called_once_with({"branch": "main"}) def test_declined(self) -> None: server = mock.MagicMock() server.GetApprovedManifestV2.return_value = { "success": False, "error_message": "not in rollout", } result = sync._GetApprovedManifest(server, "main", None, True) self.assertEqual(result, (False, "not in rollout")) def test_invalid_response(self) -> None: server = mock.MagicMock() server.GetApprovedManifestV2.return_value = [True, ""] with self.assertRaises(sync.SmartSyncError): sync._GetApprovedManifest(server, "main", None, False) def test_success_missing_manifest(self) -> None: server = mock.MagicMock() server.GetApprovedManifestV2.return_value = {"success": True} with self.assertRaises(sync.SmartSyncError): sync._GetApprovedManifest(server, "main", None, False) def test_success_non_string_manifest(self) -> None: server = mock.MagicMock() for value in (None, 42, [""]): with self.subTest(manifest_xml=value): server.GetApprovedManifestV2.return_value = { "success": True, "manifest_xml": value, } with self.assertRaises(sync.SmartSyncError): sync._GetApprovedManifest(server, "main", None, False) def test_fallback_on_method_not_found_code(self) -> None: server = mock.MagicMock() server.GetApprovedManifestV2.side_effect = xmlrpc.client.Fault( -32601, "no such method" ) server.GetApprovedManifest.return_value = [True, ""] result = sync._GetApprovedManifest(server, "main", "tgt", True) self.assertEqual(result, (True, "")) server.GetApprovedManifest.assert_called_once_with("main", "tgt") def test_other_fault_is_not_retried(self) -> None: server = mock.MagicMock() server.GetApprovedManifestV2.side_effect = xmlrpc.client.Fault( 1, "internal error" ) with self.assertRaises(xmlrpc.client.Fault): sync._GetApprovedManifest(server, "main", "tgt", False) server.GetApprovedManifest.assert_not_called() @contextlib.contextmanager def _LiveManifestServer(**funcs: Any) -> Iterator[xmlrpc.client.ServerProxy]: """Runs a local XML-RPC server exposing |funcs|.""" server = xmlrpc.server.SimpleXMLRPCServer( ("127.0.0.1", 0), logRequests=False ) for name, func in funcs.items(): server.register_function(func, name) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() try: host, port = server.server_address yield xmlrpc.client.ServerProxy(f"http://{host}:{port}/") finally: server.shutdown() server.server_close() thread.join() class GetApprovedManifestLiveTests(unittest.TestCase): """Tests _GetApprovedManifest against a real XML-RPC server.""" def test_legacy_server_fallback(self) -> None: calls = [] def legacy(*args: str) -> List[Any]: calls.append(args) return [True, ""] with _LiveManifestServer(GetApprovedManifest=legacy) as proxy: self.assertEqual( sync._GetApprovedManifest(proxy, "main", "tgt", True), (True, ""), ) self.assertEqual( sync._GetApprovedManifest(proxy, "main", None, True), (True, ""), ) self.assertEqual(calls, [("main", "tgt"), ("main",)]) def test_v2_server(self) -> None: requests = [] def v2(request: Dict[str, Any]) -> Dict[str, Any]: requests.append(request) return {"success": False, "error_message": "declined"} with _LiveManifestServer(GetApprovedManifestV2=v2) as proxy: self.assertEqual( sync._GetApprovedManifest(proxy, "main", None, True), (False, "declined"), ) self.assertEqual(requests, [{"branch": "main", "best_effort": True}])