"""Offline tests for the Python embedded in sync-github-metadata.yml.""" import io import json import urllib.error import pytest from workflow_source import load_sync_module @pytest.fixture() def sync(): return load_sync_module() class TestGithubToken: def test_returns_stripped_token_from_json_object(self, sync): raw = json.dumps({"GITHUB_PAT": " ghp_abc "}) assert sync["parse_github_token"](raw) == "ghp_abc" @pytest.mark.parametrize("raw", [None, "", " "]) def test_rejects_missing_or_empty_value(self, sync, raw): with pytest.raises(RuntimeError, match="missing or empty"): sync["parse_github_token"](raw) def test_rejects_invalid_json(self, sync): with pytest.raises(RuntimeError, match="valid JSON"): sync["parse_github_token"]("ghp_plain_token") def test_rejects_json_that_is_not_an_object(self, sync): with pytest.raises(RuntimeError, match="JSON object"): sync["parse_github_token"]('["x"]') @pytest.mark.parametrize("value", [{}, {"GITHUB_PAT": ""}, {"GITHUB_PAT": 5}]) def test_rejects_missing_or_invalid_pat(self, sync, value): with pytest.raises(RuntimeError, match="GITHUB_PAT"): sync["parse_github_token"](json.dumps(value)) class TestGiteaToken: def test_accepts_bare_token(self, sync): assert sync["parse_gitea_token"]("abc123") == "abc123" def test_accepts_json_object(self, sync): assert sync["parse_gitea_token"]('{"GITEA_TOKEN": " t "}') == "t" def test_accepts_json_string(self, sync): assert sync["parse_gitea_token"]('"quoted"') == "quoted" @pytest.mark.parametrize("raw", [None, "", " "]) def test_rejects_missing_or_empty_value(self, sync, raw): with pytest.raises(RuntimeError, match="missing or empty"): sync["parse_gitea_token"](raw) @pytest.mark.parametrize("raw", ["{}", '{"GITEA_TOKEN": ""}', "12", "[]"]) def test_rejects_value_without_token(self, sync, raw): with pytest.raises(RuntimeError, match="GITEA_TOKEN"): sync["parse_gitea_token"](raw) class TestSplitRepository: def test_splits_owner_and_repo(self, sync): assert sync["split_repository"]("TirSystem/github-action") == ( "TirSystem", "github-action", ) @pytest.mark.parametrize("value", [None, "", "owner", "/repo", "owner/"]) def test_rejects_incomplete_names(self, sync, value): with pytest.raises(RuntimeError, match="source repository"): sync["split_repository"](value) class TestFindGithubTargets: @pytest.mark.parametrize( "address", [ "git@github.com:Org/repo.git", "https://github.com/Org/repo.git", "https://user:token@github.com/Org/repo", "ssh://git@github.com/Org/repo.git", ], ) def test_extracts_owner_and_repo(self, sync, address): mirrors = [{"remote_address": address}] assert sync["find_github_targets"](mirrors) == ("Org", "repo") def test_ignores_non_github_mirrors(self, sync): mirrors = [ {"remote_address": "https://gitlab.com/Org/other.git"}, {"remote_address": "https://github.com/Org/repo.git"}, ] assert sync["find_github_targets"](mirrors) == ("Org", "repo") @pytest.mark.parametrize( "mirrors", [ [], [{"remote_address": "https://gitlab.com/Org/other.git"}], ], ) def test_fails_when_no_github_mirror_exists(self, sync, mirrors): with pytest.raises(RuntimeError, match="found 0"): sync["find_github_targets"](mirrors) def test_fails_when_several_github_mirrors_exist(self, sync): mirrors = [ {"remote_address": "https://github.com/Org/one.git"}, {"remote_address": "https://github.com/Org/two.git"}, ] with pytest.raises(RuntimeError, match="found 2"): sync["find_github_targets"](mirrors) def test_fails_when_path_has_wrong_shape(self, sync): mirrors = [{"remote_address": "https://github.com/only-owner"}] with pytest.raises(RuntimeError, match="owner and repository"): sync["find_github_targets"](mirrors) class TestRequestJson: def test_maps_http_error_without_leaking_the_response_body(self, sync, monkeypatch): def fake_urlopen(request, timeout): raise urllib.error.HTTPError( request.full_url, 403, "Forbidden", {}, io.BytesIO(b"secret detail") ) monkeypatch.setattr(sync["urllib"].request, "urlopen", fake_urlopen) with pytest.raises(RuntimeError) as caught: sync["request_json"]("https://example.test/x") assert str(caught.value) == "API request failed with HTTP 403 (Forbidden)" class FakeApi: """Records calls and answers like the Gitea and GitHub REST APIs.""" def __init__(self, mirrors, source=None): self.mirrors = mirrors self.source = source or {"description": "Desc", "topics": ["a", "b"]} self.calls = [] def __call__(self, url, method="GET", headers=None, body=None): self.calls.append((method, url, headers, body)) if url.endswith("/push_mirrors"): return self.mirrors if url.startswith("https://git.example.test"): return self.source return None @pytest.fixture() def environment(monkeypatch): monkeypatch.setenv("GITEA_API_URL", "https://git.example.test/api/v1/") monkeypatch.setenv("GITEA_CREDENTIALS", "gitea-token") monkeypatch.setenv("SOURCE_REPOSITORY", "TirSystem/repo") monkeypatch.setenv("GITHUB_CREDENTIALS", json.dumps({"GITHUB_PAT": "gh-token"})) class TestMain: def test_syncs_description_and_topics_to_the_single_github_mirror( self, sync, environment, capsys ): api = FakeApi([{"remote_address": "git@github.com:Org/repo.git"}]) sync["request_json"] = api assert sync["main"]() == 0 requests = [(call[0], call[1]) for call in api.calls] assert requests == [ ("GET", "https://git.example.test/api/v1/repos/TirSystem/repo"), ("GET", "https://git.example.test/api/v1/repos/TirSystem/repo/push_mirrors"), ("PATCH", "https://api.github.com/repos/Org/repo"), ("PUT", "https://api.github.com/repos/Org/repo/topics"), ] assert api.calls[0][2]["Authorization"] == "token gitea-token" assert api.calls[2][2]["Authorization"] == "Bearer gh-token" assert api.calls[2][3] == {"description": "Desc"} assert api.calls[3][3] == {"names": ["a", "b"]} assert "Synced description and topics to Org/repo." in capsys.readouterr().out def test_sends_empty_values_when_gitea_has_none(self, sync, environment): api = FakeApi( [{"remote_address": "git@github.com:Org/repo.git"}], source={"description": None, "topics": None}, ) sync["request_json"] = api sync["main"]() assert api.calls[2][3] == {"description": ""} assert api.calls[3][3] == {"names": []} def test_skips_with_a_warning_when_mirror_list_is_invalid( self, sync, environment, capsys ): api = FakeApi({"message": "unexpected"}) sync["request_json"] = api assert sync["main"]() == 0 assert all(call[0] == "GET" for call in api.calls) assert "WARNING" in capsys.readouterr().out def test_fails_before_any_request_when_credentials_are_missing( self, sync, environment, monkeypatch ): monkeypatch.setenv("GITHUB_CREDENTIALS", "") api = FakeApi([]) sync["request_json"] = api with pytest.raises(RuntimeError, match="CREDENTIALS_FOR_GITHUB"): sync["main"]() assert api.calls == []