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():
    """Step 8's GitHub adapter implements exactly this surface — pin it."""
    for method in ("read_file", "default_branch", "resolve_repo", "check"):
        assert callable(getattr(GiteaForge, method))
    assert GiteaForge.kind == "gitea"


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)