github-action/tests/test_sync_github_metadata.py

"""Offline tests for tools/sync_github_metadata.py."""

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_the_stripped_token(self, sync):
        assert sync["parse_github_token"]("  ghp_abc\n") == "ghp_abc"

    def test_accepts_a_fine_grained_token(self, sync):
        token = "github_pat_11ABCDEFG0abcdefghijkl_mnopqrstuvwxyz0123456789"
        assert sync["parse_github_token"](token) == token

    @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)


class TestGiteaToken:
    def test_returns_the_stripped_token(self, sync):
        assert sync["parse_gitea_token"]("  abc123\n") == "abc123"

    @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)


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_returns_none_when_no_github_mirror_exists(self, sync, mirrors):
        assert sync["find_github_targets"](mirrors) is None

    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:
    @staticmethod
    def raise_http_error(sync, monkeypatch, code, reason):
        def fake_urlopen(request, timeout):
            raise urllib.error.HTTPError(
                request.full_url, code, reason, {}, io.BytesIO(b"secret detail")
            )

        monkeypatch.setattr(sync["urllib"].request, "urlopen", fake_urlopen)

    @pytest.mark.parametrize(
        ("code", "reason", "category"),
        [
            (401, "Unauthorized", "credentials"),
            (403, "Forbidden", "permissions"),
            (404, "Not Found", "configuration"),
            (500, "Server Error", "api"),
        ],
    )
    def test_categorizes_http_errors(self, sync, monkeypatch, code, reason, category):
        self.raise_http_error(sync, monkeypatch, code, reason)
        with pytest.raises(sync["WorkflowError"]) as caught:
            sync["request_json"]("https://example.test/x")
        assert caught.value.category == category
        assert str(caught.value).startswith(f"[{category}] Gitea API request failed")
        assert f"HTTP {code} ({reason})" in str(caught.value)

    def test_does_not_leak_the_response_body(self, sync, monkeypatch):
        self.raise_http_error(sync, monkeypatch, 403, "Forbidden")
        with pytest.raises(RuntimeError) as caught:
            sync["request_json"]("https://example.test/x")
        assert "secret detail" not in str(caught.value)

    def test_names_github_for_github_urls(self, sync, monkeypatch):
        self.raise_http_error(sync, monkeypatch, 401, "Unauthorized")
        with pytest.raises(RuntimeError, match="GitHub API request failed"):
            sync["request_json"]("https://api.github.com/repos/Org/repo")

    def test_reports_unreachable_service_as_api_failure(self, sync, monkeypatch):
        def fake_urlopen(request, timeout):
            raise urllib.error.URLError("name resolution failed")

        monkeypatch.setattr(sync["urllib"].request, "urlopen", fake_urlopen)
        with pytest.raises(sync["WorkflowError"]) as caught:
            sync["request_json"]("https://example.test/x")
        assert caught.value.category == "api"
        assert "unreachable" in str(caught.value)


class TestWorkflowError:
    def test_configuration_errors_carry_their_category(self, sync):
        with pytest.raises(sync["WorkflowError"]) as caught:
            sync["parse_github_token"]("")
        assert caught.value.category == "configuration"
        assert str(caught.value).startswith("[configuration] ")


class FakeApi:
    """Records calls and answers like the Gitea and GitHub REST APIs."""

    def __init__(self, mirrors, source=None, github=None):
        self.mirrors = mirrors
        self.github = github if github is not None else {"permissions": {"admin": True}}
        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
        if method == "GET" and url.startswith("https://api.github.com"):
            return self.github
        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("PAT_FOR_GITHUB", "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"),
            ("GET", "https://api.github.com/repos/Org/repo"),
            ("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[3][3] == {"description": "Desc"}
        assert api.calls[4][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[3][3] == {"description": ""}
        assert api.calls[4][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

    @pytest.mark.parametrize(
        "mirrors",
        [
            [],
            [{"remote_address": "https://gitlab.com/Org/other.git"}],
        ],
    )
    def test_succeeds_without_changes_when_no_github_mirror_exists(
        self, sync, environment, monkeypatch, capsys, mirrors
    ):
        monkeypatch.setenv("PAT_FOR_GITHUB", "")
        api = FakeApi(mirrors)
        sync["request_json"] = api

        assert sync["main"]() == 0

        assert all(call[0] == "GET" for call in api.calls)
        assert not any(call[1].startswith("https://api.github.com") for call in api.calls)
        assert "No GitHub push mirror configured" in capsys.readouterr().out

    def test_fails_before_any_change_when_github_credentials_are_missing(
        self, sync, environment, monkeypatch
    ):
        monkeypatch.setenv("PAT_FOR_GITHUB", "")
        api = FakeApi([{"remote_address": "git@github.com:Org/repo.git"}])
        sync["request_json"] = api

        with pytest.raises(RuntimeError, match="PAT_FOR_GITHUB"):
            sync["main"]()
        assert all(call[0] == "GET" for call in api.calls)

    def test_stops_before_any_change_when_github_token_cannot_administer(
        self, sync, environment
    ):
        api = FakeApi(
            [{"remote_address": "git@github.com:Org/repo.git"}],
            github={"permissions": {"admin": False, "push": True}},
        )
        sync["request_json"] = api

        with pytest.raises(sync["WorkflowError"]) as caught:
            sync["main"]()

        assert caught.value.category == "permissions"
        assert all(call[0] == "GET" for call in api.calls)

    def test_does_not_print_any_secret(self, sync, environment, capsys):
        api = FakeApi([{"remote_address": "git@github.com:Org/repo.git"}])
        sync["request_json"] = api

        sync["main"]()

        output = capsys.readouterr()
        assert "gh-token" not in output.out + output.err
        assert "gitea-token" not in output.out + output.err