diff --git a/docs/superpowers/plans/2026-06-28-hardening-cli-and-sync.md b/docs/superpowers/plans/2026-06-28-hardening-cli-and-sync.md new file mode 100644 index 0000000..f9e7bc6 --- /dev/null +++ b/docs/superpowers/plans/2026-06-28-hardening-cli-and-sync.md @@ -0,0 +1,131 @@ +# Hardening CLI And Sync Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Make Gitea mirror fail reliably on API/git/config/path errors and expose a proper package CLI entrypoint. + +**Architecture:** Keep the existing module layout, but move executable orchestration into an importable `src.cli` module. Preserve `gitea-mirror.py` as a compatibility wrapper while making `process_repo` return success/failure for aggregate exit handling. + +**Tech Stack:** Python 3.12, requests, pydantic v2, pytest, ruff, uv. + +--- + +### Task 1: API Errors Are Fatal + +**Files:** +- Modify: `src/gitea_api.py` +- Test: `tests/test_gitea_api.py` + +- [ ] **Step 1: Write failing tests** + +```python +def test_get_repositories_raises_on_http_error(monkeypatch): + api = GiteaApi(endpoint="https://example.com", api_token="token") + session = FakeSession([FakeResponse(status_code=500, data=[])]) + monkeypatch.setattr(requests, "session", lambda: session) + + with pytest.raises(GiteaApiError): + api.get_repositories() +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `uv run pytest tests/test_gitea_api.py -v` +Expected: FAIL because `GiteaApiError` does not exist or is not raised. + +- [ ] **Step 3: Implement minimal code** + +Add `GiteaApiError`, request timeout, network exception handling, JSON/schema exception handling, and raise instead of returning `[]`. + +- [ ] **Step 4: Run test to verify it passes** + +Run: `uv run pytest tests/test_gitea_api.py -v` +Expected: PASS. + +### Task 2: Git Failures Affect Exit Code + +**Files:** +- Create: `src/cli.py` +- Modify: `gitea-mirror.py` +- Test: `tests/test_mirror_process.py`, `tests/test_cli.py` + +- [ ] **Step 1: Write failing tests** + +```python +def test_process_repo_returns_false_when_clone_fails(monkeypatch, tmp_path): + monkeypatch.setattr(cli, "git_clone", lambda **kwargs: False) + assert not cli.process_repo(config=make_config(tmp_path, FetchConfig(ssh_key="/tmp/key")), repo=make_repo()) +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `uv run pytest tests/test_mirror_process.py tests/test_cli.py -v` +Expected: FAIL because `process_repo` currently returns `None` and no importable CLI module exists. + +- [ ] **Step 3: Implement minimal code** + +Move orchestration to `src.cli`, return booleans from `process_repo`, aggregate failures in `run`, and make `main` return non-zero for failures. + +- [ ] **Step 4: Run test to verify it passes** + +Run: `uv run pytest tests/test_mirror_process.py tests/test_cli.py -v` +Expected: PASS. + +### Task 3: Validate Repository Paths And Config + +**Files:** +- Modify: `src/models.py` +- Modify: `src/repository_name.py` +- Test: `tests/test_config.py`, `tests/test_repository_name.py` + +- [ ] **Step 1: Write failing tests** + +```python +def test_output_path_rejects_parent_traversal(tmp_path): + repo = make_repo() + with pytest.raises(ValueError): + get_repository_path(str(tmp_path), "../{name}", repo) +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `uv run pytest tests/test_config.py tests/test_repository_name.py -v` +Expected: FAIL because validation is missing. + +- [ ] **Step 3: Implement minimal code** + +Add repository format validation in `Config`, log level validation in `LoggingConfig`, and a safe output path helper that rejects absolute paths and paths escaping `out_dir`. + +- [ ] **Step 4: Run test to verify it passes** + +Run: `uv run pytest tests/test_config.py tests/test_repository_name.py -v` +Expected: PASS. + +### Task 4: Package CLI Entrypoint + +**Files:** +- Modify: `pyproject.toml` +- Modify: `gitea-mirror.py` +- Test: `tests/test_cli.py` + +- [ ] **Step 1: Write failing test** + +```python +def test_project_defines_console_script(): + data = tomllib.loads(Path("pyproject.toml").read_text()) + assert data["project"]["scripts"]["gitea-mirror"] == "src.cli:main" +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `uv run pytest tests/test_cli.py -v` +Expected: FAIL because `[project.scripts]` is absent. + +- [ ] **Step 3: Implement minimal code** + +Add `[project.scripts] gitea-mirror = "src.cli:main"` and make `gitea-mirror.py` delegate to `src.cli.main`. + +- [ ] **Step 4: Run full verification** + +Run: `make check` +Expected: PASS. diff --git a/gitea-mirror.py b/gitea-mirror.py index 9b72a3b..3ee6d95 100644 --- a/gitea-mirror.py +++ b/gitea-mirror.py @@ -1,102 +1,6 @@ -import logging -import os.path import sys -from os import makedirs - -from src.config import Config, read_toml_config -from src.git import git_clone, git_pull, http_git_env, ssh_git_env -from src.gitea_api import GiteaApi -from src.log_fields import ( - OPERATION, - PATH, - REPOSITORY, - REPOSITORY_COUNT, - REPOSITORY_URL, -) -from src.logging_config import setup_logging -from src.models import GiteaRepository -from src.repository_name import get_repository_name, is_valid_repository_names - -BASE_PATH = "out" -FORMAT = "{owner}/{name}" -logger = logging.getLogger(__name__) - - -def get_fetch_params(config: Config, repo: GiteaRepository): - if config.fetch.ssh_key_path: - return repo.ssh_url, ssh_git_env(config.fetch.ssh_key_path) - return repo.clone_url, http_git_env( - user=config.fetch.user, - token=config.fetch.token, - ) - - -def process_repo(config: Config, repo: GiteaRepository): - path = get_repository_name(name_format=config.repository_format, r=repo) - out_path = os.path.join(config.out_dir, path) - repository_url, git_env = get_fetch_params(config=config, repo=repo) - log_context = { - REPOSITORY: path, - PATH: out_path, - REPOSITORY_URL: repository_url, - } - makedirs(out_path, exist_ok=True) - if os.path.exists(os.path.join(out_path, ".git")): - logger.info( - "repository pull started", - extra=log_context | {OPERATION: "pull"}, - ) - git_pull(out_path, env=git_env) - return - logger.info( - "repository clone started", - extra=log_context | {OPERATION: "clone"}, - ) - git_clone(repository_url=repository_url, repository=out_path, env=git_env) - - -def main(): - if len(sys.argv) < 2: - logging.basicConfig( - level=logging.ERROR, - format="%(levelname)s %(message)s", - ) - logger.error("Usage: python gitea-mirror.py CONFIG_PATH") - sys.exit(1) - try: - config = read_toml_config(sys.argv[1]) - except RuntimeError as err: - logging.basicConfig( - level=logging.ERROR, - format="%(levelname)s %(message)s", - ) - logger.error("invalid config: %s", err) - sys.exit(1) - setup_logging(config.logging) - - api = GiteaApi( - endpoint=config.endpoint, - api_token=config.api_token, - ) - repos = api.get_repositories() - logger.info( - "repositories ready", - extra={OPERATION: "mirror", REPOSITORY_COUNT: len(repos)}, - ) - - if not is_valid_repository_names( - name_format=config.repository_format, - repos=repos, - ): - logger.error( - "repository name format creates duplicates", - extra={OPERATION: "validate_repository_names"}, - ) - sys.exit(1) - - for repo in repos: - process_repo(config=config, repo=repo) +from src.cli import main if __name__ == "__main__": - main() + sys.exit(main()) diff --git a/pyproject.toml b/pyproject.toml index d1f23da..88d3c37 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -9,6 +9,9 @@ dependencies = [ "requests>=2.32", ] +[project.scripts] +gitea-mirror = "src.cli:main" + [dependency-groups] dev = [ "pre-commit>=4.0", diff --git a/src/cli.py b/src/cli.py new file mode 100644 index 0000000..af9ccfb --- /dev/null +++ b/src/cli.py @@ -0,0 +1,119 @@ +import logging +import os.path +import sys +from os import makedirs + +from .config import Config, read_toml_config +from .git import git_clone, git_pull, http_git_env, ssh_git_env +from .gitea_api import GiteaApi, GiteaApiError +from .log_fields import ( + OPERATION, + PATH, + REPOSITORY, + REPOSITORY_COUNT, + REPOSITORY_URL, +) +from .logging_config import setup_logging +from .models import GiteaRepository +from .repository_name import get_repository_path, is_valid_repository_names + +logger = logging.getLogger(__name__) + + +def get_fetch_params(config: Config, repo: GiteaRepository): + if config.fetch.ssh_key_path: + return repo.ssh_url, ssh_git_env(config.fetch.ssh_key_path) + return repo.clone_url, http_git_env( + user=config.fetch.user, + token=config.fetch.token, + ) + + +def process_repo(config: Config, repo: GiteaRepository) -> bool: + out_path = get_repository_path( + out_dir=config.out_dir, + name_format=config.repository_format, + repo=repo, + ) + path = os.path.relpath(out_path, config.out_dir) + repository_url, git_env = get_fetch_params(config=config, repo=repo) + log_context = { + REPOSITORY: path, + PATH: out_path, + REPOSITORY_URL: repository_url, + } + makedirs(out_path, exist_ok=True) + if os.path.exists(os.path.join(out_path, ".git")): + logger.info( + "repository pull started", + extra=log_context | {OPERATION: "pull"}, + ) + return git_pull(out_path, env=git_env) + logger.info( + "repository clone started", + extra=log_context | {OPERATION: "clone"}, + ) + return git_clone( + repository_url=repository_url, + repository=out_path, + env=git_env, + ) + + +def run(config: Config) -> int: + api = GiteaApi( + endpoint=config.endpoint, + api_token=config.api_token, + ) + try: + repos = api.get_repositories() + except GiteaApiError as err: + logger.error( + "failed to fetch repositories", + extra={OPERATION: "mirror"}, + ) + logger.debug("gitea api error", exc_info=err) + return 1 + + logger.info( + "repositories ready", + extra={OPERATION: "mirror", REPOSITORY_COUNT: len(repos)}, + ) + + if not is_valid_repository_names( + name_format=config.repository_format, + repos=repos, + ): + logger.error( + "repository name format creates duplicates", + extra={OPERATION: "validate_repository_names"}, + ) + return 1 + + has_failure = False + for repo in repos: + if not process_repo(config=config, repo=repo): + has_failure = True + return 1 if has_failure else 0 + + +def main(argv: list[str] | None = None) -> int: + args = sys.argv[1:] if argv is None else argv + if len(args) < 1: + logging.basicConfig( + level=logging.ERROR, + format="%(levelname)s %(message)s", + ) + logger.error("Usage: gitea-mirror CONFIG_PATH") + return 1 + try: + config = read_toml_config(args[0]) + except RuntimeError as err: + logging.basicConfig( + level=logging.ERROR, + format="%(levelname)s %(message)s", + ) + logger.error("invalid config: %s", err) + return 1 + setup_logging(config.logging) + return run(config) diff --git a/src/config.py b/src/config.py index cfdba0c..03c2592 100644 --- a/src/config.py +++ b/src/config.py @@ -17,6 +17,8 @@ ssh_key = "/home/user/.ssh/id_rsa" import os import tomllib +from pydantic import ValidationError + from .models import Config, FetchConfig, LoggingConfig MAIN_SECTION = "main" @@ -36,14 +38,18 @@ def read_toml_config(path: str) -> Config: out_dir = main["out_dir"] fetch = FetchConfig(**data["fetch"]) logging_config = LoggingConfig(**data.get("logging", {})) - except (KeyError, ValueError, tomllib.TOMLDecodeError) as err: + return Config( + repository_format=repository_format, + endpoint=endpoint, + api_token=api_token, + out_dir=out_dir, + fetch=fetch, + logging=logging_config, + ) + except ( + KeyError, + ValueError, + ValidationError, + tomllib.TOMLDecodeError, + ) as err: raise RuntimeError(f"Invalid TOML config: {err}") from err - - return Config( - repository_format=repository_format, - endpoint=endpoint, - api_token=api_token, - out_dir=out_dir, - fetch=fetch, - logging=logging_config, - ) diff --git a/src/gitea_api.py b/src/gitea_api.py index f85fd9d..9475179 100644 --- a/src/gitea_api.py +++ b/src/gitea_api.py @@ -1,14 +1,20 @@ import logging +from json import JSONDecodeError from urllib.parse import urljoin import requests -from pydantic import TypeAdapter +from pydantic import TypeAdapter, ValidationError from .log_fields import OPERATION, PAGE, REPOSITORY_COUNT, STATUS_CODE from .models import GiteaRepository REPOSITORIES_ADAPTER = TypeAdapter(list[GiteaRepository]) logger = logging.getLogger(__name__) +REQUEST_TIMEOUT = (5, 30) + + +class GiteaApiError(RuntimeError): + pass class GiteaApi: @@ -25,13 +31,19 @@ class GiteaApi: all_repos = {} # hack for unique repositories in result page_id = 1 while True: - r = session.get( - urljoin( - self._endpoint, - "/api/v1/user/repos", - ), - params={"limit": page_size, "page": page_id}, - ) + try: + r = session.get( + urljoin( + self._endpoint, + "/api/v1/user/repos", + ), + params={"limit": page_size, "page": page_id}, + timeout=REQUEST_TIMEOUT, + ) + except requests.RequestException as err: + raise GiteaApiError( + f"request failed while fetching page {page_id}: {err}" + ) from err if r.status_code != 200: logger.error( "failed to fetch repositories", @@ -41,13 +53,25 @@ class GiteaApi: PAGE: page_id, }, ) - return [] - repos_data = r.json() + raise GiteaApiError( + f"failed to fetch repositories: status {r.status_code}" + ) + try: + repos_data = r.json() + except (JSONDecodeError, ValueError) as err: + raise GiteaApiError( + f"invalid JSON while fetching page {page_id}: {err}" + ) from err if not repos_data: break else: page_id += 1 - cur_repos = REPOSITORIES_ADAPTER.validate_python(repos_data) + try: + cur_repos = REPOSITORIES_ADAPTER.validate_python(repos_data) + except ValidationError as err: + raise GiteaApiError( + f"invalid repository payload on page {page_id - 1}: {err}" + ) from err for repo in cur_repos: all_repos[repo.repo_id] = repo logger.info( diff --git a/src/models.py b/src/models.py index 170add8..5e1a622 100644 --- a/src/models.py +++ b/src/models.py @@ -1,7 +1,8 @@ import datetime +import logging from typing import Literal -from pydantic import BaseModel, Field, HttpUrl, model_validator +from pydantic import BaseModel, Field, HttpUrl, field_validator, model_validator class FetchConfig(BaseModel): @@ -29,6 +30,20 @@ class Config(BaseModel): fetch: FetchConfig logging: "LoggingConfig" = Field(default_factory=lambda: LoggingConfig()) + @field_validator("repository_format") + @classmethod + def validate_repository_format(cls, value: str) -> str: + try: + value.format( + name="repo", + repository_id=1, + owner="owner", + owner_id=1, + ) + except KeyError: + raise ValueError("repository format contains unknown fields") + return value + class LoggingOutputConfig(BaseModel): type: Literal["console", "file"] = "console" @@ -51,6 +66,13 @@ class LoggingConfig(BaseModel): default_factory=lambda: [LoggingOutputConfig()] ) + @field_validator("level") + @classmethod + def validate_level(cls, value: str) -> str: + if not isinstance(logging.getLevelName(value), int): + raise ValueError(f"unknown logging level: {value}") + return value + class GiteaUser(BaseModel): user_id: int = Field(alias="id") diff --git a/src/repository_name.py b/src/repository_name.py index d9c46b2..97db467 100644 --- a/src/repository_name.py +++ b/src/repository_name.py @@ -1,4 +1,5 @@ import datetime +import os from typing import List from .models import GiteaRepository, GiteaUser @@ -36,6 +37,22 @@ def get_repository_name(name_format: str, r: GiteaRepository) -> str: ) +def get_repository_path( + out_dir: str, + name_format: str, + repo: GiteaRepository, +) -> str: + name = get_repository_name(name_format, repo) + if os.path.isabs(name): + raise ValueError("repository path must be relative") + + out_dir_path = os.path.abspath(out_dir) + repo_path = os.path.abspath(os.path.join(out_dir_path, name)) + if os.path.commonpath([out_dir_path, repo_path]) != out_dir_path: + raise ValueError("repository path must stay inside out_dir") + return repo_path + + def is_valid_repository_names(name_format: str, repos: List[GiteaRepository]): names = set(get_repository_name(name_format, r) for r in repos) return len(names) == len(repos) # all names must be unique diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..f478f2b --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,111 @@ +import datetime +import tomllib +from pathlib import Path + +from src import cli +from src.config import Config, FetchConfig +from src.gitea_api import GiteaApiError +from src.models import GiteaRepository, GiteaUser + + +def make_repo(repo_id=42, name="repo") -> GiteaRepository: + return GiteaRepository( + ssh_url="ssh://git@example.com/alex/repo.git", + clone_url="https://example.com/alex/repo.git", + name=name, + id=repo_id, + updated_at=datetime.datetime.now(), + owner=GiteaUser( + id=23, + login="alex", + email="alex@example.com", + ), + ) + + +def make_config(tmp_path, fetch: FetchConfig | None = None) -> Config: + return Config( + api_token="api-token", + endpoint="https://example.com", + repository_format="{owner}/{name}", + out_dir=str(tmp_path), + fetch=fetch or FetchConfig(ssh_key="/tmp/test_key"), + ) + + +def test_project_defines_console_script(): + data = tomllib.loads(Path("pyproject.toml").read_text()) + + assert data["project"]["scripts"]["gitea-mirror"] == "src.cli:main" + + +def test_run_returns_failure_when_api_fetch_fails(monkeypatch, tmp_path): + class FakeApi: + def __init__(self, endpoint, api_token): + pass + + def get_repositories(self): + raise GiteaApiError("request failed") + + monkeypatch.setattr(cli, "GiteaApi", FakeApi) + + assert cli.run(make_config(tmp_path)) == 1 + + +def test_run_returns_failure_when_any_repo_fails(monkeypatch, tmp_path): + class FakeApi: + def __init__(self, endpoint, api_token): + pass + + def get_repositories(self): + return [ + make_repo(repo_id=1, name="repo-1"), + make_repo(repo_id=2, name="repo-2"), + ] + + calls = [] + + def fake_process_repo(config, repo): + calls.append(repo.repo_id) + return repo.repo_id == 1 + + monkeypatch.setattr(cli, "GiteaApi", FakeApi) + monkeypatch.setattr(cli, "process_repo", fake_process_repo) + + assert cli.run(make_config(tmp_path)) == 1 + assert calls == [1, 2] + + +def test_run_returns_success_when_all_repos_succeed(monkeypatch, tmp_path): + class FakeApi: + def __init__(self, endpoint, api_token): + pass + + def get_repositories(self): + return [make_repo()] + + monkeypatch.setattr(cli, "GiteaApi", FakeApi) + monkeypatch.setattr(cli, "process_repo", lambda config, repo: True) + + assert cli.run(make_config(tmp_path)) == 0 + + +def test_main_returns_failure_for_invalid_config(monkeypatch): + def fake_read_toml_config(path): + raise RuntimeError("bad") + + monkeypatch.setattr(cli, "read_toml_config", fake_read_toml_config) + + assert cli.main(["bad.toml"]) == 1 + + +def test_main_uses_config_path(monkeypatch, tmp_path): + config = make_config(tmp_path) + calls = [] + + monkeypatch.setattr(cli, "read_toml_config", lambda path: config) + monkeypatch.setattr(cli, "setup_logging", lambda logging_config: None) + monkeypatch.setattr(cli, "run", lambda config: calls.append(config) or 0) + + assert cli.main(["config.toml"]) == 0 + assert calls == [config] diff --git a/tests/test_config.py b/tests/test_config.py index 6286bb2..3f20f39 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -120,6 +120,29 @@ from src.models import LoggingOutputConfig 'token = "repo-token"', None, ), + ( + '[main]\napi_token = "something"\n' + 'format = "{unknown}"\n' + 'endpoint = "https://example.com"\n' + 'out_dir = "/home/user/repositories"\n' + "\n" + "[fetch]\n" + 'ssh_key = "/tmp/no_key"', + None, + ), + ( + '[main]\napi_token = "something"\n' + 'format = "{owner}/{name}"\n' + 'endpoint = "https://example.com"\n' + 'out_dir = "/home/user/repositories"\n' + "\n" + "[fetch]\n" + 'ssh_key = "/tmp/no_key"\n' + "\n" + "[logging]\n" + 'level = "INF"', + None, + ), ("not valid toml", None), ], ) diff --git a/tests/test_gitea_api.py b/tests/test_gitea_api.py new file mode 100644 index 0000000..ab0229c --- /dev/null +++ b/tests/test_gitea_api.py @@ -0,0 +1,101 @@ +import datetime + +import pytest +import requests +from pydantic import ValidationError + +from src.gitea_api import GiteaApi, GiteaApiError + + +def repo_data(repo_id=42): + return { + "id": repo_id, + "ssh_url": "ssh://git@example.com/alex/repo.git", + "clone_url": "https://example.com/alex/repo.git", + "name": "repo", + "updated_at": datetime.datetime.now(datetime.UTC).isoformat(), + "owner": { + "id": 23, + "login": "alex", + "email": "alex@example.com", + }, + } + + +class FakeResponse: + def __init__(self, status_code=200, data=None, json_error=None): + self.status_code = status_code + self._data = data + self._json_error = json_error + + def json(self): + if self._json_error: + raise self._json_error + return self._data + + +class FakeSession: + def __init__(self, responses=None, request_error=None): + self.headers = {} + self.responses = list(responses or []) + self.request_error = request_error + self.calls = [] + + def get(self, url, params, timeout): + self.calls.append((url, params, timeout)) + if self.request_error: + raise self.request_error + return self.responses.pop(0) + + +def test_get_repositories_fetches_all_pages_with_timeout(monkeypatch): + session = FakeSession( + [ + FakeResponse(data=[repo_data(repo_id=1)]), + FakeResponse(data=[repo_data(repo_id=2)]), + FakeResponse(data=[]), + ] + ) + monkeypatch.setattr(requests, "session", lambda: session) + + repos = GiteaApi( + endpoint="https://example.com", + api_token="api-token", + ).get_repositories(page_size=1) + + assert [repo.repo_id for repo in repos] == [1, 2] + assert session.headers == {"Authorization": "token api-token"} + assert session.calls[0][2] == (5, 30) + + +def test_get_repositories_raises_on_http_error(monkeypatch): + session = FakeSession([FakeResponse(status_code=500, data=[])]) + monkeypatch.setattr(requests, "session", lambda: session) + + with pytest.raises(GiteaApiError, match="status 500"): + GiteaApi( + endpoint="https://example.com", + api_token="api-token", + ).get_repositories() + + +def test_get_repositories_raises_on_network_error(monkeypatch): + session = FakeSession(request_error=requests.Timeout("timed out")) + monkeypatch.setattr(requests, "session", lambda: session) + + with pytest.raises(GiteaApiError, match="request failed"): + GiteaApi( + endpoint="https://example.com", + api_token="api-token", + ).get_repositories() + + +def test_get_repositories_raises_on_invalid_payload(monkeypatch): + session = FakeSession([FakeResponse(data=[{"id": 1}])]) + monkeypatch.setattr(requests, "session", lambda: session) + + with pytest.raises((GiteaApiError, ValidationError)): + GiteaApi( + endpoint="https://example.com", + api_token="api-token", + ).get_repositories() diff --git a/tests/test_mirror_process.py b/tests/test_mirror_process.py index 88b3586..1384ad6 100644 --- a/tests/test_mirror_process.py +++ b/tests/test_mirror_process.py @@ -1,19 +1,10 @@ import datetime -import importlib.util -from pathlib import Path +from src import cli from src.config import Config, FetchConfig from src.models import GiteaRepository, GiteaUser -def load_mirror_module(): - module_path = Path(__file__).parent.parent / "gitea-mirror.py" - spec = importlib.util.spec_from_file_location("gitea_mirror", module_path) - module = importlib.util.module_from_spec(spec) - spec.loader.exec_module(module) - return module - - def make_repo() -> GiteaRepository: return GiteaRepository( ssh_url="ssh://git@example.com/alex/repo.git", @@ -40,16 +31,15 @@ def make_config(tmp_path, fetch: FetchConfig) -> Config: def test_process_repo_clones_with_ssh_url_and_key_env(monkeypatch, tmp_path): - mirror = load_mirror_module() calls = [] def fake_git_clone(repository_url, repository, env): calls.append((repository_url, repository, env)) return True - monkeypatch.setattr(mirror, "git_clone", fake_git_clone) + monkeypatch.setattr(cli, "git_clone", fake_git_clone) - mirror.process_repo( + assert cli.process_repo( config=make_config(tmp_path, FetchConfig(ssh_key="/tmp/test_key")), repo=make_repo(), ) @@ -68,16 +58,15 @@ def test_process_repo_clones_with_ssh_url_and_key_env(monkeypatch, tmp_path): def test_process_repo_clones_with_https_url_and_auth_env(monkeypatch, tmp_path): - mirror = load_mirror_module() calls = [] def fake_git_clone(repository_url, repository, env): calls.append((repository_url, repository, env)) return True - monkeypatch.setattr(mirror, "git_clone", fake_git_clone) + monkeypatch.setattr(cli, "git_clone", fake_git_clone) - mirror.process_repo( + assert cli.process_repo( config=make_config(tmp_path, FetchConfig(user="alex", token="secret")), repo=make_repo(), ) @@ -91,7 +80,6 @@ def test_process_repo_clones_with_https_url_and_auth_env(monkeypatch, tmp_path): def test_process_repo_pulls_existing_repo_with_fetch_env(monkeypatch, tmp_path): - mirror = load_mirror_module() calls = [] repo_path = tmp_path / "alex" / "repo" / ".git" repo_path.mkdir(parents=True) @@ -100,9 +88,9 @@ def test_process_repo_pulls_existing_repo_with_fetch_env(monkeypatch, tmp_path): calls.append((repository, env)) return True - monkeypatch.setattr(mirror, "git_pull", fake_git_pull) + monkeypatch.setattr(cli, "git_pull", fake_git_pull) - mirror.process_repo( + assert cli.process_repo( config=make_config(tmp_path, FetchConfig(ssh_key="/tmp/test_key")), repo=make_repo(), ) @@ -117,3 +105,23 @@ def test_process_repo_pulls_existing_repo_with_fetch_env(monkeypatch, tmp_path): }, ) ] + + +def test_process_repo_returns_false_when_clone_fails(monkeypatch, tmp_path): + monkeypatch.setattr(cli, "git_clone", lambda **kwargs: False) + + assert not cli.process_repo( + config=make_config(tmp_path, FetchConfig(ssh_key="/tmp/test_key")), + repo=make_repo(), + ) + + +def test_process_repo_returns_false_when_pull_fails(monkeypatch, tmp_path): + repo_path = tmp_path / "alex" / "repo" / ".git" + repo_path.mkdir(parents=True) + monkeypatch.setattr(cli, "git_pull", lambda repository, env: False) + + assert not cli.process_repo( + config=make_config(tmp_path, FetchConfig(ssh_key="/tmp/test_key")), + repo=make_repo(), + ) diff --git a/tests/test_repository_name.py b/tests/test_repository_name.py index 8bd6665..ea5a235 100644 --- a/tests/test_repository_name.py +++ b/tests/test_repository_name.py @@ -1,6 +1,7 @@ import pytest -from src.repository_name import is_valid_format +from src.repository_name import get_repository_path, is_valid_format +from tests.test_mirror_process import make_repo @pytest.mark.parametrize( @@ -13,3 +14,27 @@ from src.repository_name import is_valid_format ) def test_name_formatting(name_format, expected): assert is_valid_format(name_format) == expected + + +@pytest.mark.parametrize( + "name_format", + [ + "../{name}", + "/tmp/{name}", + ], +) +def test_repository_path_rejects_paths_outside_out_dir(tmp_path, name_format): + with pytest.raises(ValueError): + get_repository_path( + out_dir=str(tmp_path), + name_format=name_format, + repo=make_repo(), + ) + + +def test_repository_path_returns_path_inside_out_dir(tmp_path): + assert get_repository_path( + out_dir=str(tmp_path), + name_format="{owner}/{name}", + repo=make_repo(), + ) == str(tmp_path / "alex" / "repo")