FabledGrimoire/tests/test_services_forge.py

"""Forge adapter contract tests — the Gitea implementation against a
mocked transport, plus the configuration gate.

httpx.MockTransport is the fake forge: the adapter takes an injectable
transport precisely so the CONTRACT (URLs hit, auth header shape, payload
decoding, error taxonomy) is testable with no live server and no new
dependency. These are the reference behaviors step 8's GitHub adapter must
reproduce.

The most load-bearing tests are the OFF ones: a user with no usable connection
must get an empty selector, because every consumer treats that as "behave as
if the module didn't exist" (the baseline user has no forge).
DB-backed keyring behavior (get_forges: rows, the project pin, the admin-only
env entry) lives in tests/test_integration_forge_keyring.py.
"""
import base64

import httpx
import pytest

from grimoire.services.forge import (
    ForgeError,
    ForgeNotFound,
    ForgeSelector,
    GiteaForge,
    build_adapter,
)

BASE = "https://git.example.com"


def _forge(handler) -> GiteaForge:
    return GiteaForge(BASE, "tok-123", transport=httpx.MockTransport(handler))


def _json(status: int, payload) -> httpx.Response:
    return httpx.Response(status, json=payload)


# --- resolve_repo: the join between recorded repos and this forge ------------

@pytest.mark.parametrize(
    ("recorded", "expected"),
    [
        ("https://git.example.com/alice/Widget.git", "alice/widget"),
        ("git@git.example.com:alice/widget.git", "alice/widget"),
        ("git.example.com/alice/widget", "alice/widget"),
        # Nested (GitLab-style) groups survive as the API path remainder.
        ("https://git.example.com/team/sub/widget", "team/sub/widget"),
        # Another host is a NORMAL miss, not an error.
        ("https://github.com/alice/widget", None),
        ("", None),
        ("not a url", None),
        # Host alone, no owner/repo remainder.
        ("git.example.com", None),
    ],
)
def test_resolve_repo_matches_by_host_and_yields_the_api_path(recorded, expected):
    forge = GiteaForge(BASE, "tok")
    assert forge.resolve_repo(recorded) == expected


# --- read_file ---------------------------------------------------------------

async def test_read_file_decodes_content_and_carries_the_commit():
    seen = {}

    def handler(request: httpx.Request) -> httpx.Response:
        seen["url"] = str(request.url)
        seen["auth"] = request.headers.get("Authorization")
        return _json(200, {
            "type": "file",
            "encoding": "base64",
            "content": base64.b64encode("def x():\n    return 1\n".encode()).decode(),
            "sha": "blob" * 10,
            "last_commit_sha": "c" * 40,
            "path": "src/x.py",
        })

    got = await _forge(handler).read_file("alice/widget", "src/x.py", ref="dev")
    assert got.content == "def x():\n    return 1\n"
    assert got.commit_sha == "c" * 40
    assert got.path == "src/x.py"
    assert "/api/v1/repos/alice/widget/contents/src/x.py" in seen["url"]
    assert "ref=dev" in seen["url"]
    assert seen["auth"] == "token tok-123"


async def test_read_file_404_is_not_found_and_a_directory_is_too():
    with pytest.raises(ForgeNotFound):
        await _forge(lambda r: _json(404, {"message": "no"})).read_file(
            "alice/widget", "gone.py"
        )
    # The contents API returns a LIST for a directory — that's "no such file",
    # not a decoding error.
    with pytest.raises(ForgeNotFound):
        await _forge(lambda r: _json(200, [{"type": "file"}])).read_file(
            "alice/widget", "src"
        )


async def test_read_file_auth_failure_names_the_scope_never_the_token():
    with pytest.raises(ForgeError) as err:
        await _forge(lambda r: _json(401, {})).read_file("alice/widget", "x.py")
    assert "tok-123" not in str(err.value)
    assert "scope" in str(err.value)


async def test_read_file_binary_content_is_a_forge_error():
    def handler(request):
        return _json(200, {
            "type": "file", "encoding": "base64",
            "content": base64.b64encode(b"\xff\xfe\x00\x01").decode(),
        })

    with pytest.raises(ForgeError):
        await _forge(handler).read_file("alice/widget", "img.bin")


async def test_unreachable_forge_is_a_forge_error_not_a_crash():
    def handler(request):
        raise httpx.ConnectError("boom", request=request)

    with pytest.raises(ForgeError):
        await _forge(handler).read_file("alice/widget", "x.py")


# --- default_branch / check --------------------------------------------------

async def test_default_branch_reads_the_repo_record():
    forge = _forge(lambda r: _json(200, {"default_branch": "dev"}))
    assert await forge.default_branch("alice/widget") == "dev"


async def test_check_reports_version_and_identity():
    def handler(request):
        if request.url.path.endswith("/version"):
            return _json(200, {"version": "1.23.1"})
        return _json(200, {"login": "grimoire-bot"})

    result = await _forge(handler).check()
    assert result == {"ok": True, "version": "1.23.1", "username": "grimoire-bot"}


# --- the configuration gate (build_adapter) ----------------------------------

def test_empty_or_partial_values_build_nothing():
    # A base URL with no token (or vice versa) must not half-enable anything.
    assert build_adapter("", "", "") is None
    assert build_adapter("gitea", BASE, "") is None
    assert build_adapter("gitea", "", "tok") is None
    assert build_adapter("", BASE, "tok") is None  # no kind selected


def test_unknown_kind_disables_with_a_warning_not_a_crash():
    assert build_adapter("sourcehut", BASE, "tok") is None


def test_schemeless_base_url_disables():
    assert build_adapter("gitea", "git.example.com", "tok") is None


def test_full_values_build_a_gitea_adapter():
    forge = build_adapter("Gitea", BASE + "/", "tok")  # case + slash normalized
    assert isinstance(forge, GiteaForge)
    assert forge.base_url == BASE
    assert forge.host == "git.example.com"


def test_full_values_build_a_github_adapter():
    from grimoire.services.forge import GitHubForge

    assert isinstance(
        build_adapter("github", "https://github.com", "tok"), GitHubForge
    )


# --- the selector: host-keyed resolution over a keyring ----------------------

def test_empty_selector_is_the_no_forge_baseline():
    selector = ForgeSelector()
    assert not selector.configured
    assert selector.resolve(f"{BASE}/alice/widget") is None


def test_selector_resolves_by_host_across_the_keyring():
    gitea = build_adapter("gitea", BASE, "tok")
    github = build_adapter("github", "https://github.com", "tok2")
    selector = ForgeSelector((gitea, github))
    assert selector.configured

    hit = selector.resolve("https://git.example.com/alice/widget")
    assert hit == (gitea, "alice/widget")
    hit = selector.resolve("git@github.com:alice/Widget.git")
    assert hit == (github, "alice/widget")
    # A host neither connection serves is a NORMAL miss, not an error.
    assert selector.resolve("https://elsewhere.example.org/a/b") is None


def test_forge_error_taxonomy_is_catchable_as_one_family():
    assert issubclass(ForgeNotFound, ForgeError)
    assert issubclass(ForgeError, RuntimeError)


def test_adapter_contract_surface():
    """Both adapters implement exactly this surface — the second
    implementation is what proves it's a contract."""
    from grimoire.services.forge import FORGE_KINDS, GitHubForge

    for cls in (GiteaForge, GitHubForge):
        for method in (
            "read_file", "latest_commit", "archive",
            "default_branch", "resolve_repo", "check",
        ):
            assert callable(getattr(cls, method))
    assert GiteaForge.kind == "gitea"
    assert GitHubForge.kind == "github"
    assert set(FORGE_KINDS) == {"gitea", "github"}


def test_forge_routes_registered_at_their_post_2778_homes():
    """Connections are USER settings; only the webhook secret stays admin.
    The old instance-wide /api/admin/forge endpoints must be GONE, not
    coexisting — a legacy config surface that still wrote admin settings
    would silently configure nothing (one path, not two)."""
    from grimoire.app import create_app
    from grimoire.routes import admin as admin_routes

    for name in ("get_forge_webhook_settings", "update_forge_webhook_settings"):
        assert callable(getattr(admin_routes, name))
    rules = {r.rule for r in create_app().url_map.iter_rules()}
    assert "/api/admin/forge-webhook" in rules
    assert "/api/admin/forge" not in rules
    assert "/api/admin/forge/test" not in rules
    assert "/api/settings/forge-connections" in rules
    assert "/api/settings/forge-connections/<int:connection_id>" in rules
    assert "/api/settings/forge-connections/<int:connection_id>/test" in rules
    assert "/api/projects/<int:project_id>/forge" in rules


def test_settings_kv_surface_masks_the_webhook_secret():
    """The generic /api/settings dump masks secrets too — otherwise the admin
    endpoints' masking is bypassable by reading the raw KV rows. The forge
    token lives on keyring rows, not in the KV, so it is deliberately not a
    secret KEY here."""
    from grimoire.routes.settings import _SECRET_KEYS, _masked

    out = _masked({"forge_webhook_secret": "sec", "smtp_password": "pw", "theme": "dark"})
    assert out["forge_webhook_secret"] == "********"
    assert out["smtp_password"] == "********"
    assert out["theme"] == "dark"
    assert {"forge_webhook_secret", "smtp_password"} <= set(_SECRET_KEYS)
    assert "forge_token" not in _SECRET_KEYS
    # An unset secret stays empty rather than reading as a set-but-masked one.
    assert _masked({"forge_webhook_secret": ""})["forge_webhook_secret"] == ""


def test_connection_to_dict_never_carries_the_token():
    """The model is the last line: every list/create/update route returns
    to_dict(), so a field added here is a field leaked there."""
    from grimoire.models.forge_connection import ForgeConnection

    assert "token" not in ForgeConnection.to_dict.__code__.co_consts


def test_config_has_the_docker_secret_channel():
    from grimoire.config import Config
    for attr in ("FORGE_KIND", "FORGE_BASE_URL", "FORGE_TOKEN"):
        assert hasattr(Config, attr)


# --- the GitHub adapter -----------------------------------------------------
# Same contract, second implementation. Where behavior below differs from the
# Gitea tests above, that difference IS the adapter's job: API host mapping,
# Bearer auth, the missing last_commit_sha, the codeload redirect.

def _github(handler, base: str = "https://github.com"):
    from grimoire.services.forge import GitHubForge

    return GitHubForge(base, "gh-tok", transport=httpx.MockTransport(handler))


def test_github_resolve_repo_is_the_same_host_join():
    from grimoire.services.forge import GitHubForge

    forge = GitHubForge("https://github.com", "t")
    assert forge.resolve_repo("git@github.com:alice/Widget.git") == "alice/widget"
    # A Gitea-hosted repo is a NORMAL miss for a GitHub forge, and vice versa.
    assert forge.resolve_repo("https://git.example.com/alice/widget") is None


async def test_github_api_base_maps_dot_com_and_enterprise():
    seen = []

    def handler(request):
        seen.append(str(request.url))
        return _json(200, {"default_branch": "main"})

    await _github(handler).default_branch("alice/widget")
    await _github(handler, base="https://ghe.example.com").default_branch("alice/widget")
    assert seen[0] == "https://api.github.com/repos/alice/widget"
    assert seen[1] == "https://ghe.example.com/api/v3/repos/alice/widget"


async def test_github_read_file_decodes_and_stamps_from_the_commits_call():
    content = "def canonical():\n    return 1\n"

    def handler(request):
        assert request.headers["Authorization"] == "Bearer gh-tok"
        assert request.headers["X-GitHub-Api-Version"]
        if request.url.path.endswith("/commits"):
            assert request.url.params["path"] == "src/x.py"
            assert request.url.params["per_page"] == "1"
            return _json(200, [{"sha": "c" * 40}])
        return _json(200, {
            "type": "file", "encoding": "base64",
            "content": base64.b64encode(content.encode()).decode(),
            "path": "src/x.py", "sha": "blob-sha-not-a-point-in-history",
        })

    f = await _github(handler).read_file("alice/widget", "src/x.py")
    assert f.content == content
    # From /commits — GitHub's contents payload only carries the blob sha,
    # which is a content address, not the provenance stamp.
    assert f.commit_sha == "c" * 40


async def test_github_read_file_serves_content_even_when_the_stamp_fails():
    content = "x = 1\n"

    def handler(request):
        if request.url.path.endswith("/commits"):
            return httpx.Response(500)
        return _json(200, {"type": "file", "encoding": "base64",
                           "content": base64.b64encode(content.encode()).decode()})

    f = await _github(handler).read_file("alice/widget", "x.py")
    assert f.content == content
    assert f.commit_sha == ""  # unknown stamp, not a failed read


async def test_github_archive_follows_the_codeload_redirect():
    def handler(request):
        if request.url.host == "api.github.com":
            return httpx.Response(302, headers={
                "Location": "https://codeload.github.com/alice/widget/tar.gz/main",
            })
        assert request.url.host == "codeload.github.com"
        # httpx drops Authorization on the cross-host hop — codeload's URL
        # carries its own grant, and leaking the PAT there would be a bug.
        assert "Authorization" not in request.headers
        return httpx.Response(200, content=b"tarball-bytes")

    assert await _github(handler).archive("alice/widget", "main") == b"tarball-bytes"


async def test_github_check_probes_the_token_with_user():
    result = await _github(lambda r: _json(200, {"login": "octo"})).check()
    assert result["ok"] is True
    assert result["username"] == "octo"


async def test_latest_commit_parses_tolerantly_on_both_adapters():
    """The one caller treats latest_commit as an optimization with a fallback,
    so a surprising payload must read as "don't know", never raise."""
    assert await _github(
        lambda r: _json(200, [{"sha": "d" * 40}])
    ).latest_commit("a/w", "x.py") == "d" * 40
    assert await _github(
        lambda r: _json(200, {"weird": True})
    ).latest_commit("a/w", "x.py") == ""
    assert await _forge(
        lambda r: _json(200, [{"sha": "e" * 40}])
    ).latest_commit("a/w", "x.py") == "e" * 40
    assert await _forge(lambda r: _json(200, [])).latest_commit("a/w", "x.py") == ""