diff --git a/git/objects/submodule/base.py b/git/objects/submodule/base.py index c023280e9..15fe877e6 100644 --- a/git/objects/submodule/base.py +++ b/git/objects/submodule/base.py @@ -4,17 +4,32 @@ __all__ = ["Submodule", "UpdateProgress"] import gc -from io import BytesIO import logging import ntpath import os import os.path as osp -from pathlib import Path import shlex import stat import sys -import uuid import urllib.parse +import uuid +from io import BytesIO +from pathlib import Path + +# typing ---------------------------------------------------------------------- +from typing import ( + TYPE_CHECKING, + Any, + Callable, + Dict, + Iterator, + List, + Literal, + Mapping, + Sequence, + Union, + cast, +) import git from git.cmd import Git @@ -46,23 +61,7 @@ sm_section, ) -# typing ---------------------------------------------------------------------- - -from typing import ( - Any, - Callable, - Dict, - Iterator, - List, - Literal, - Mapping, - Sequence, - TYPE_CHECKING, - Union, - cast, -) - -from git.types import Commit_ish, PathLike, TBD +from git.types import TBD, Commit_ish, PathLike if TYPE_CHECKING: from git.index import IndexFile @@ -729,6 +728,7 @@ def update( clone_multi_options: Union[Sequence[TBD], None] = None, allow_unsafe_options: bool = False, allow_unsafe_protocols: bool = False, + no_fetch: bool = False, ) -> "Submodule": """Update the repository of this submodule to point to the checkout we point at with the binsha of this instance. @@ -791,6 +791,11 @@ def update( :param allow_unsafe_options: Allow unsafe options to be used, like ``--upload-pack``. + :param no_fetch: + If ``True``, update using locally available objects and remote-tracking + refs without fetching or cloning. Repositories retained after + :meth:`deinit` can be restored without fetching. + :note: Does nothing in bare repositories. @@ -853,7 +858,8 @@ def fetch_remotes(module_repo: "Repo") -> None: ####################################### try: mrepo = self.module() - fetch_remotes(mrepo) + if not no_fetch: + fetch_remotes(mrepo) except InvalidGitRepositoryError: mrepo = None if not init: @@ -884,7 +890,8 @@ def fetch_remotes(module_repo: "Repo") -> None: self._write_git_file_and_module_config(checkout_module_abspath, module_abspath) mrepo = git.Repo(checkout_module_abspath) mrepo.head.reset(mrepo.head.commit, index=True, working_tree=True) - fetch_remotes(mrepo) + if not no_fetch: + fetch_remotes(mrepo) with self.repo.config_writer() as writer: writer.set_value(sm_section(self.name), "url", self.url) @@ -909,6 +916,8 @@ def fetch_remotes(module_repo: "Repo") -> None: + "Cloning url '%s' to '%s' in submodule %r" % (self.url, checkout_module_abspath, self.name), ) if not dry_run: + if no_fetch: + raise ValueError("Missing module at %r but fetching is disabled" % self.path) from None if self.url.startswith("."): url = urllib.parse.urljoin(self.repo.remotes.origin.url + "/", self.url) else: @@ -1057,6 +1066,7 @@ def fetch_remotes(module_repo: "Repo") -> None: dry_run=dry_run, force=force, keep_going=keep_going, + no_fetch=no_fetch, ) # END handle recursive update # END handle dry run diff --git a/git/objects/submodule/root.py b/git/objects/submodule/root.py index d068049c1..a41c50009 100644 --- a/git/objects/submodule/root.py +++ b/git/objects/submodule/root.py @@ -5,19 +5,17 @@ import logging +# typing ------------------------------------------------------------------- +from typing import TYPE_CHECKING, Union + import git from git.exc import InvalidGitRepositoryError +from git.types import Commit_ish from git.util import IterableList from .base import Submodule, UpdateProgress from .util import find_first_remote_branch -# typing ------------------------------------------------------------------- - -from typing import TYPE_CHECKING, Union - -from git.types import Commit_ish - if TYPE_CHECKING: from git.repo import Repo @@ -87,6 +85,7 @@ def update( # type: ignore[override] dry_run: bool = False, force_reset: bool = False, keep_going: bool = False, + no_fetch: bool = False, ) -> "RootModule": """Update the submodules of this repository to the current HEAD commit. @@ -146,6 +145,11 @@ def update( # type: ignore[override] In conjunction with `dry_run`, this can be useful to anticipate all errors when updating submodules. + :param no_fetch: + If ``True``, update using locally available objects and remote-tracking + refs without fetching or cloning. Cached refs are preserved and used even + when a submodule's URL changes. + :return: self """ @@ -254,7 +258,7 @@ def update( # type: ignore[override] # HANDLE URL CHANGE ################### if sm.url != psm.url: - # Add the new remote, remove the old one. + # When fetching, add the new remote and remove the old one. # This way, if the url just changes, the commits will not have # to be re-retrieved. nn = "__new_origin__" @@ -272,33 +276,19 @@ def update( # type: ignore[override] ) if not dry_run: - assert nn not in [r.name for r in rmts] - smr = smm.create_remote(nn, sm.url) - smr.fetch(progress=progress) - - # If we have a tracking branch, it should be available - # in the new remote as well. - if len([r for r in smr.refs if r.remote_head == sm.branch_name]) == 0: - raise ValueError( - "Submodule branch named %r was not available in new submodule remote at %r" - % (sm.branch_name, sm.url) - ) - # END head is not detached - - # Now delete the changed one. - rmt_for_deletion = None + previous_remote = None for remote in rmts: if remote.url == psm.url: - rmt_for_deletion = remote + previous_remote = remote break # END if urls match # END for each remote # If we didn't find a matching remote, but have exactly # one, we can safely use this one. - if rmt_for_deletion is None: + if previous_remote is None: if len(rmts) == 1: - rmt_for_deletion = rmts[0] + previous_remote = rmts[0] else: # If we have not found any remote with the # original URL we may not have a name. This is a @@ -311,45 +301,64 @@ def update( # type: ignore[override] # END handle one single remote # END handle check we found a remote - orig_name = rmt_for_deletion.name - smm.delete_remote(rmt_for_deletion) - # NOTE: Currently we leave tags from the deleted remotes - # as well as separate tracking branches in the possibly - # totally changed repository (someone could have changed - # the url to another project). At some point, one might - # want to clean it up, but the danger is high to remove - # stuff the user has added explicitly. - - # Rename the new remote back to what it was. - smr.rename(orig_name) - - # Early on, we verified that the our current tracking - # branch exists in the remote. Now we have to ensure - # that the sha we point to is still contained in the new - # remote tracking branch. - smsha = sm.binsha - found = False - rref = smr.refs[self.branch_name] - for c in rref.commit.traverse(): - if c.binsha == smsha: - found = True - break - # END traverse all commits in search for sha - # END for each commit - - if not found: - # Adjust our internal binsha to use the one of the - # remote this way, it will be checked out in the - # next step. This will change the submodule relative - # to us, so the user will be able to commit the - # change easily. - _logger.warning( - "Current sha %s was not contained in the tracking\ + if no_fetch: + # A new remote would have no cached refs. Preserve + # the existing refs and tracking configuration for + # offline updates instead of replacing the remote. + previous_remote.set_url(git.Git.polish_url(sm.url, expand_vars=False)) + else: + assert nn not in [r.name for r in rmts] + smr = smm.create_remote(nn, sm.url) + smr.fetch(progress=progress) + + # If we have a tracking branch, it should be available + # in the new remote as well. + if len([r for r in smr.refs if r.remote_head == sm.branch_name]) == 0: + raise ValueError( + "Submodule branch named %r was not available in new submodule remote at %r" + % (sm.branch_name, sm.url) + ) + # END head is not detached + + orig_name = previous_remote.name + smm.delete_remote(previous_remote) + # NOTE: Currently we leave tags from the deleted remotes + # as well as separate tracking branches in the possibly + # totally changed repository (someone could have changed + # the url to another project). At some point, one might + # want to clean it up, but the danger is high to remove + # stuff the user has added explicitly. + + # Rename the new remote back to what it was. + smr.rename(orig_name) + + # Early on, we verified that the our current tracking + # branch exists in the remote. Now we have to ensure + # that the sha we point to is still contained in the new + # remote tracking branch. + smsha = sm.binsha + found = False + rref = smr.refs[self.branch_name] + for c in rref.commit.traverse(): + if c.binsha == smsha: + found = True + break + # END traverse all commits in search for sha + # END for each commit + + if not found: + # Adjust our internal binsha to use the one of the + # remote this way, it will be checked out in the + # next step. This will change the submodule relative + # to us, so the user will be able to commit the + # change easily. + _logger.warning( + "Current sha %s was not contained in the tracking\ branch at the new remote, setting it the the remote's tracking branch", - sm.hexsha, - ) - sm.binsha = rref.commit.binsha - # END reset binsha + sm.hexsha, + ) + sm.binsha = rref.commit.binsha + # END reset binsha # NOTE: All checkout is performed by the base # implementation of update. @@ -379,11 +388,12 @@ def update( # type: ignore[override] if not dry_run: smm = sm.module() smmr = smm.remotes - # As the branch might not exist yet, we will have to fetch - # all remotes to be sure... - for remote in smmr: - remote.fetch(progress=progress) - # END for each remote + # As the branch might not exist yet, fetch all remotes + # unless restricted to locally cached refs. + if not no_fetch: + for remote in smmr: + remote.fetch(progress=progress) + # END for each remote try: tbr = git.Head.create( @@ -433,6 +443,7 @@ def update( # type: ignore[override] dry_run=dry_run, force=force_reset, keep_going=keep_going, + no_fetch=no_fetch, ) # Update recursively depth first - question is which inconsistent state will @@ -451,6 +462,7 @@ def update( # type: ignore[override] dry_run=dry_run, force_reset=force_reset, keep_going=keep_going, + no_fetch=no_fetch, ) # END handle dry_run # END handle recursive diff --git a/test/test_submodule_no_fetch.py b/test/test_submodule_no_fetch.py new file mode 100644 index 000000000..72cec71a7 --- /dev/null +++ b/test/test_submodule_no_fetch.py @@ -0,0 +1,417 @@ +# This module is part of GitPython and is released under the +# 3-Clause BSD License: https://opensource.org/license/bsd-3-clause/ + +from functools import partial +from pathlib import Path +from unittest import mock + +import pytest + +from git import Git, Remote, Repo, RootModule, Submodule +from git.exc import GitCommandError + + +def _commit_file(repo, content): + Path(repo.working_tree_dir, "file").write_text(content, encoding="utf-8") + repo.index.add(["file"]) + return repo.index.commit("Write " + content) + + +def _cached_remote_refs(repo): + return {ref.path: ref.commit.hexsha for remote in repo.remotes for ref in remote.refs} + + +@pytest.fixture +def local_submodule(tmp_path): + """Use only local repositories, with two commits already available in the clone.""" + with Repo.init(tmp_path / "source") as source, Repo.init(tmp_path / "parent") as parent: + # RootModule's URL-change handling currently assumes a master branch. + source.git.symbolic_ref("HEAD", "refs/heads/master") + _commit_file(source, "initial") + _commit_file(source, "cached") + submodule = parent.create_submodule( + "module", "module", source.working_tree_dir, branch=source.head.reference.name + ) + parent.index.commit("Add submodule") + with submodule.module() as module: + yield submodule, source, module + + +@pytest.fixture(params=["submodule", "root", "repo"]) +def update_submodule(request, local_submodule): + submodule, _, _ = local_submodule + if request.param == "submodule": + return submodule.update + if request.param == "root": + update = RootModule(submodule.repo).update + else: + update = submodule.repo.submodule_update + return partial(update, previous_commit=submodule.repo.head.commit, recursive=False) + + +@pytest.fixture(params=["missing", "deinitialized"]) +def uninitialized_submodule(request, local_submodule): + submodule, _, module = local_submodule + metadata = Path(module.git_dir) + module.close() + if request.param == "deinitialized": + submodule.deinit() + else: + submodule.remove(configuration=False, force=True) + assert not submodule.module_exists() + assert metadata.is_dir() == (request.param == "deinitialized") + return submodule, metadata + + +@pytest.mark.parametrize("no_fetch", [None, False, True], ids=["default", "fetch", "no-fetch"]) +def test_update_no_fetch_checks_out_cached_commit(local_submodule, update_submodule, no_fetch): + submodule, source, module = local_submodule + module.create_remote("backup", source.working_tree_dir) + module.head.reset("HEAD~1", index=True, working_tree=True) + assert module.head.commit.binsha != submodule.binsha + options = {} if no_fetch is None else {"no_fetch": no_fetch} + + with mock.patch.object(Remote, "fetch", autospec=True, side_effect=Remote.fetch) as fetch: + update_submodule(**options) + + assert module.head.commit.binsha == submodule.binsha + assert Path(submodule.abspath, "file").read_text(encoding="utf-8") == "cached" + assert fetch.call_count == (0 if no_fetch else 2) + assert {call[0][0].name for call in fetch.call_args_list} == (set() if no_fetch else {"origin", "backup"}) + + +def test_update_no_fetch_to_latest_revision_uses_cached_tip(local_submodule, update_submodule): + submodule, source, module = local_submodule + cached_tip = module.head.reference.tracking_branch().commit + module.head.reset("HEAD~1", index=True, working_tree=True) + submodule.binsha = module.head.commit.binsha + submodule.repo.index.add([submodule]) + submodule.repo.index.commit("Pin submodule to initial commit") + remote_tip = _commit_file(source, "remote-only") + assert submodule.binsha != cached_tip.binsha != remote_tip.binsha + + with mock.patch.object(Remote, "fetch", side_effect=AssertionError("Unexpected fetch")) as fetch: + update_submodule(no_fetch=True, to_latest_revision=True) + + fetch.assert_not_called() + assert module.head.commit == cached_tip + assert module.head.reference.tracking_branch().commit == cached_tip + assert Path(submodule.abspath, "file").read_text(encoding="utf-8") == "cached" + + +def test_update_no_fetch_cannot_check_out_missing_commit(local_submodule, update_submodule): + submodule, source, module = local_submodule + cached_tip = module.head.commit + submodule.binsha = _commit_file(source, "remote-only").binsha + submodule.repo.index.add([submodule]) + submodule.repo.index.commit("Pin submodule to uncached commit") + + with mock.patch.object(Remote, "fetch", side_effect=AssertionError("Unexpected fetch")) as fetch: + with pytest.raises(GitCommandError, match="merge-base"): + update_submodule(no_fetch=True) + + fetch.assert_not_called() + assert module.head.commit == cached_tip + assert module.head.reference.tracking_branch().commit == cached_tip + assert Path(submodule.abspath, "file").read_text(encoding="utf-8") == "cached" + + +@pytest.mark.parametrize("uninitialized_submodule", ["missing"], indirect=True) +@pytest.mark.parametrize("keep_going", [False, True], ids=["raise", "keep-going"]) +def test_update_no_fetch_rejects_missing_submodule(uninitialized_submodule, update_submodule, keep_going, caplog): + submodule, metadata = uninitialized_submodule + parent_config = Path(submodule.repo.git_dir, "config").read_bytes() + + with mock.patch.object(Remote, "fetch") as fetch, mock.patch.object(Submodule, "_clone_repo") as clone: + if keep_going: + update_submodule(no_fetch=True, keep_going=True) + assert "fetching is disabled" in caplog.text + else: + with pytest.raises(ValueError, match="fetching is disabled"): + update_submodule(no_fetch=True) + + fetch.assert_not_called() + clone.assert_not_called() + assert not submodule.module_exists() + assert not Path(submodule.abspath, ".git").exists() + assert not metadata.exists() + assert Path(submodule.repo.git_dir, "config").read_bytes() == parent_config + + +@pytest.mark.parametrize( + "options", + [ + pytest.param({"init": False}, id="init-false"), + pytest.param({"dry_run": True}, id="dry-run"), + ], +) +def test_update_no_fetch_can_skip_uninitialized_submodule(uninitialized_submodule, update_submodule, options): + submodule, metadata = uninitialized_submodule + retained_metadata = metadata.is_dir() + parent_config = Path(submodule.repo.git_dir, "config").read_bytes() + + with mock.patch.object(Remote, "fetch") as fetch, mock.patch.object(Submodule, "_clone_repo") as clone: + update_submodule(no_fetch=True, **options) + + fetch.assert_not_called() + clone.assert_not_called() + assert not submodule.module_exists() + assert not Path(submodule.abspath, ".git").exists() + assert metadata.is_dir() == retained_metadata + assert Path(submodule.repo.git_dir, "config").read_bytes() == parent_config + + +@pytest.mark.parametrize("to_latest_revision", [False, True], ids=["gitlink", "cached-tip"]) +@pytest.mark.parametrize("keep_going", [False, True], ids=["raise", "keep-going"]) +def test_update_no_fetch_restores_deinitialized_submodule( + local_submodule, update_submodule, to_latest_revision, keep_going, caplog +): + submodule, source, module = local_submodule + tracking_branch = module.head.reference.tracking_branch() + cached_tip = tracking_branch.commit + branch_path = module.head.reference.path + cached_refs = _cached_remote_refs(module) + module.head.reset("HEAD~1", index=True, working_tree=True) + assert module.head.commit != cached_tip + if to_latest_revision: + submodule.binsha = module.head.commit.binsha + submodule.repo.index.add([submodule]) + submodule.repo.index.commit("Pin submodule to initial commit") + assert submodule.binsha != cached_tip.binsha + + metadata = Path(module.git_dir) + module.close() + submodule.deinit(force=True) + assert not submodule.module_exists() + assert metadata.is_dir() + remote_tip = _commit_file(source, "remote-only") + assert remote_tip != cached_tip + + with mock.patch.object(Remote, "fetch") as fetch, mock.patch.object(Submodule, "_clone_repo") as clone: + update_submodule(no_fetch=True, to_latest_revision=to_latest_revision, keep_going=keep_going) + + fetch.assert_not_called() + clone.assert_not_called() + assert not caplog.records + assert submodule.module_exists() + assert Path(submodule.abspath, ".git").is_file() + with submodule.module() as restored: + assert Path(restored.git_dir).samefile(metadata) + assert restored.head.commit == cached_tip + assert restored.head.reference.path == branch_path + assert restored.head.reference.tracking_branch().path == tracking_branch.path + assert _cached_remote_refs(restored) == cached_refs + assert Path(submodule.abspath, "file").read_text(encoding="utf-8") == "cached" + assert not restored.is_dirty(untracked_files=True) + with submodule.repo.config_reader() as reader: + assert reader.get_value(f'submodule "{submodule.name}"', "url") == submodule.url + + +@pytest.mark.parametrize("no_fetch", [None, False], ids=["default", "fetch"]) +def test_update_after_deinit_fetches_remote_tip(local_submodule, update_submodule, no_fetch): + submodule, source, module = local_submodule + metadata = Path(module.git_dir) + module.close() + submodule.deinit() + remote_tip = _commit_file(source, "remote-only") + options = {} if no_fetch is None else {"no_fetch": no_fetch} + + with mock.patch.object(Remote, "fetch", autospec=True, side_effect=Remote.fetch) as fetch: + with mock.patch.object(Submodule, "_clone_repo") as clone: + update_submodule(to_latest_revision=True, **options) + + clone.assert_not_called() + fetch.assert_called_once() + assert fetch.call_args[0][0].name == "origin" + assert Path(fetch.call_args[0][0].repo.git_dir).samefile(metadata) + with submodule.module() as restored: + assert Path(restored.git_dir).samefile(metadata) + assert restored.head.commit == remote_tip + assert restored.head.reference.tracking_branch().commit == remote_tip + assert Path(submodule.abspath, "file").read_text(encoding="utf-8") == "remote-only" + + +@pytest.mark.parametrize("uninitialized_submodule", ["deinitialized"], indirect=True) +def test_update_no_fetch_preserves_nonempty_deinitialized_checkout(uninitialized_submodule, update_submodule): + submodule, metadata = uninitialized_submodule + checkout_file = Path(submodule.abspath, "file") + checkout_file.write_text("user content", encoding="utf-8") + parent_config = Path(submodule.repo.git_dir, "config").read_bytes() + + with mock.patch.object(Remote, "fetch") as fetch, mock.patch.object(Submodule, "_clone_repo") as clone: + with pytest.raises(OSError, match="does already exist and is non-empty"): + update_submodule(no_fetch=True) + + fetch.assert_not_called() + clone.assert_not_called() + assert checkout_file.read_text(encoding="utf-8") == "user content" + assert not Path(submodule.abspath, ".git").exists() + assert metadata.is_dir() + assert Path(submodule.repo.git_dir, "config").read_bytes() == parent_config + + +@pytest.mark.parametrize("no_fetch", [False, True], ids=["fetch", "no-fetch"]) +def test_update_no_fetch_is_recursive(local_submodule, update_submodule, no_fetch): + submodule, source, module = local_submodule + child = module.create_submodule("nested", "nested", source.working_tree_dir, branch=source.head.reference.name) + with child.module() as nested: + cached_tip = nested.head.commit + nested.head.reset("HEAD~1", index=True, working_tree=True) + child.binsha = nested.head.commit.binsha + module.index.add([child]) + previous = module.index.commit("Add nested submodule at initial commit") + child.binsha = cached_tip.binsha + module.index.add([child]) + target = module.index.commit("Advance nested submodule") + submodule.binsha = target.binsha + submodule.repo.index.add([submodule]) + submodule.repo.index.commit("Record nested submodule update") + module.head.reset(previous, index=True, working_tree=True) + + with mock.patch.object(Remote, "fetch", autospec=True, side_effect=Remote.fetch) as fetch: + update_submodule(recursive=True, no_fetch=no_fetch) + + assert module.head.commit == target + assert nested.head.commit == cached_tip + assert Path(child.abspath, "file").read_text(encoding="utf-8") == "cached" + assert fetch.call_count == (0 if no_fetch else 2) + assert {call[0][0].repo.git_dir for call in fetch.call_args_list} == ( + set() if no_fetch else {module.git_dir, nested.git_dir} + ) + + +@pytest.mark.parametrize("no_fetch", [False, True], ids=["fetch", "no-fetch"]) +def test_root_update_no_fetch_on_branch_change(local_submodule, no_fetch): + submodule, source, module = local_submodule + branch_name = source.head.reference.name + "-other" + source.create_head(branch_name) + module.remotes.origin.fetch() + tracking_branch = module.remotes.origin.refs[branch_name] + previous = submodule.repo.head.commit + with submodule.config_writer() as writer: + writer.set_value("branch", branch_name) + submodule.repo.index.commit("Change submodule branch") + module.head.reset("HEAD~1", index=True, working_tree=True) + + with mock.patch.object(Remote, "fetch", autospec=True, side_effect=Remote.fetch) as fetch: + RootModule(submodule.repo).update(previous_commit=previous, recursive=False, no_fetch=no_fetch) + + assert module.head.reference.name == branch_name + assert module.head.reference.tracking_branch() == tracking_branch + assert module.head.commit.binsha == submodule.binsha + if no_fetch: + fetch.assert_not_called() + else: + assert fetch.called + + +@pytest.mark.parametrize("no_fetch", [False, True], ids=["fetch", "no-fetch"]) +def test_root_update_no_fetch_on_url_change(local_submodule, tmp_path, no_fetch): + submodule, source, module = local_submodule + previous = submodule.repo.head.commit + branch_path = module.head.reference.path + tracking_branch = module.head.reference.tracking_branch() + cached_refs = _cached_remote_refs(module) + module.head.reset("HEAD~1", index=True, working_tree=True) + assert module.head.commit.binsha != submodule.binsha + with source.clone(tmp_path / "mirror") as mirror: + mirror_url = Git.polish_url(mirror.working_tree_dir) + with submodule.config_writer() as writer: + writer.set_value("url", mirror_url) + submodule.repo.index.commit("Change submodule URL") + + with mock.patch.object(Remote, "fetch", autospec=True, side_effect=Remote.fetch) as fetch: + RootModule(submodule.repo).update(previous_commit=previous, recursive=False, no_fetch=no_fetch) + + if no_fetch: + fetch.assert_not_called() + assert module.head.reference.path == branch_path + assert module.head.reference.tracking_branch() == tracking_branch + assert _cached_remote_refs(module) == cached_refs + else: + assert fetch.call_count == 2 + assert {call[0][0].name for call in fetch.call_args_list} == {"origin"} + assert module.remotes.origin.url == mirror_url + assert {remote.name for remote in module.remotes} == {"origin"} + + assert module.head.commit.binsha == submodule.binsha + assert Path(submodule.abspath, "file").read_text(encoding="utf-8") == "cached" + + +@pytest.mark.parametrize("change_branch", [False, True], ids=["same-branch", "branch-change"]) +def test_root_update_no_fetch_url_change_uses_cached_tip(local_submodule, tmp_path, change_branch): + submodule, source, module = local_submodule + branch_name = source.head.reference.name + if change_branch: + branch_name += "-other" + source.create_head(branch_name) + module.remotes.origin.fetch() + tracking_branch = module.remotes.origin.refs[branch_name] + cached_tip = tracking_branch.commit + cached_refs = _cached_remote_refs(module) + module.head.reset("HEAD~1", index=True, working_tree=True) + submodule.binsha = module.head.commit.binsha + submodule.repo.index.add([submodule]) + previous = submodule.repo.index.commit("Pin submodule to initial commit") + assert submodule.binsha != cached_tip.binsha + + with source.clone(tmp_path / "mirror") as mirror: + if change_branch: + mirror.create_head(branch_name).checkout() + remote_tip = _commit_file(mirror, "remote-only") + assert remote_tip != cached_tip + mirror_url = Git.polish_url(mirror.working_tree_dir) + with submodule.config_writer() as writer: + writer.set_value("url", mirror_url) + writer.set_value("branch", branch_name) + submodule.repo.index.commit("Change submodule URL and tracking branch") + + with mock.patch.object(Remote, "fetch", side_effect=AssertionError("Unexpected fetch")) as fetch: + RootModule(submodule.repo).update( + previous_commit=previous, recursive=False, no_fetch=True, to_latest_revision=True + ) + + fetch.assert_not_called() + assert module.remotes.origin.url == mirror_url + assert {remote.name for remote in module.remotes} == {"origin"} + assert _cached_remote_refs(module) == cached_refs + assert module.head.reference.name == branch_name + assert module.head.reference.tracking_branch() == tracking_branch + assert module.head.commit == cached_tip + assert Path(submodule.abspath, "file").read_text(encoding="utf-8") == "cached" + + +@pytest.mark.parametrize("multiple_remotes", [False, True], ids=["single-remote-fallback", "matching-remote"]) +def test_root_update_no_fetch_selects_url_change_remote(local_submodule, tmp_path, multiple_remotes): + submodule, source, module = local_submodule + previous = submodule.repo.head.commit + original_url = module.remotes.origin.url + module.remotes.origin.set_url(Git.polish_url(str(tmp_path / "unrelated"))) + if multiple_remotes: + upstream = module.create_remote("upstream", original_url) + upstream.fetch() + module.head.reference.set_tracking_branch(upstream.refs[submodule.branch_name]) + else: + module.remotes.origin.rename("upstream") + tracking_branch = module.head.reference.tracking_branch() + cached_refs = _cached_remote_refs(module) + module.head.reset("HEAD~1", index=True, working_tree=True) + assert module.head.commit.binsha != submodule.binsha + + with source.clone(tmp_path / "mirror") as mirror: + mirror_url = Git.polish_url(mirror.working_tree_dir) + expected_urls = {remote.name: remote.url for remote in module.remotes} + expected_urls["upstream"] = mirror_url + with submodule.config_writer() as writer: + writer.set_value("url", mirror_url) + submodule.repo.index.commit("Change submodule URL") + + with mock.patch.object(Remote, "fetch", side_effect=AssertionError("Unexpected fetch")) as fetch: + RootModule(submodule.repo).update(previous_commit=previous, recursive=False, no_fetch=True) + + fetch.assert_not_called() + assert {remote.name: remote.url for remote in module.remotes} == expected_urls + assert _cached_remote_refs(module) == cached_refs + assert module.head.reference.tracking_branch() == tracking_branch + assert module.head.commit.binsha == submodule.binsha + assert Path(submodule.abspath, "file").read_text(encoding="utf-8") == "cached"