benstull/rfc-app
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(registry): hard-cut META_REPO -> REGISTRY_REPO; wire mirror into startup/webhook/sweep

Browse Source 
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Ben Stull
2026-06-03 23:16:51 -07:00
parent f7bd466f31
commit cecc6c0b41
18 changed files with 231 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/test_docs_sessions_vertical.py
+18 -13
View File
@@ -70,27 +70,32 @@ class _UpstreamHandler:
@pytest.fixture
def patched_httpx(monkeypatch):
def patched_httpx(monkeypatch, app_with_fake_gitea): # noqa: F811
"""Provide a hook the test can call to install a MockTransport.
Returns a closure: `install(handler)` patches
`app.docs_sessions.httpx.AsyncClient` so every constructed client
uses the handler's transport.
Returns a closure: `install(handler)` patches `httpx.AsyncClient`
(via `app.docs_sessions.httpx.AsyncClient`) so every constructed
client uses the handler's transport.
NB: the upstream `app_with_fake_gitea` fixture also patches
`httpx.AsyncClient` (to route gitea calls to a FakeGitea handler),
and because `httpx` is a single shared module, that patch mutates
the *same* `AsyncClient` attribute we're about to overwrite. We
therefore import the unpatched class directly from the
`httpx._client` module so our install path can construct a fresh
real client around our MockTransport without going through the
FakeGitea wrapper.
M3 note: lifespan now calls `refresh_registry` which hits FakeGitea
via the gitea transport. Since `httpx` is a module singleton, installing
the docs transport would clobber the FakeGitea mock already installed
by `app_with_fake_gitea`. We use a COMPOSITE handler: Gitea API
requests (to `http://gitea.test/`) are delegated to FakeGitea; all
other requests go to the test-specific handler.
"""
from httpx._client import AsyncClient as RealAsyncClient
_fake = app_with_fake_gitea[1]
def install(handler):
def composite(request: httpx.Request) -> httpx.Response:
if "gitea.test" in str(request.url):
return _fake.handle(request)
return handler(request)
def patched(*args, **kwargs):
kwargs["transport"] = httpx.MockTransport(handler)
kwargs["transport"] = httpx.MockTransport(composite)
return RealAsyncClient(*args, **kwargs)
monkeypatch.setattr("app.docs_sessions.httpx.AsyncClient", patched)
backend/tests/test_docs_specs_vertical.py
+15 -6
View File
@@ -79,19 +79,28 @@ class _UpstreamHandler:
@pytest.fixture
def patched_httpx(monkeypatch):
def patched_httpx(monkeypatch, app_with_fake_gitea): # noqa: F811
"""Provide a hook the test can call to install a MockTransport.
Same shape as the docs_sessions fixture — `app_with_fake_gitea`
monkeypatches `httpx.AsyncClient` for the gitea side, so we
construct from the unpatched class directly to avoid the
FakeGitea wrapper.
M3 note: lifespan now calls `refresh_registry` which hits FakeGitea
via the gitea transport. Since `httpx` is a module singleton, installing
the docs transport would clobber the FakeGitea mock already installed
by `app_with_fake_gitea`. We use a COMPOSITE handler: Gitea API
requests (to `http://gitea.test/`) are delegated to FakeGitea; all
other requests go to the test-specific handler.
"""
from httpx._client import AsyncClient as RealAsyncClient
_fake = app_with_fake_gitea[1]
def install(handler):
def composite(request: httpx.Request) -> httpx.Response:
if "gitea.test" in str(request.url):
return _fake.handle(request)
return handler(request)
def patched(*args, **kwargs):
kwargs["transport"] = httpx.MockTransport(handler)
kwargs["transport"] = httpx.MockTransport(composite)
return RealAsyncClient(*args, **kwargs)
monkeypatch.setattr("app.docs_specs.httpx.AsyncClient", patched)
backend/tests/test_migration_027.py
+1 -1
View File
@@ -13,7 +13,7 @@ def _fresh_config() -> Config:
tmp = Path(tempfile.mkdtemp(prefix="mig027-")) / "t.db"
return Config(
gitea_url="x", gitea_bot_user="x", gitea_bot_token="x", gitea_org="x",
meta_repo="meta", registry_repo="registry",
registry_repo="registry",
oauth_client_id="x", oauth_client_secret="x", app_url="x",
secret_key="x", database_path=tmp, owner_gitea_login="x",
webhook_secret="x",
backend/tests/test_multi_project_spine_vertical.py
+17 -17
View File
@@ -1,11 +1,11 @@
"""Slice M1 — the §22 multi-project spine.
"""Slice M1+M3 — the §22 multi-project spine.
Migration 026 introduces the `projects` and `project_members` tables, seeds
the single `default` project (the N=1 case, §22.13), and threads a
`project_id` column onto every slug-bearing table, backfilled to `default`.
The §22.13 startup backfill then fills the default project's content_repo
from META_REPO. These tests prove the spine lands without disturbing the
single-project app.
M3 retires the META_REPO startup backfill; content_repo now comes from the
registry mirror (projects.yaml in REGISTRY_REPO). These tests prove the
spine lands without disturbing the single-project app.
"""
from __future__ import annotations
@@ -40,7 +40,8 @@ def test_default_project_seeded_and_backfilled(app_with_fake_gitea):
assert row["id"] == "default"
# public preserves the pre-multi-project open-by-default posture.
assert row["visibility"] == "public"
# §22.13 startup backfill set content_repo from META_REPO (tmp_env).
# M3: content_repo now comes from the registry mirror (projects.yaml),
# not the retired META_REPO startup backfill.
assert row["content_repo"] == "meta"
@@ -105,22 +106,19 @@ def test_project_members_table_shape(app_with_fake_gitea):
pass
def test_seed_default_project_is_idempotent(app_with_fake_gitea):
"""Re-running the backfill never clobbers a content_repo already set —
so a later registry mirror (M3) wins over the META_REPO fallback."""
from app import db, projects
from app.config import load_config
def test_registry_mirror_is_idempotent(app_with_fake_gitea):
"""Re-running the registry mirror is safe — it upserts (overwrites) the
projects row from projects.yaml each time without raising. M3 retirement
of seed_default_project: the registry mirror is now the sole authority."""
from app import db
app, _ = app_with_fake_gitea
with TestClient(app):
db.conn().execute(
"UPDATE projects SET content_repo = 'ohm-content' WHERE id = 'default'"
)
projects.seed_default_project(load_config())
# After startup the row should have content_repo from projects.yaml.
got = db.conn().execute(
"SELECT content_repo FROM projects WHERE id = 'default'"
).fetchone()["content_repo"]
assert got == "ohm-content"
assert got == "meta"
def test_registry_repo_config_wired(app_with_fake_gitea, monkeypatch):
@@ -128,6 +126,8 @@ def test_registry_repo_config_wired(app_with_fake_gitea, monkeypatch):
monkeypatch.setenv("REGISTRY_REPO", "wiggleverse-registry")
assert load_config().registry_repo == "wiggleverse-registry"
# Optional through M1: absent resolves to empty, not a startup failure.
# M3: REGISTRY_REPO is now required — absent raises RuntimeError.
monkeypatch.delenv("REGISTRY_REPO", raising=False)
assert load_config().registry_repo == ""
import pytest
with pytest.raises(RuntimeError, match="REGISTRY_REPO"):
load_config()
backend/tests/test_propose_vertical.py
+19 -1
View File
@@ -55,6 +55,24 @@ class FakeGitea:
self._pr_counter = 0
self._commit_counter = 0
self._seed_repo("wiggleverse", "meta")
# §22 M3: the deployment's project registry. Startup refresh_registry
# reads projects.yaml here; the single 'default' project's content_repo
# points back at the seeded meta repo so the corpus mirror is unchanged.
self._seed_repo("wiggleverse", "registry")
self.files[("wiggleverse", "registry", "main", "projects.yaml")] = {
"content": (
"deployment:\n"
" name: Test Deployment\n"
" tagline: A test deployment\n"
"projects:\n"
" - id: default\n"
" name: Test Deployment\n"
" type: document\n"
" content_repo: meta\n"
" visibility: public\n"
),
"sha": "regsha0001",
}
def _seed_repo(self, owner, repo):
self.branches[(owner, repo)] = {"main": {"sha": "initial", "ts": "2026-05-23T00:00:00Z"}}
@@ -431,7 +449,7 @@ def tmp_env(monkeypatch):
"GITEA_BOT_USER": "rfc-bot",
"GITEA_BOT_TOKEN": "bot-token",
"GITEA_ORG": "wiggleverse",
"META_REPO": "meta",
"REGISTRY_REPO": "registry",
"OAUTH_CLIENT_ID": "cid",
"OAUTH_CLIENT_SECRET": "csec",
"APP_URL": "http://localhost:8000",
backend/tests/test_registry.py
+1 -1
View File
@@ -13,7 +13,7 @@ from app.config import Config
def _db():
cfg = Config(
gitea_url="x", gitea_bot_user="x", gitea_bot_token="x", gitea_org="x",
meta_repo="meta", registry_repo="registry", oauth_client_id="x",
registry_repo="registry", oauth_client_id="x",
oauth_client_secret="x", app_url="x", secret_key="x",
database_path=Path(tempfile.mkdtemp(prefix="reg-")) / "t.db",
owner_gitea_login="x", webhook_secret="x",
backend/tests/test_registry_wiring.py
+50
View File
@@ -0,0 +1,50 @@
"""Startup mirrors the registry; the registry webhook re-mirrors it."""
from __future__ import annotations
from fastapi.testclient import TestClient
from test_propose_vertical import app_with_fake_gitea, tmp_env # noqa: F401
def test_startup_mirrors_registry_into_projects_and_deployment(app_with_fake_gitea):
from app import db
app, _ = app_with_fake_gitea
with TestClient(app):
prow = db.conn().execute(
"SELECT content_repo, type, initial_state FROM projects WHERE id='default'"
).fetchone()
assert prow["content_repo"] == "meta" # from the registry, not META_REPO
assert prow["type"] == "document"
drow = db.conn().execute("SELECT name FROM deployment WHERE id=1").fetchone()
assert drow["name"] # deployment name mirrored from the registry
def test_registry_webhook_remirrors(app_with_fake_gitea):
import hashlib
import hmac
import json as _json
app, fake = app_with_fake_gitea
with TestClient(app) as client:
from app import db
new_yaml = (
"deployment:\n name: OHM\n tagline: Edited tagline\n"
"projects:\n - id: default\n name: OHM\n type: document\n"
" content_repo: meta\n visibility: public\n"
)
fake.files[("wiggleverse", "registry", "main", "projects.yaml")] = {
"content": new_yaml, "sha": "regsha2",
}
body = _json.dumps({"repository": {"full_name": "wiggleverse/registry"}}).encode()
secret = "test-webhook-secret-for-signature-verification"
sig = hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
r = client.post(
"/api/webhooks/gitea",
content=body,
headers={"X-Gitea-Event": "push", "X-Gitea-Signature": sig,
"Content-Type": "application/json"},
)
assert r.status_code == 200
tagline = db.conn().execute("SELECT tagline FROM deployment WHERE id=1").fetchone()["tagline"]
assert tagline == "Edited tagline"
backend/tests/test_webhooks_vertical.py
+2
View File
@@ -73,6 +73,7 @@ def test_config_loads_with_empty_secret_when_bypass_is_set(monkeypatch, tmp_path
monkeypatch.setenv("GITEA_BOT_USER", "rfc-bot")
monkeypatch.setenv("GITEA_BOT_TOKEN", "bot-token")
monkeypatch.setenv("GITEA_ORG", "wiggleverse")
monkeypatch.setenv("REGISTRY_REPO", "registry")
monkeypatch.setenv("OAUTH_CLIENT_ID", "cid")
monkeypatch.setenv("OAUTH_CLIENT_SECRET", "csec")
monkeypatch.setenv("SECRET_KEY", "test-secret-key")
@@ -92,6 +93,7 @@ def test_config_loads_with_secret_set(monkeypatch, tmp_path):
monkeypatch.setenv("GITEA_BOT_USER", "rfc-bot")
monkeypatch.setenv("GITEA_BOT_TOKEN", "bot-token")
monkeypatch.setenv("GITEA_ORG", "wiggleverse")
monkeypatch.setenv("REGISTRY_REPO", "registry")
monkeypatch.setenv("OAUTH_CLIENT_ID", "cid")
monkeypatch.setenv("OAUTH_CLIENT_SECRET", "csec")
monkeypatch.setenv("SECRET_KEY", "test-secret-key")