Add migration endpoint for single repo by name (#5)

Author: Muscaw

gitea_github_sync/cli.py

@@ -3,7 +3,7 @@
 import click
 from rich import print
 
-from . import gitea, github, repository
+from . import config, gitea, github, repository
 
 
 @click.group()
@@ -45,3 +45,19 @@ def list_all_gitea_repositories(stats: bool) -> None:
     gt = gitea.get_gitea()
     repos = gt.get_repos()
     print_repositories(repos, stats)
+
+
+@cli.command()
+@click.argument("full_repo_name")
+def migrate_repo(full_repo_name: str) -> None:
+    conf = config.load_config()
+    gt = gitea.get_gitea()
+    gh = github.get_github()
+    github_repos = github.list_all_repositories(gh)
+    try:
+        repo = next((repo for repo in github_repos if repo.full_repo_name == full_repo_name))
+    except StopIteration:
+        print(f"[b red]Repository {full_repo_name} does not exist on Github[/]")
+        raise click.Abort()
+
+    gt.migrate_repo(repo=repo, github_token=conf.github_token)

gitea_github_sync/gitea.py

@@ -10,6 +10,14 @@
 from .repository import Repository, Visibility
 
 
+@dataclass(frozen=True)
+class GiteaMigrationError(ValueError):
+    full_repo_name: str
+
+    def __str__(self) -> str:
+        return f"Could not migrate {self.full_repo_name}"
+
+
 @dataclass(frozen=True)
 class Gitea:
     api_url: str
@@ -42,6 +50,25 @@ def get_repos(self) -> List[Repository]:
             for repo in repos
         ]
 
+    def migrate_repo(self, repo: Repository, github_token: str) -> None:
+        request_data = {
+            "auth_token": github_token,
+            "clone_addr": f"https://github.com/{repo.full_repo_name}",
+            "repo_name": repo.get_repo_name(),
+            "service": "github",
+            "mirror": True,
+            "private": repo.visibility == Visibility.PRIVATE,
+        }
+        res = requests.post(
+            f"{self.api_url}/repos/migrate",
+            headers=self._get_authorization_header(),
+            json=request_data,
+        )
+        try:
+            res.raise_for_status()
+        except requests.HTTPError as e:
+            raise GiteaMigrationError(repo.full_repo_name) from e
+
 
 def get_gitea(conf: Optional[config.Config] = None) -> Gitea:
     if conf is None:

tests/test_cli.py

@@ -1,7 +1,7 @@
 import textwrap
 from io import StringIO
 from typing import List
-from unittest.mock import MagicMock, patch
+from unittest.mock import MagicMock, PropertyMock, patch
 
 import pytest
 from click.testing import CliRunner
@@ -71,6 +71,62 @@ def test_list_all_gitea_repositories(
     mock_print_repositories.assert_called_once_with(repositories_fixture, expected_stat)
 
 
+@patch("gitea_github_sync.cli.config.load_config", autospec=True)
+@patch("gitea_github_sync.cli.github.list_all_repositories", autospec=True)
+@patch("gitea_github_sync.cli.github.get_github", autospec=True)
+@patch("gitea_github_sync.cli.gitea.get_gitea", autospec=True)
+def test_migrate_repo(
+    mock_get_gitea: MagicMock,
+    mock_get_github: MagicMock,
+    mock_list_all_repositories: MagicMock,
+    mock_load_config: MagicMock,
+    repositories_fixture: List[Repository],
+) -> None:
+    expected_repo = Repository("Muscaw/gitea-github-sync", Visibility.PRIVATE)
+    expected_github_token = "some-github-token"
+
+    type(mock_load_config.return_value).github_token = PropertyMock(
+        return_value=expected_github_token
+    )
+    mock_list_all_repositories.return_value = repositories_fixture + [expected_repo]
+
+    runner = CliRunner()
+    command = ["migrate-repo", "Muscaw/gitea-github-sync"]
+    result = runner.invoke(cli, command)
+
+    assert result.exit_code == 0
+    mock_list_all_repositories.assert_called_once_with(mock_get_github.return_value)
+    mock_get_gitea.return_value.migrate_repo.assert_called_once_with(
+        repo=expected_repo, github_token=expected_github_token
+    )
+
+
+@patch("gitea_github_sync.cli.config.load_config", autospec=True)
+@patch("gitea_github_sync.cli.github.list_all_repositories", autospec=True)
+@patch("gitea_github_sync.cli.github.get_github", autospec=True)
+@patch("gitea_github_sync.cli.gitea.get_gitea", autospec=True)
+def test_migrate_repo_no_match(
+    mock_get_gitea: MagicMock,
+    mock_get_github: MagicMock,
+    mock_list_all_repositories: MagicMock,
+    mock_load_config: MagicMock,
+    repositories_fixture: List[Repository],
+) -> None:
+    mock_list_all_repositories.return_value = repositories_fixture
+    repo_name = "Muscaw/gitea-github-sync"
+
+    runner = CliRunner()
+    command = ["migrate-repo", repo_name]
+    result = runner.invoke(cli, command)
+
+    assert result.exit_code != 0
+    assert "Aborted!" in result.stdout
+    assert f"Repository {repo_name} does not exist on Github" in result.stdout
+    mock_list_all_repositories.assert_called_once_with(mock_get_github.return_value)
+    mock_get_gitea.return_value.migrate_repo.assert_not_called()
+    mock_load_config.assert_called_once()
+
+
 @patch("sys.stdout", new_callable=StringIO)
 def test_print_repositories_without_stats(
     stdout: StringIO,

tests/test_gitea.py

@@ -5,7 +5,7 @@
 from responses import matchers
 
 from gitea_github_sync.config import Config
-from gitea_github_sync.gitea import Gitea, get_gitea
+from gitea_github_sync.gitea import Gitea, GiteaMigrationError, get_gitea
 from gitea_github_sync.repository import Repository, Visibility
 
 GITEA_BASE_API_URL = "https://gitea.yourinstance.com/api/v1"
@@ -25,7 +25,7 @@ def gitea_fixture(conf_fixture: Config) -> Gitea:
 
 
 @responses.activate
-def test_get_repos(gitea_fixture: Gitea) -> None:
+def test_gitea_get_repos(gitea_fixture: Gitea) -> None:
 
     json = [
         {"full_name": "some-team/a-repo", "private": True},
@@ -47,7 +47,7 @@ def test_get_repos(gitea_fixture: Gitea) -> None:
 
 
 @responses.activate
-def test_get_repos_multiple_pages(gitea_fixture: Gitea) -> None:
+def test_gitea_get_repos_multiple_pages(gitea_fixture: Gitea) -> None:
 
     json_1 = [
         {"full_name": "some-team/a-repo", "private": True},
@@ -110,6 +110,62 @@ def test_get_repos_multiple_pages(gitea_fixture: Gitea) -> None:
     assert expected_repos == result
 
 
+@responses.activate
+@pytest.mark.parametrize("is_private", [True, False])
+def test_gitea_migrate_repo(gitea_fixture: Gitea, is_private: bool) -> None:
+    gh_token = "some-github-token"
+    expected_data = {
+        "auth_token": gh_token,
+        "clone_addr": "https://github.com/Muscaw/gitea-github-sync",
+        "repo_name": "gitea-github-sync",
+        "service": "github",
+        "mirror": True,
+        "private": is_private,
+    }
+    repo = Repository(
+        full_repo_name="Muscaw/gitea-github-sync",
+        visibility=Visibility.PRIVATE if is_private else Visibility.PUBLIC,
+    )
+    responses.post(
+        f"{GITEA_BASE_API_URL}/repos/migrate",
+        match=[
+            matchers.header_matcher({"Authorization": f"token {GITEA_TOKEN}"}),
+            matchers.json_params_matcher(expected_data),
+        ],
+    )
+
+    gitea_fixture.migrate_repo(repo, gh_token)
+
+
+@responses.activate
+@pytest.mark.parametrize("is_private", [True, False])
+def test_gitea_migrate_repo_failure_to_migrate(gitea_fixture: Gitea, is_private: bool) -> None:
+    gh_token = "some-github-token"
+    expected_data = {
+        "auth_token": gh_token,
+        "clone_addr": "https://github.com/Muscaw/gitea-github-sync",
+        "repo_name": "gitea-github-sync",
+        "service": "github",
+        "mirror": True,
+        "private": is_private,
+    }
+    repo = Repository(
+        full_repo_name="Muscaw/gitea-github-sync",
+        visibility=Visibility.PRIVATE if is_private else Visibility.PUBLIC,
+    )
+    responses.post(
+        f"{GITEA_BASE_API_URL}/repos/migrate",
+        match=[
+            matchers.header_matcher({"Authorization": f"token {GITEA_TOKEN}"}),
+            matchers.json_params_matcher(expected_data),
+        ],
+        status=409,
+    )
+
+    with pytest.raises(GiteaMigrationError):
+        gitea_fixture.migrate_repo(repo, gh_token)
+
+
 def test_gitea(gitea_fixture: Gitea, conf_fixture: Config) -> None:
     gt = get_gitea(conf_fixture)
 
@@ -125,3 +181,8 @@ def test_gitea_default_value(
 
     assert gt == gitea_fixture
     mock_load_config.assert_called_once()
+
+
+def test_gitea_migration_error() -> None:
+    error = GiteaMigrationError("Muscaw/gitea-github-sync")
+    assert str(error) == "Could not migrate Muscaw/gitea-github-sync"