coulomb/kontextual-engine
2
0
Fork 0
generated from coulomb/repo-seed
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Markitect schema-validation integration use case and fixture for Markdown proxy documents

Browse Source
This commit is contained in:
Bernd Worsch
2026-05-06 04:03:50 +02:00
parent c271385e35
commit dbe93be1a9
16 changed files with 518 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

116
tests/test_asset_registry.py
View File

@@ -16,6 +16,7 @@ from kontextual_engine import (
MetadataFieldDefinition,
MetadataRecord,
MetadataSchema,
MetadataSchemaAssignment,
MetadataValueType,
OperationContext,
PolicyDecision,
@@ -224,6 +225,60 @@ def test_asset_registry_validates_metadata_schema_before_writes() -> None:
assert [record.key for record in repo.list_metadata_records(created.asset.id)] == ["owner"]
def test_asset_registry_applies_persisted_metadata_schema_assignments() -> None:
repo = InMemoryAssetRegistryRepository()
service = AssetRegistryService(repo)
context = operation_context()
schema = MetadataSchema(
schema_id="schema-policy-note-v1",
name="Policy Note Metadata",
allow_unknown=False,
fields=(
MetadataFieldDefinition("owner", MetadataValueType.STRING, required=True, require_confirmed=True),
MetadataFieldDefinition("state", MetadataValueType.STRING, allowed_values=("draft", "approved")),
),
)
assignment = MetadataSchemaAssignment(
assignment_id="assignment-policy-note",
schema_id=schema.schema_id,
asset_types=("policy-note",),
sensitivities=(Sensitivity.INTERNAL,),
policy_ref="local://metadata-policy/policy-note",
)
service.register_metadata_schema(schema, context)
service.assign_metadata_schema(assignment, context)
with pytest.raises(ValidationError) as exc_info:
service.create_asset(
"Policy Note",
Classification(asset_type="policy-note", sensitivity=Sensitivity.INTERNAL),
context,
asset_id="asset-policy-note-invalid",
metadata_records=[MetadataRecord("state", "published")],
)
assert {issue["code"] for issue in exc_info.value.details["issues"]} == {
"metadata.required_missing",
"metadata.value_not_allowed",
}
assert repo.list_assets() == []
created = service.create_asset(
"Policy Note",
Classification(asset_type="policy-note", sensitivity=Sensitivity.INTERNAL),
context,
asset_id="asset-policy-note",
metadata_records=[
MetadataRecord("owner", "Platform Knowledge", confirmed=True),
MetadataRecord("state", "approved", confirmed=True),
],
)
assert created.asset.id == "asset-policy-note"
assert service.list_metadata_schema_assignments()[0].policy_ref == "local://metadata-policy/policy-note"
def test_sqlite_asset_registry_survives_reinstantiation(tmp_path: Path) -> None:
db_path = tmp_path / "registry.sqlite"
repo = SQLiteAssetRegistryRepository(db_path)
@@ -305,6 +360,67 @@ def test_sqlite_registry_persists_context_entities_relationships_and_idempotency
]
def test_sqlite_registry_persists_metadata_schemas_and_assignments(tmp_path: Path) -> None:
db_path = tmp_path / "registry.sqlite"
repo = SQLiteAssetRegistryRepository(db_path)
service = AssetRegistryService(repo)
context = operation_context()
schema = MetadataSchema(
schema_id="schema-review-v1",
name="Review Metadata",
allow_unknown=False,
fields=(
MetadataFieldDefinition("reviewer", MetadataValueType.STRING, required=True, require_confirmed=True),
MetadataFieldDefinition("score", MetadataValueType.NUMBER, min_value=0, max_value=1),
),
)
service.register_metadata_schema(schema, context)
service.assign_metadata_schema(
MetadataSchemaAssignment(
assignment_id="assignment-review-documents",
schema_id=schema.schema_id,
asset_types=("review",),
),
context,
)
reloaded_service = AssetRegistryService(SQLiteAssetRegistryRepository(db_path))
with pytest.raises(ValidationError) as exc_info:
reloaded_service.create_asset(
"Review",
Classification(asset_type="review", sensitivity=Sensitivity.INTERNAL),
context,
asset_id="asset-review-invalid",
metadata_records=[
MetadataRecord("reviewer", "Ada", confirmed=False),
MetadataRecord("score", 1.7),
],
)
assert {issue["code"] for issue in exc_info.value.details["issues"]} == {
"metadata.confirmation_required",
"metadata.value_too_large",
}
created = reloaded_service.create_asset(
"Review",
Classification(asset_type="review", sensitivity=Sensitivity.INTERNAL),
context,
asset_id="asset-review",
metadata_records=[
MetadataRecord("reviewer", "Ada", confirmed=True),
MetadataRecord("score", 0.92),
],
)
reloaded_repo = SQLiteAssetRegistryRepository(db_path)
assert created.asset.id == "asset-review"
assert reloaded_repo.get_metadata_schema("schema-review-v1").name == "Review Metadata"
assert reloaded_repo.get_metadata_schema_assignment("assignment-review-documents").schema_id == "schema-review-v1"
def test_sqlite_registry_enforces_representation_asset_reference(tmp_path: Path) -> None:
repo = SQLiteAssetRegistryRepository(tmp_path / "registry.sqlite")
representation = AssetRepresentation.from_content(