reuse-surface/tests/test_hub.py

from __future__ import annotations

import json
import os
from pathlib import Path
from unittest.mock import patch

import pytest
from fastapi.testclient import TestClient

from reuse_surface.hub.app import create_app
from reuse_surface.hub.store import HubStore

REMOTE_INDEX = """
version: 1
domain: helix_forge
updated: "2026-06-15"
capabilities:
  - id: capability.remote.sample
    name: Remote Sample
    domain: helix_forge
    vector: D2/A0/C0/R0
    owner: example
    path: registry/capabilities/capability.remote.sample.md
    summary: Sample capability from a remote index
    tags: [sample]
    consumption_modes: [planning]
"""


@pytest.fixture
def hub_client(tmp_path, monkeypatch):
    db_path = tmp_path / "hub.db"
    cache_dir = tmp_path / "cache"
    monkeypatch.setenv("REUSE_SURFACE_TOKEN", "test-token")
    monkeypatch.setenv("REUSE_SURFACE_DB", str(db_path))
    monkeypatch.setenv("REUSE_SURFACE_CACHE_DIR", str(cache_dir))
    app = create_app()
    with TestClient(app) as client:
        yield client


def test_health(hub_client):
    response = hub_client.get("/health")
    assert response.status_code == 200
    assert response.json()["status"] == "ok"


def test_register_requires_auth(hub_client):
    response = hub_client.post(
        "/v1/repos",
        json={
            "repo": "demo",
            "url": "https://example.com/capabilities.yaml",
            "domain": "helix_forge",
        },
    )
    assert response.status_code == 401


def test_register_and_list(hub_client):
    payload = {
        "repo": "demo",
        "url": "https://example.com/capabilities.yaml",
        "domain": "helix_forge",
        "description": "test",
    }
    response = hub_client.post(
        "/v1/repos",
        json=payload,
        headers={"Authorization": "Bearer test-token"},
    )
    assert response.status_code == 201
    listed = hub_client.get("/v1/repos")
    assert listed.status_code == 200
    assert listed.json()["count"] == 1
    assert "auth_env" not in listed.json()["repos"][0]


def test_update_registration(hub_client):
    hub_client.post(
        "/v1/repos",
        json={
            "repo": "demo",
            "url": "https://example.com/capabilities.yaml",
            "domain": "helix_forge",
        },
        headers={"Authorization": "Bearer test-token"},
    )
    response = hub_client.patch(
        "/v1/repos/demo",
        json={"enabled": False},
        headers={"Authorization": "Bearer test-token"},
    )
    assert response.status_code == 200
    assert response.json()["enabled"] is False


def test_compose_federated_with_mock_fetch(hub_client, monkeypatch):
    hub_client.post(
        "/v1/repos",
        json={
            "repo": "remote-repo",
            "url": "https://example.com/capabilities.yaml",
            "domain": "helix_forge",
            "enabled": True,
        },
        headers={"Authorization": "Bearer test-token"},
    )
    payload = REMOTE_INDEX.encode("utf-8")

    class FakeResponse:
        def __enter__(self):
            return self

        def __exit__(self, *args):
            return False

        def read(self):
            return payload

    with patch("urllib.request.urlopen", return_value=FakeResponse()):
        response = hub_client.get("/v1/federated?refresh=true")
    assert response.status_code == 200
    data = response.json()
    ids = {item["id"] for item in data["capabilities"]}
    assert "capability.remote.sample" in ids


def test_store_validation(tmp_path):
    store = HubStore(tmp_path / "hub.db")
    with pytest.raises(ValueError):
        store.create_repo({"repo": "BAD", "url": "ftp://x", "domain": "helix_forge"})


# --- T02: composed_at / stale tracking ---


def test_compose_state_starts_unset(tmp_path):
    store = HubStore(tmp_path / "hub.db")
    state = store.get_compose_state()
    assert state == {"composed_at": None, "stale": False}


def test_record_compose_sets_timestamp_and_clears_stale(tmp_path):
    store = HubStore(tmp_path / "hub.db")
    store.mark_stale()
    assert store.get_compose_state()["stale"] is True
    composed_at = store.record_compose()
    state = store.get_compose_state()
    assert state["composed_at"] == composed_at
    assert state["stale"] is False


def test_plain_get_does_not_clear_stale(hub_client, monkeypatch):
    hub_client.post(
        "/v1/repos",
        json={"repo": "remote-repo", "url": "https://example.com/capabilities.yaml", "domain": "helix_forge"},
        headers={"Authorization": "Bearer test-token"},
    )
    payload = REMOTE_INDEX.encode("utf-8")

    class FakeResponse:
        def __enter__(self):
            return self

        def __exit__(self, *args):
            return False

        def read(self):
            return payload

    with patch("urllib.request.urlopen", return_value=FakeResponse()):
        # force a real compose so composed_at/stale are established
        first = hub_client.get("/v1/federated?refresh=true")
        assert first.json()["stale"] is False
        composed_at_1 = first.json()["composed_at"]

        # a plain GET (no refresh) must not report itself as freshly composed
        second = hub_client.get("/v1/federated")
        assert second.json()["composed_at"] == composed_at_1
        assert second.json()["stale"] is False


def test_get_federated_reports_stale_after_mark(hub_client, monkeypatch):
    hub_client.post(
        "/v1/repos",
        json={"repo": "remote-repo", "url": "https://example.com/capabilities.yaml", "domain": "helix_forge"},
        headers={"Authorization": "Bearer test-token"},
    )
    payload = REMOTE_INDEX.encode("utf-8")

    class FakeResponse:
        def __enter__(self):
            return self

        def __exit__(self, *args):
            return False

        def read(self):
            return payload

    with patch("urllib.request.urlopen", return_value=FakeResponse()):
        hub_client.get("/v1/federated?refresh=true")

    from reuse_surface.hub.store import HubStore as _HubStore

    db_path = os.environ["REUSE_SURFACE_DB"]
    _HubStore(Path(db_path)).mark_stale()

    response = hub_client.get("/v1/federated")
    assert response.json()["stale"] is True


# --- T02: Forgejo webhook receiver ---

WEBHOOK_SECRET = "test-webhook-secret"


def _sign(body: bytes, secret: str = WEBHOOK_SECRET) -> str:
    import hashlib
    import hmac as hmac_module

    return hmac_module.new(secret.encode("utf-8"), body, hashlib.sha256).hexdigest()


@pytest.fixture
def webhook_client(hub_client, monkeypatch):
    monkeypatch.setenv("REUSE_SURFACE_FORGEJO_WEBHOOK_SECRET", WEBHOOK_SECRET)
    return hub_client


def test_webhook_rejects_missing_signature(webhook_client):
    response = webhook_client.post("/v1/webhooks/forgejo", json={"commits": []})
    assert response.status_code == 401


def test_webhook_rejects_wrong_signature(webhook_client):
    body = json.dumps({"commits": []}).encode("utf-8")
    response = webhook_client.post(
        "/v1/webhooks/forgejo",
        content=body,
        headers={"X-Forgejo-Signature": "0" * 64, "Content-Type": "application/json"},
    )
    assert response.status_code == 401


def test_webhook_rejects_when_secret_not_configured(hub_client):
    body = json.dumps({"commits": []}).encode("utf-8")
    response = hub_client.post(
        "/v1/webhooks/forgejo",
        content=body,
        headers={"X-Forgejo-Signature": _sign(body), "Content-Type": "application/json"},
    )
    assert response.status_code == 503


def test_webhook_ignores_push_without_registry_change(webhook_client):
    body = json.dumps({"commits": [{"added": ["README.md"], "modified": [], "removed": []}]}).encode("utf-8")
    response = webhook_client.post(
        "/v1/webhooks/forgejo",
        content=body,
        headers={"X-Forgejo-Signature": _sign(body), "Content-Type": "application/json"},
    )
    assert response.status_code == 200
    assert response.json()["accepted"] is False


def test_webhook_triggers_recompose_on_registry_change(webhook_client, monkeypatch):
    webhook_client.post(
        "/v1/repos",
        json={"repo": "remote-repo", "url": "https://example.com/capabilities.yaml", "domain": "helix_forge"},
        headers={"Authorization": "Bearer test-token"},
    )
    payload_bytes = REMOTE_INDEX.encode("utf-8")

    class FakeResponse:
        def __enter__(self):
            return self

        def __exit__(self, *args):
            return False

        def read(self):
            return payload_bytes

    body = json.dumps(
        {"commits": [{"added": ["registry/indexes/capabilities.yaml"], "modified": [], "removed": []}]}
    ).encode("utf-8")
    with patch("urllib.request.urlopen", return_value=FakeResponse()):
        response = webhook_client.post(
            "/v1/webhooks/forgejo",
            content=body,
            headers={"X-Forgejo-Signature": _sign(body), "Content-Type": "application/json"},
        )
    assert response.status_code == 200
    assert response.json()["accepted"] is True
    assert response.json()["composed_at"]

    # federated index should now report the newly composed data, not stale
    follow_up = webhook_client.get("/v1/federated")
    assert follow_up.json()["stale"] is False
    ids = {item["id"] for item in follow_up.json()["capabilities"]}
    assert "capability.remote.sample" in ids


def test_webhook_accepts_gitea_signature_header(webhook_client):
    body = json.dumps({"commits": [{"added": ["README.md"], "modified": [], "removed": []}]}).encode("utf-8")
    response = webhook_client.post(
        "/v1/webhooks/forgejo",
        content=body,
        headers={"X-Gitea-Signature": _sign(body), "Content-Type": "application/json"},
    )
    assert response.status_code == 200