v0.48.0 — SLICE-2: collection field schema + central validation (§22.4a)
Collections declare a `fields:` schema in `.collection.yaml`; entries carry typed metadata (enum/tags/text). New central `metadata_schema` module parses the schema leniently (INV-3) and validates entry values — advisory at read (corpus mirror flags violations as `metadata_malformed` without hard-failing), the enforcement point for the write boundary (edit endpoints land SLICE-4/5). - app/metadata_schema.py: parse_fields (lenient/normalizing) + validate - registry.parse_collection_manifest reads `fields:` into collection config - collections.get_collection unpacks `fields`; served by the collection API - cache._refresh_collection_corpus validates each entry advisory-only Non-breaking, opt-in: no `fields:` → unchanged (INV-5); default `document` collection declares none (N=1 unchanged). No DB migration — schema rides in collections.config_json. SLICE-2 of docs/design/2026-06-06-configurable-collection-metadata.md §7.2. Backend suite green (601 passed). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -2,6 +2,7 @@
|
||||
subfolder_of."""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
@@ -62,3 +63,23 @@ def test_get_collection_and_subfolder():
|
||||
assert collections_mod.subfolder_of("features") == "features"
|
||||
assert collections_mod.subfolder_of("default") == ""
|
||||
assert collections_mod.get_collection("nope") is None
|
||||
|
||||
|
||||
# ---- §22.4a SLICE-2: field schema served on the collection ----
|
||||
|
||||
def test_get_collection_fields_none_when_unset():
|
||||
# INV-5: a collection with no `fields:` exposes fields=None (the default
|
||||
# `document` collection sees zero change).
|
||||
_db()
|
||||
_seed()
|
||||
assert collections_mod.get_collection("features")["fields"] is None
|
||||
|
||||
|
||||
def test_get_collection_exposes_field_schema():
|
||||
_db()
|
||||
_seed()
|
||||
schema = {"priority": {"type": "enum", "values": ["P0", "P1"]}}
|
||||
db.conn().execute(
|
||||
"UPDATE collections SET config_json = ? WHERE id = 'features'",
|
||||
(json.dumps({"fields": schema}),))
|
||||
assert collections_mod.get_collection("features")["fields"] == schema
|
||||
|
||||
@@ -61,6 +61,36 @@ def test_parse_collection_manifest_rejects_bad_visibility():
|
||||
registry.parse_collection_manifest("type: bdd\nvisibility: nope\n")
|
||||
|
||||
|
||||
# ---- §22.4a SLICE-2: a `fields:` block flows into the collection config ----
|
||||
|
||||
def test_parse_collection_manifest_reads_field_schema():
|
||||
doc = registry.parse_collection_manifest(
|
||||
"type: bdd\n"
|
||||
"fields:\n"
|
||||
" priority:\n"
|
||||
" type: enum\n"
|
||||
" values: [P0, P1, P2]\n"
|
||||
" tags:\n"
|
||||
" type: tags\n"
|
||||
)
|
||||
assert doc.config["fields"] == {
|
||||
"priority": {"type": "enum", "values": ["P0", "P1", "P2"]},
|
||||
"tags": {"type": "tags"},
|
||||
}
|
||||
|
||||
|
||||
def test_parse_collection_manifest_lenient_on_bad_field():
|
||||
# A bad field def is skipped (INV-3), the manifest still parses, and a
|
||||
# manifest with no surviving fields carries no `fields` config key at all.
|
||||
doc = registry.parse_collection_manifest(
|
||||
"type: bdd\n"
|
||||
"fields:\n"
|
||||
" broken:\n"
|
||||
" type: ref\n"
|
||||
)
|
||||
assert "fields" not in doc.config
|
||||
|
||||
|
||||
# --- mirror discovery ---------------------------------------------------------
|
||||
|
||||
|
||||
|
||||
@@ -6,6 +6,7 @@ Per docs/design/2026-06-06-configurable-collection-metadata.md §6.2-6.3.
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
@@ -105,6 +106,61 @@ def test_malformed_sidecar_on_migrated_entry_still_loads_flagged(app_with_fake_g
|
||||
assert row["body"].strip() == "Just the body, no frontmatter."
|
||||
|
||||
|
||||
def _set_default_fields_schema(schema):
|
||||
# apply_registry leaves the default collection's config_json untouched, so a
|
||||
# schema set here survives a corpus refresh (refresh_meta_repo only).
|
||||
db.conn().execute(
|
||||
"UPDATE collections SET config_json = ? WHERE id = 'default'",
|
||||
(json.dumps({"fields": schema}),))
|
||||
|
||||
|
||||
# ---- §22.4a SLICE-2: advisory schema validation at ingest (INV-3) ----
|
||||
|
||||
def test_schema_violation_flags_malformed(app_with_fake_gitea):
|
||||
app, fake = app_with_fake_gitea
|
||||
with TestClient(app):
|
||||
_set_default_fields_schema(
|
||||
{"priority": {"type": "enum", "values": ["P0", "P1", "P2"]}})
|
||||
fake.files[("wiggleverse", "meta", "main", "rfcs/bad-prio.md")] = {
|
||||
"content": "---\nslug: bad-prio\ntitle: Bad\nstate: active\npriority: P9\n---\n\nB.\n",
|
||||
"sha": "bp1"}
|
||||
fake.files[("wiggleverse", "meta", "main", "rfcs/good-prio.md")] = {
|
||||
"content": "---\nslug: good-prio\ntitle: Good\nstate: active\npriority: P0\n---\n\nB.\n",
|
||||
"sha": "gp1"}
|
||||
_refresh()
|
||||
|
||||
assert _row("bad-prio")["metadata_malformed"] == 1 # INV-3: flagged
|
||||
assert _row("bad-prio")["title"] == "Bad" # still loads
|
||||
assert _row("good-prio")["metadata_malformed"] == 0
|
||||
|
||||
|
||||
def test_schema_ignores_undeclared_keys(app_with_fake_gitea):
|
||||
# INV-7: keys the schema doesn't declare ride along and never flag malformed.
|
||||
app, fake = app_with_fake_gitea
|
||||
with TestClient(app):
|
||||
_set_default_fields_schema(
|
||||
{"priority": {"type": "enum", "values": ["P0", "P1"]}})
|
||||
fake.files[("wiggleverse", "meta", "main", "rfcs/extra-key.md")] = {
|
||||
"content": "---\nslug: extra-key\ntitle: Extra\nstate: active\nowner: hasan\n---\n\nB.\n",
|
||||
"sha": "ek1"}
|
||||
_refresh()
|
||||
|
||||
assert _row("extra-key")["metadata_malformed"] == 0
|
||||
|
||||
|
||||
def test_no_schema_never_flags(app_with_fake_gitea):
|
||||
# INV-5: a collection with no field schema validates nothing, even when an
|
||||
# entry carries values that would fail a schema if one existed.
|
||||
app, fake = app_with_fake_gitea
|
||||
with TestClient(app):
|
||||
fake.files[("wiggleverse", "meta", "main", "rfcs/anything.md")] = {
|
||||
"content": "---\nslug: anything\ntitle: Any\nstate: active\npriority: whatever\n---\n\nB.\n",
|
||||
"sha": "an1"}
|
||||
_refresh()
|
||||
|
||||
assert _row("anything")["metadata_malformed"] == 0
|
||||
|
||||
|
||||
def test_legacy_collection_without_sidecars_unchanged(app_with_fake_gitea):
|
||||
app, fake = app_with_fake_gitea
|
||||
with TestClient(app):
|
||||
|
||||
@@ -0,0 +1,149 @@
|
||||
"""SLICE-2 unit tests — collection field schema + central validation.
|
||||
|
||||
Pure functions only (no DB / no Gitea). Per
|
||||
docs/design/2026-06-06-configurable-collection-metadata.md §7.2 (SLICE-2):
|
||||
`metadata_schema.parse_fields` (lenient schema parsing) and
|
||||
`metadata_schema.validate` (advisory at read / enforcement point at write).
|
||||
Honors INV-3 (never hard-fails), INV-5 (no-fields unchanged), INV-7 (undeclared
|
||||
keys ride along).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from app import metadata_schema as ms
|
||||
|
||||
|
||||
# ---- parse_fields: normalization + leniency ----
|
||||
|
||||
def test_parse_fields_each_type():
|
||||
raw = {
|
||||
"priority": {"type": "enum", "values": ["P0", "P1", "P2"], "label": "Priority"},
|
||||
"tags": {"type": "tags"},
|
||||
"owner": {"type": "text"},
|
||||
}
|
||||
fields = ms.parse_fields(raw)
|
||||
assert list(fields) == ["priority", "tags", "owner"] # order preserved
|
||||
assert fields["priority"] == {
|
||||
"type": "enum",
|
||||
"values": ["P0", "P1", "P2"],
|
||||
"label": "Priority",
|
||||
}
|
||||
assert fields["tags"] == {"type": "tags"}
|
||||
assert fields["owner"] == {"type": "text"}
|
||||
|
||||
|
||||
def test_parse_fields_controlled_tags_keeps_values():
|
||||
fields = ms.parse_fields({"area": {"type": "tags", "values": ["a", "b"]}})
|
||||
assert fields["area"] == {"type": "tags", "values": ["a", "b"]}
|
||||
|
||||
|
||||
def test_parse_fields_missing_block_is_empty():
|
||||
assert ms.parse_fields(None) == {}
|
||||
assert ms.parse_fields({}) == {}
|
||||
|
||||
|
||||
def test_parse_fields_non_mapping_block_skipped():
|
||||
assert ms.parse_fields(["not", "a", "mapping"]) == {}
|
||||
assert ms.parse_fields("nope") == {}
|
||||
|
||||
|
||||
def test_parse_fields_enum_without_values_skipped():
|
||||
# enum requires a non-empty values list — skipped, not fatal.
|
||||
assert ms.parse_fields({"p": {"type": "enum"}}) == {}
|
||||
assert ms.parse_fields({"p": {"type": "enum", "values": []}}) == {}
|
||||
|
||||
|
||||
def test_parse_fields_unknown_type_skipped():
|
||||
fields = ms.parse_fields(
|
||||
{"good": {"type": "text"}, "bad": {"type": "ref"}, "huh": {"type": "frob"}}
|
||||
)
|
||||
assert list(fields) == ["good"]
|
||||
|
||||
|
||||
def test_parse_fields_non_mapping_def_skipped():
|
||||
fields = ms.parse_fields({"good": {"type": "text"}, "bad": "scalar"})
|
||||
assert list(fields) == ["good"]
|
||||
|
||||
|
||||
def test_parse_fields_values_coerced_to_str_list():
|
||||
fields = ms.parse_fields({"p": {"type": "enum", "values": [0, 1, 2]}})
|
||||
assert fields["p"]["values"] == ["0", "1", "2"]
|
||||
|
||||
|
||||
# ---- validate: advisory problem reporting ----
|
||||
|
||||
SCHEMA = ms.parse_fields(
|
||||
{
|
||||
"priority": {"type": "enum", "values": ["P0", "P1", "P2"]},
|
||||
"tags": {"type": "tags"},
|
||||
"area": {"type": "tags", "values": ["checkout", "cart"]},
|
||||
"owner": {"type": "text"},
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def test_validate_happy():
|
||||
values = {
|
||||
"priority": "P0",
|
||||
"tags": ["anything", "free"],
|
||||
"area": ["checkout"],
|
||||
"owner": "ben",
|
||||
}
|
||||
assert ms.validate(values, SCHEMA) == []
|
||||
|
||||
|
||||
def test_validate_absent_fields_ok():
|
||||
# A declared field that the entry omits is fine (no required fields in v1).
|
||||
assert ms.validate({}, SCHEMA) == []
|
||||
|
||||
|
||||
def test_validate_enum_bad_value():
|
||||
problems = ms.validate({"priority": "P9"}, SCHEMA)
|
||||
assert [p.field for p in problems] == ["priority"]
|
||||
assert problems[0].code == "not-in-values"
|
||||
|
||||
|
||||
def test_validate_enum_wrong_type():
|
||||
problems = ms.validate({"priority": ["P0"]}, SCHEMA)
|
||||
assert problems[0].field == "priority"
|
||||
assert problems[0].code == "wrong-type"
|
||||
|
||||
|
||||
def test_validate_tags_free_form_ok():
|
||||
assert ms.validate({"tags": ["x", "y", "z"]}, SCHEMA) == []
|
||||
|
||||
|
||||
def test_validate_tags_wrong_type():
|
||||
problems = ms.validate({"tags": "notalist"}, SCHEMA)
|
||||
assert problems[0].field == "tags"
|
||||
assert problems[0].code == "wrong-type"
|
||||
|
||||
|
||||
def test_validate_controlled_tags_bad_member():
|
||||
problems = ms.validate({"area": ["checkout", "nope"]}, SCHEMA)
|
||||
assert problems[0].field == "area"
|
||||
assert problems[0].code == "not-in-values"
|
||||
|
||||
|
||||
def test_validate_text_wrong_type():
|
||||
problems = ms.validate({"owner": ["a", "b"]}, SCHEMA)
|
||||
assert problems[0].field == "owner"
|
||||
assert problems[0].code == "wrong-type"
|
||||
|
||||
|
||||
def test_validate_undeclared_keys_ignored():
|
||||
# INV-7: keys outside the schema ride along untouched, never flagged.
|
||||
assert ms.validate({"random": "value", "slug": "x", "title": "y"}, SCHEMA) == []
|
||||
|
||||
|
||||
def test_validate_empty_schema_no_problems():
|
||||
# INV-5: a collection with no fields validates everything as clean.
|
||||
assert ms.validate({"priority": "anything", "x": 1}, {}) == []
|
||||
|
||||
|
||||
def test_problem_as_dict():
|
||||
p = ms.Problem(field="priority", code="not-in-values", message="bad")
|
||||
assert p.as_dict() == {
|
||||
"field": "priority",
|
||||
"code": "not-in-values",
|
||||
"message": "bad",
|
||||
}
|
||||
Reference in New Issue
Block a user