FabledScribe/tests/test_services_forge.py

"""Forge adapter contract tests (#2689) — 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: an unconfigured instance must
get None from get_forge(), because every consumer treats None as "behave as if
the module didn't exist" (rule #115 — the baseline install has no forge).
"""
import base64
from unittest.mock import AsyncMock, patch

import httpx
import pytest

from scribe.services.forge import (
    ForgeError,
    ForgeNotFound,
    GiteaForge,
    get_forge,
)

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": "scribe-bot"})

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


# --- the configuration gate --------------------------------------------------

def _settings(values: dict):
    async def fake(key, default=""):
        return values.get(key, default)
    return patch("scribe.services.forge.get_admin_setting", AsyncMock(side_effect=fake))


async def test_unconfigured_instance_gets_none():
    with _settings({}), patch("scribe.services.forge.Config") as cfg:
        cfg.FORGE_KIND = cfg.FORGE_BASE_URL = cfg.FORGE_TOKEN = ""
        assert await get_forge() is None


async def test_partial_config_is_still_off():
    # A base URL with no token (or vice versa) must not half-enable anything.
    for values in (
        {"forge_kind": "gitea", "forge_base_url": BASE},
        {"forge_kind": "gitea", "forge_token": "tok"},
        {"forge_base_url": BASE, "forge_token": "tok"},  # no kind selected
    ):
        with _settings(values), patch("scribe.services.forge.Config") as cfg:
            cfg.FORGE_KIND = cfg.FORGE_BASE_URL = cfg.FORGE_TOKEN = ""
            assert await get_forge() is None


async def test_unknown_kind_disables_with_a_warning_not_a_crash():
    with _settings({
        "forge_kind": "sourcehut", "forge_base_url": BASE, "forge_token": "tok",
    }), patch("scribe.services.forge.Config") as cfg:
        cfg.FORGE_KIND = cfg.FORGE_BASE_URL = cfg.FORGE_TOKEN = ""
        assert await get_forge() is None


async def test_full_config_builds_a_gitea_adapter():
    with _settings({
        "forge_kind": "gitea",
        "forge_base_url": BASE + "/",  # trailing slash normalized away
        "forge_token": "tok",
    }), patch("scribe.services.forge.Config") as cfg:
        cfg.FORGE_KIND = cfg.FORGE_BASE_URL = cfg.FORGE_TOKEN = ""
        forge = await get_forge()
    assert isinstance(forge, GiteaForge)
    assert forge.base_url == BASE
    assert forge.host == "git.example.com"


async def test_env_channel_fills_gaps_but_db_wins():
    # Docker-secret deployments set FORGE_* env; an admin-UI value overrides.
    with _settings({"forge_base_url": "https://db.example.com"}), \
            patch("scribe.services.forge.Config") as cfg:
        cfg.FORGE_KIND = "gitea"
        cfg.FORGE_BASE_URL = "https://env.example.com"
        cfg.FORGE_TOKEN = "env-tok"
        forge = await get_forge()
    assert isinstance(forge, GiteaForge)
    assert forge.host == "db.example.com"


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 (#2693)."""
    from scribe.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_admin_routes_registered():
    from scribe.app import create_app
    from scribe.routes import admin as admin_routes

    for name in ("get_forge_settings", "update_forge_settings", "test_forge"):
        assert callable(getattr(admin_routes, name))
    rules = {r.rule for r in create_app().url_map.iter_rules()}
    assert "/api/admin/forge" in rules
    assert "/api/admin/forge/test" in rules


def test_settings_kv_surface_masks_the_forge_token():
    """The generic /api/settings dump masked nothing — the admin endpoints'
    masking was bypassable by reading the raw KV rows (found while wiring the
    forge token; smtp_password had the same exposure)."""
    from scribe.routes.settings import _SECRET_KEYS, _masked

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


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


# --- the GitHub adapter (#2693) ----------------------------------------------
# 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 scribe.services.forge import GitHubForge

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


def test_github_resolve_repo_is_the_same_host_join():
    from scribe.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") == ""


async def test_full_config_builds_a_github_adapter():
    from scribe.services.forge import GitHubForge

    with _settings({
        "forge_kind": "github",
        "forge_base_url": "https://github.com",
        "forge_token": "tok",
    }), patch("scribe.services.forge.Config") as cfg:
        cfg.FORGE_KIND = cfg.FORGE_BASE_URL = cfg.FORGE_TOKEN = ""
        forge = await get_forge()
    assert isinstance(forge, GitHubForge)