tegwick
d181043717
Implements the hub side of the shared reuse-event schema (already drafted in WP-0018-T01, schemas/reuse-event.schema.json): a SQLite reuse_events table, POST /v1/reuse-events (token-auth), GET /v1/reuse-events?capability_id= (read-only). reuse_surface/plan_check.py: refactored record_outcome around a new shared post_or_fallback_reuse_event() helper -- tries the hub first, falls back to the local JSONL only on failure/unreachability, never both. New record_manual_reuse_event() backs a new CLI command, reuse-surface record-reuse, for retroactive facts recorded outside plan-check. Privacy/scope (repo slugs and capability ids only, no code, no secrets) is enforced structurally via the schema's additionalProperties: false, not just by convention. 21 new pytest cases, 145 total pass. Live-verified against a real running hub instance: POST/GET /v1/reuse-events directly, record-reuse and plan-check --record-outcome both posting successfully to the hub, and -- after actually killing the hub process -- confirmed the fallback path writes correctly to the local JSONL instead of erroring. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
396 lines
No EOL
12 KiB
Python
396 lines
No EOL
12 KiB
Python
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
|
|
|
|
|
|
# --- T04: reuse telemetry store ---
|
|
|
|
VALID_REUSE_EVENT = {
|
|
"ts": "2026-07-08T00:00:00Z",
|
|
"consumer_repo": "some-repo",
|
|
"capability_id": "capability.infotech.issue-tracking",
|
|
"verdict": "reuse",
|
|
"outcome": "reused",
|
|
"source": "plan-check",
|
|
}
|
|
|
|
|
|
def test_record_reuse_event_requires_auth(hub_client):
|
|
response = hub_client.post("/v1/reuse-events", json=VALID_REUSE_EVENT)
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_record_reuse_event_and_list(hub_client):
|
|
response = hub_client.post(
|
|
"/v1/reuse-events",
|
|
json=VALID_REUSE_EVENT,
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
assert response.status_code == 201
|
|
|
|
listed = hub_client.get("/v1/reuse-events")
|
|
assert listed.status_code == 200
|
|
assert listed.json()["count"] == 1
|
|
assert listed.json()["events"][0]["consumer_repo"] == "some-repo"
|
|
|
|
|
|
def test_record_reuse_event_rejects_invalid_verdict(hub_client):
|
|
bad = {**VALID_REUSE_EVENT, "verdict": "not-a-verdict"}
|
|
response = hub_client.post(
|
|
"/v1/reuse-events",
|
|
json=bad,
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
|
|
def test_record_reuse_event_rejects_extra_fields(hub_client):
|
|
bad = {**VALID_REUSE_EVENT, "unexpected_field": "nope"}
|
|
response = hub_client.post(
|
|
"/v1/reuse-events",
|
|
json=bad,
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
|
|
def test_list_reuse_events_filters_by_capability_id(hub_client):
|
|
other = {**VALID_REUSE_EVENT, "capability_id": "capability.infotech.other"}
|
|
hub_client.post("/v1/reuse-events", json=VALID_REUSE_EVENT, headers={"Authorization": "Bearer test-token"})
|
|
hub_client.post("/v1/reuse-events", json=other, headers={"Authorization": "Bearer test-token"})
|
|
|
|
listed = hub_client.get("/v1/reuse-events?capability_id=capability.infotech.other")
|
|
assert listed.status_code == 200
|
|
assert listed.json()["count"] == 1
|
|
assert listed.json()["events"][0]["capability_id"] == "capability.infotech.other"
|
|
|
|
|
|
def test_list_reuse_events_empty_by_default(hub_client):
|
|
listed = hub_client.get("/v1/reuse-events")
|
|
assert listed.status_code == 200
|
|
assert listed.json() == {"count": 0, "events": []}
|
|
|
|
|
|
def test_store_record_reuse_event_and_list(tmp_path):
|
|
store = HubStore(tmp_path / "hub.db")
|
|
store.record_reuse_event(VALID_REUSE_EVENT)
|
|
events = store.list_reuse_events()
|
|
assert len(events) == 1
|
|
assert events[0]["consumer_repo"] == "some-repo"
|
|
|
|
|
|
def test_store_record_reuse_event_rejects_invalid(tmp_path):
|
|
store = HubStore(tmp_path / "hub.db")
|
|
with pytest.raises(ValueError):
|
|
store.record_reuse_event({**VALID_REUSE_EVENT, "verdict": "nope"}) |