hide tokens in Git Backends automatically, push to repository

Author: ytausch

conda_forge_tick/git_utils.py

@@ -548,21 +548,38 @@ def get_remote_url(
         owner: str,
         repo_name: str,
         connection_mode: GitConnectionMode = GitConnectionMode.HTTPS,
+        token: str | None = None,
     ) -> str:
         """
         Get the URL of a remote repository.
         :param owner: The owner of the repository.
         :param repo_name: The name of the repository.
         :param connection_mode: The connection mode to use.
+        :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.
         """
         # Currently we don't need any abstraction for other platforms than GitHub, so we don't build such abstractions.
         match connection_mode:
             case GitConnectionMode.HTTPS:
-                return f"https://github.com/{owner}/{repo_name}.git"
+                return f"https://{f'{token}@' if token else ''}github.com/{owner}/{repo_name}.git"
             case _:
                 raise ValueError(f"Unsupported connection mode: {connection_mode}")
 
+    @abstractmethod
+    def push_to_repository(
+        self, owner: str, repo_name: str, git_dir: Path, branch: str
+    ):
+        """
+        Push changes to a repository.
+        :param owner: The owner of the repository.
+        :param repo_name: The name of the repository.
+        :param git_dir: The directory of the git repository.
+        :param branch: The branch to push to.
+        :raises GitPlatformError: If the push fails.
+        """
+        pass
+
     @abstractmethod
     def fork(self, owner: str, repo_name: str):
         """
@@ -729,7 +746,7 @@ def __init__(
         self,
         github3_client: github3.GitHub,
         pygithub_client: github.Github,
-        token_to_hide: str | None = None,
+        token: str,
     ):
         """
         Create a new GitHubBackend.
@@ -739,12 +756,14 @@ def __init__(
 
         :param github3_client: The github3 client to use for interacting with the GitHub API.
         :param pygithub_client: The PyGithub client to use for interacting with the GitHub API.
-        :param token_to_hide: A token to hide in the CLI output. If None, no tokens are hidden.
+        :param token: The token that will be hidden in CLI outputs and used for writing to git repositories. Note that
+        you need to authenticate github3 and PyGithub yourself. Use the `from_token` class method to create an instance
+        that has all necessary clients set up.
         """
         cli = GitCli()
-        if token_to_hide:
-            cli.add_hidden_token(token_to_hide)
+        cli.add_hidden_token(token)
         super().__init__(cli)
+        self.__token = token
         self.github3_client = github3_client
         self._github3_session = _Github3SessionWrapper(self.github3_client.session)
         self.github3_client.session = self._github3_session
@@ -756,13 +775,22 @@ def from_token(cls, token: str):
         return cls(
             github3.login(token=token),
             github.Github(auth=github.Auth.Token(token), per_page=cls._GITHUB_PER_PAGE),
-            token_to_hide=token,
+            token=token,
         )
 
     def does_repository_exist(self, owner: str, repo_name: str) -> bool:
         repo = self.github3_client.repository(owner, repo_name)
         return repo is not None
 
+    def push_to_repository(
+        self, owner: str, repo_name: str, git_dir: Path, branch: str
+    ):
+        # we need an authenticated URL with write access
+        remote_url = self.get_remote_url(
+            owner, repo_name, GitConnectionMode.HTTPS, self.__token
+        )
+        self.cli.push_to_url(git_dir, remote_url, branch)
+
     @lock_git_operation()
     def fork(self, owner: str, repo_name: str):
         if self.does_repository_exist(self.user, repo_name):
@@ -923,6 +951,13 @@ def does_repository_exist(self, owner: str, repo_name: str) -> bool:
             self.get_remote_url(owner, repo_name, GitConnectionMode.HTTPS)
         )
 
+    def push_to_repository(
+        self, owner: str, repo_name: str, git_dir: Path, branch: str
+    ):
+        logger.debug(
+            f"Dry Run: Pushing changes from {git_dir} to {owner}/{repo_name} on branch {branch}."
+        )
+
     @lock_git_operation()
     def fork(self, owner: str, repo_name: str):
         if repo_name in self._repos:

tests/test_git_utils.py

@@ -1053,6 +1053,16 @@ def test_git_platform_backend_get_remote_url_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"
 
@@ -1088,7 +1098,7 @@ def test_github_backend_token_to_hide(caplog, capfd, from_token: bool):
 def test_github_backend_does_repository_exist(does_exist: bool):
     github3_client = MagicMock()
 
-    backend = GitHubBackend(github3_client, MagicMock())
+    backend = GitHubBackend(github3_client, MagicMock(), "")
 
     github3_client.repository.return_value = MagicMock() if does_exist else None
 
@@ -1110,7 +1120,7 @@ def test_github_backend_fork_not_exists_repo_found(
     repository = MagicMock()
     github3_client.repository.return_value = repository
 
-    backend = GitHubBackend(github3_client, MagicMock())
+    backend = GitHubBackend(github3_client, MagicMock(), "")
     user_mock.return_value = "USER"
     backend.fork("UPSTREAM-OWNER", "REPO")
 
@@ -1120,6 +1130,21 @@ def test_github_backend_fork_not_exists_repo_found(
     sleep_mock.assert_called_once_with(5)
 
 
+@mock.patch("conda_forge_tick.git_utils.GitCli.push_to_url")
+def test_github_backend_push_to_repository(push_to_url_mock: MagicMock):
+    backend = GitHubBackend.from_token("THIS_IS_THE_TOKEN")
+
+    git_dir = Path("GIT_DIR")
+
+    backend.push_to_repository("OWNER", "REPO", git_dir, "BRANCH_NAME")
+
+    push_to_url_mock.assert_called_once_with(
+        git_dir,
+        "https://[email protected]/OWNER/REPO.git",
+        "BRANCH_NAME",
+    )
+
+
 @pytest.mark.parametrize("branch_already_synced", [True, False])
 @mock.patch("time.sleep", return_value=None)
 @mock.patch(
@@ -1158,7 +1183,7 @@ def get_repo(full_name: str):
         upstream_repo.default_branch = "UPSTREAM_BRANCH_NAME"
         fork_repo.default_branch = "FORK_BRANCH_NAME"
 
-    backend = GitHubBackend(MagicMock(), pygithub_client)
+    backend = GitHubBackend(MagicMock(), pygithub_client, "")
     backend.fork("UPSTREAM-OWNER", "REPO")
 
     if not branch_already_synced:
@@ -1181,7 +1206,7 @@ def test_github_backend_remote_does_not_exist(
     github3_client = MagicMock()
     github3_client.repository.return_value = None
 
-    backend = GitHubBackend(github3_client, MagicMock())
+    backend = GitHubBackend(github3_client, MagicMock(), "")
 
     user_mock.return_value = "USER"
 
@@ -1198,7 +1223,7 @@ def test_github_backend_user():
     user.login = "USER"
     pygithub_client.get_user.return_value = user
 
-    backend = GitHubBackend(MagicMock(), pygithub_client)
+    backend = GitHubBackend(MagicMock(), pygithub_client, "")
 
     for _ in range(4):
         # cached property
@@ -1213,7 +1238,7 @@ def test_github_backend_get_api_requests_left_github_exception(caplog):
         "API Error"
     )
 
-    backend = GitHubBackend(github3_client, MagicMock())
+    backend = GitHubBackend(github3_client, MagicMock(), "")
 
     assert backend.get_api_requests_left() is None
     assert "API error while fetching" in caplog.text
@@ -1225,7 +1250,7 @@ def test_github_backend_get_api_requests_left_unexpected_response_schema(caplog)
     github3_client = MagicMock()
     github3_client.rate_limit.return_value = {"some": "gibberish data"}
 
-    backend = GitHubBackend(github3_client, MagicMock())
+    backend = GitHubBackend(github3_client, MagicMock(), "")
 
     assert backend.get_api_requests_left() is None
     assert "API Error while parsing"
@@ -1237,7 +1262,7 @@ def test_github_backend_get_api_requests_left_nonzero():
     github3_client = MagicMock()
     github3_client.rate_limit.return_value = {"resources": {"core": {"remaining": 5}}}
 
-    backend = GitHubBackend(github3_client, MagicMock())
+    backend = GitHubBackend(github3_client, MagicMock(), "")
 
     assert backend.get_api_requests_left() == 5
 
@@ -1249,7 +1274,7 @@ def test_github_backend_get_api_requests_left_zero_invalid_reset_time(caplog):
 
     github3_client.rate_limit.return_value = {"resources": {"core": {"remaining": 0}}}
 
-    backend = GitHubBackend(github3_client, MagicMock())
+    backend = GitHubBackend(github3_client, MagicMock(), "")
 
     assert backend.get_api_requests_left() == 0
 
@@ -1269,7 +1294,7 @@ def test_github_backend_get_api_requests_left_zero_valid_reset_time(caplog):
         "resources": {"core": {"remaining": 0, "reset": reset_timestamp}}
     }
 
-    backend = GitHubBackend(github3_client, MagicMock())
+    backend = GitHubBackend(github3_client, MagicMock(), "")
 
     assert backend.get_api_requests_left() == 0
 
@@ -1308,7 +1333,7 @@ def request_side_effect(method, _url, **_kwargs):
     pygithub_mock = MagicMock()
     pygithub_mock.get_user.return_value.login = "CURRENT_USER"
 
-    backend = GitHubBackend(github3.login(token="TOKEN"), pygithub_mock)
+    backend = GitHubBackend(github3.login(token="TOKEN"), pygithub_mock, "")
 
     pr_data = backend.create_pull_request(
         "conda-forge",
@@ -1394,7 +1419,7 @@ def request_side_effect(method, url, **_kwargs):
 
     request_mock.side_effect = request_side_effect
 
-    backend = GitHubBackend(github3.login(token="TOKEN"), MagicMock())
+    backend = GitHubBackend(github3.login(token="TOKEN"), MagicMock(), "")
 
     backend.comment_on_pull_request(
         "conda-forge",
@@ -1426,7 +1451,7 @@ def request_side_effect(method, url, **_kwargs):
 
     request_mock.side_effect = request_side_effect
 
-    backend = GitHubBackend(github3.login(token="TOKEN"), MagicMock())
+    backend = GitHubBackend(github3.login(token="TOKEN"), MagicMock(), "")
 
     with pytest.raises(RepositoryNotFoundError):
         backend.comment_on_pull_request(
@@ -1463,7 +1488,7 @@ def request_side_effect(method, url, **_kwargs):
         assert False, f"Unexpected endpoint: {method} {url}"
 
     request_mock.side_effect = request_side_effect
-    backend = GitHubBackend(github3.login(token="TOKEN"), MagicMock())
+    backend = GitHubBackend(github3.login(token="TOKEN"), MagicMock(), "")
 
     with pytest.raises(
         GitPlatformError,
@@ -1516,7 +1541,7 @@ def request_side_effect(method, url, **_kwargs):
 
     request_mock.side_effect = request_side_effect
 
-    backend = GitHubBackend(github3.login(token="TOKEN"), MagicMock())
+    backend = GitHubBackend(github3.login(token="TOKEN"), MagicMock(), "")
 
     with pytest.raises(GitPlatformError, match="Could not comment on pull request"):
         backend.comment_on_pull_request(
@@ -1528,7 +1553,7 @@ def request_side_effect(method, url, **_kwargs):
 
 
 @pytest.mark.parametrize(
-    "backend", [GitHubBackend(MagicMock(), MagicMock()), DryRunBackend()]
+    "backend", [GitHubBackend(MagicMock(), MagicMock(), ""), DryRunBackend()]
 )
 @mock.patch(
     "conda_forge_tick.git_utils.GitHubBackend.user", new_callable=mock.PropertyMock
@@ -1547,7 +1572,7 @@ def test_git_platform_backend_clone_fork_and_branch(
 
     user_mock.return_value = "USER"
 
-    backend = GitHubBackend(MagicMock(), MagicMock())
+    backend = GitHubBackend(MagicMock(), MagicMock(), "")
     backend.clone_fork_and_branch(
         upstream_owner, repo_name, target_dir, new_branch, base_branch
     )
@@ -1584,6 +1609,21 @@ def test_dry_run_backend_does_repository_exist_other_repo():
     )
 
 
+def test_dry_run_backend_push_to_repository(caplog):
+    caplog.set_level(logging.DEBUG)
+
+    backend = DryRunBackend()
+
+    git_dir = Path("GIT_DIR")
+
+    backend.push_to_repository("OWNER", "REPO", git_dir, "BRANCH_NAME")
+
+    assert (
+        "Dry Run: Pushing changes from GIT_DIR to OWNER/REPO on branch BRANCH_NAME"
+        in caplog.text
+    )
+
+
 def test_dry_run_backend_fork(caplog):
     caplog.set_level(logging.DEBUG)