Ben Stull
9ca07a3f81
- collections.py resolution helpers (default_collection_id, type, initial_state) - registry mirror writes project grouping fields + default-collection corpus fields - auth.project_of_rfc joins collections; project_member_role reads memberships - cache/api_*/funder writers+readers re-keyed to collection_id (cached_prs + denormalised project_id tags unchanged); api_deployment reads type/initial_state from the default collection - projects.py restamp detects bootstrap via collections; initial_state via collection - tests updated to the three-tier schema; test_migration_028 retired (superseded by 029) - add @S1 acceptance test (collection grain + N=1 serving) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
107 lines
4.0 KiB
Python
107 lines
4.0 KiB
Python
"""§22.2 registry parse + apply: validation, type-immutability, upsert."""
|
|
from __future__ import annotations
|
|
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from app import db, registry
|
|
from app.config import Config
|
|
|
|
|
|
def _db():
|
|
cfg = Config(
|
|
gitea_url="x", gitea_bot_user="x", gitea_bot_token="x", gitea_org="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",
|
|
)
|
|
db.run_migrations(cfg)
|
|
if db._CONN is not None:
|
|
db._CONN.close()
|
|
db._CONN = None
|
|
db.init(cfg)
|
|
return cfg
|
|
|
|
|
|
VALID = """
|
|
deployment:
|
|
name: Open Human Model
|
|
tagline: A model of human flourishing
|
|
projects:
|
|
- id: default
|
|
name: Open Human Model
|
|
type: document
|
|
content_repo: meta
|
|
visibility: public
|
|
"""
|
|
|
|
|
|
def test_parse_valid_registry():
|
|
doc = registry.parse_registry(VALID)
|
|
assert doc.deployment_name == "Open Human Model"
|
|
assert doc.deployment_tagline == "A model of human flourishing"
|
|
assert len(doc.projects) == 1
|
|
p = doc.projects[0]
|
|
assert (p.id, p.type, p.content_repo, p.visibility) == ("default", "document", "meta", "public")
|
|
assert p.initial_state == "super-draft"
|
|
|
|
|
|
def test_parse_initial_state_defaults_per_type():
|
|
doc = registry.parse_registry(
|
|
"projects:\n - {id: a, name: A, type: bdd, content_repo: a}\n"
|
|
)
|
|
assert doc.projects[0].initial_state == "active" # bdd default
|
|
|
|
|
|
@pytest.mark.parametrize("bad,msg", [
|
|
("projects: []\n", "at least one"),
|
|
("projects:\n - just-a-string\n", "must be a mapping"),
|
|
("projects:\n - {id: 'Bad Slug', name: A, type: document, content_repo: a}\n", "valid slug"),
|
|
("projects:\n - {id: a, name: A, type: nope, content_repo: a}\n", "invalid type"),
|
|
("projects:\n - {id: a, name: A, type: document}\n", "content_repo"),
|
|
("projects:\n - {id: a, name: A, type: document, content_repo: a, visibility: x}\n", "visibility"),
|
|
("projects:\n - {id: a, name: A, type: document, content_repo: a}\n - {id: a, name: B, type: document, content_repo: b}\n", "duplicate"),
|
|
])
|
|
def test_parse_rejects_invalid(bad, msg):
|
|
with pytest.raises(registry.RegistryError) as e:
|
|
registry.parse_registry(bad)
|
|
assert msg in str(e.value)
|
|
|
|
|
|
def test_apply_upserts_projects_and_deployment():
|
|
_db()
|
|
doc = registry.parse_registry(VALID)
|
|
registry.apply_registry(doc, registry_sha="regsha1", default_id="default")
|
|
# §22 three-tier: the project carries the grouping-tier fields; the
|
|
# per-corpus type/initial_state live on its default collection.
|
|
prow = db.conn().execute(
|
|
"SELECT name, content_repo, visibility, registry_sha FROM projects WHERE id='default'"
|
|
).fetchone()
|
|
assert prow["name"] == "Open Human Model"
|
|
assert prow["content_repo"] == "meta"
|
|
assert prow["registry_sha"] == "regsha1"
|
|
crow = db.conn().execute(
|
|
"SELECT type, initial_state FROM collections WHERE id='default'"
|
|
).fetchone()
|
|
assert crow["type"] == "document"
|
|
assert crow["initial_state"] == "super-draft"
|
|
drow = db.conn().execute("SELECT name, tagline FROM deployment WHERE id=1").fetchone()
|
|
assert drow["name"] == "Open Human Model"
|
|
assert drow["tagline"] == "A model of human flourishing"
|
|
|
|
|
|
def test_apply_rejects_type_change_on_existing_project():
|
|
_db()
|
|
registry.apply_registry(registry.parse_registry(VALID), "s1", default_id="default")
|
|
changed = VALID.replace("type: document", "type: specification")
|
|
registry.apply_registry(registry.parse_registry(changed), "s2", default_id="default") # skipped
|
|
# §22.4a immutable type — now enforced on the collection.
|
|
t = db.conn().execute("SELECT type FROM collections WHERE id='default'").fetchone()["type"]
|
|
assert t == "document" # immutable — unchanged
|
|
# The deployment row IS still advanced even though the type change was skipped.
|
|
drow = db.conn().execute("SELECT registry_sha FROM deployment WHERE id=1").fetchone()
|
|
assert drow["registry_sha"] == "s2"
|