get_remote_url now redirects forks to upstream

(DryRunBackend)

Author: ytausch

conda_forge_tick/git_utils.py

@@ -543,8 +543,8 @@ def does_repository_exist(self, owner: str, repo_name: str) -> bool:
         """
         pass
 
-    @staticmethod
     def get_remote_url(
+        self,
         owner: str,
         repo_name: str,
         connection_mode: GitConnectionMode = GitConnectionMode.HTTPS,
@@ -558,6 +558,8 @@ def get_remote_url(
         :param token: A token to use for authentication. If falsy, no token is used. Use get_authenticated_remote_url
         instead if you want to use the token of the current user.
         :raises ValueError: If the connection mode is not supported.
+        :raises RepositoryNotFoundError: If the repository does not exist. This is only raised if the backend relies on
+        the repository existing to generate the URL.
         """
         # Currently we don't need any abstraction for other platforms than GitHub, so we don't build such abstractions.
         match connection_mode:
@@ -937,7 +939,12 @@ class DryRunBackend(GitPlatformBackend):
 
     def __init__(self):
         super().__init__(GitCli())
-        self._repos: set[str] = set()
+        self._repos: dict[str, str] = {}
+        """
+        _repos maps from repository name to the owner of the upstream repository.
+        If a remote URL of a fork is requested with get_remote_url, _USER (the virtual current user) is
+        replaced by the owner of the upstream repository. This allows cloning the forked repository.
+        """
 
     def get_api_requests_left(self) -> Bound:
         return Bound.INFINITY
@@ -951,6 +958,26 @@ def does_repository_exist(self, owner: str, repo_name: str) -> bool:
             self.get_remote_url(owner, repo_name, GitConnectionMode.HTTPS)
         )
 
+    def get_remote_url(
+        self,
+        owner: str,
+        repo_name: str,
+        connection_mode: GitConnectionMode = GitConnectionMode.HTTPS,
+        token: str | None = None,
+    ) -> str:
+        if owner != self._USER:
+            return super().get_remote_url(owner, repo_name, connection_mode, token)
+        # redirect to the upstream repository
+        try:
+            upstream_owner = self._repos[repo_name]
+        except KeyError:
+            raise RepositoryNotFoundError(
+                f"Repository {owner}/{repo_name} appears to be a virtual fork but does not exist. Note that dry-run "
+                "forks are persistent only for the duration of the backend instance."
+            )
+
+        return super().get_remote_url(upstream_owner, repo_name, connection_mode, token)
+
     def push_to_repository(
         self, owner: str, repo_name: str, git_dir: Path, branch: str
     ):
@@ -967,7 +994,7 @@ def fork(self, owner: str, repo_name: str):
         logger.debug(
             f"Dry Run: Creating fork of {owner}/{repo_name} for user {self._USER}."
         )
-        self._repos.add(repo_name)
+        self._repos[repo_name] = owner
 
     def _sync_default_branch(self, upstream_owner: str, upstream_repo: str):
         logger.debug(

tests/test_git_utils.py

@@ -1044,25 +1044,6 @@ def test_git_cli_clone_fork_and_branch_non_existing_remote_existing_target_dir(c
         assert "trying to reset hard" in caplog.text
 
 
-def test_git_platform_backend_get_remote_url_https():
-    owner = "OWNER"
-    repo = "REPO"
-
-    url = GitPlatformBackend.get_remote_url(owner, repo, GitConnectionMode.HTTPS)
-
-    assert url == f"https://github.com/{owner}/{repo}.git"
-
-
-def test_git_platform_backend_get_remote_url_token():
-    owner = "OWNER"
-    repo = "REPO"
-    token = "TOKEN"
-
-    url = GitPlatformBackend.get_remote_url(owner, repo, GitConnectionMode.HTTPS, token)
-
-    assert url == f"https://{token}@github.com/{owner}/{repo}.git"
-
-
 def test_github_backend_from_token():
     token = "TOKEN"
 
@@ -1106,6 +1087,27 @@ def test_github_backend_does_repository_exist(does_exist: bool):
     github3_client.repository.assert_called_once_with("OWNER", "REPO")
 
 
+def test_github_backend_get_remote_url_https():
+    owner = "OWNER"
+    repo = "REPO"
+    backend = GitHubBackend(MagicMock(), MagicMock(), "")
+
+    url = backend.get_remote_url(owner, repo, GitConnectionMode.HTTPS)
+
+    assert url == f"https://github.com/{owner}/{repo}.git"
+
+
+def test_github_backend_get_remote_url_token():
+    owner = "OWNER"
+    repo = "REPO"
+    token = "TOKEN"
+    backend = GitHubBackend(MagicMock(), MagicMock(), "")
+
+    url = backend.get_remote_url(owner, repo, GitConnectionMode.HTTPS, token)
+
+    assert url == f"https://{token}@github.com/{owner}/{repo}.git"
+
+
 @mock.patch("time.sleep", return_value=None)
 @mock.patch(
     "conda_forge_tick.git_utils.GitHubBackend.user", new_callable=mock.PropertyMock
@@ -1609,6 +1611,48 @@ def test_dry_run_backend_does_repository_exist_other_repo():
     )
 
 
+@pytest.mark.parametrize("token", [None, "TOKEN"])
+def test_dry_run_backend_get_remote_url_non_fork(token: str | None):
+    backend = DryRunBackend()
+
+    url = backend.get_remote_url("OWNER", "REPO", GitConnectionMode.HTTPS, token)
+
+    if token is None:
+        assert url == "https://github.com/OWNER/REPO.git"
+    else:
+        assert url == "https://[email protected]/OWNER/REPO.git"
+
+
+@pytest.mark.parametrize("token", [None, "TOKEN"])
+def test_dry_run_backend_get_remote_url_non_existing_fork(token: str | None):
+    backend = DryRunBackend()
+
+    with pytest.raises(RepositoryNotFoundError, match="does not exist"):
+        backend.get_remote_url(backend.user, "REPO", GitConnectionMode.HTTPS, token)
+
+    backend.fork("UPSTREAM_OWNER", "REPO2")
+
+    with pytest.raises(RepositoryNotFoundError, match="does not exist"):
+        backend.get_remote_url(backend.user, "REPO", GitConnectionMode.HTTPS, token)
+
+
+@pytest.mark.parametrize("token", [None, "TOKEN"])
+def test_dry_run_backend_get_remote_url_existing_fork(token: str | None):
+    backend = DryRunBackend()
+
+    backend.fork("UPSTREAM_OWNER", "pytest-feedstock")
+
+    url = backend.get_remote_url(
+        backend.user, "pytest-feedstock", GitConnectionMode.HTTPS, token
+    )
+
+    # note that the URL does not indicate anymore that it is a fork
+    assert (
+        url
+        == f"https://{f'{token}@' if token else ''}github.com/UPSTREAM_OWNER/pytest-feedstock.git"
+    )
+
+
 def test_dry_run_backend_push_to_repository(caplog):
     caplog.set_level(logging.DEBUG)