diff --git a/civiccore/testing/__init__.py b/civiccore/testing/__init__.py index da6fb91..37e4791 100644 --- a/civiccore/testing/__init__.py +++ b/civiccore/testing/__init__.py @@ -1,16 +1,29 @@ """Reusable test contracts for CivicSuite modules.""" from civiccore.testing.mock_city import ( + DEMO_TOWN_CONTRACT_SCHEMA_VERSION, + DEMO_TOWN_DEFAULT_SEED, + DEMO_TOWN_FIXTURE_ID, + DEMO_TOWN_FIXTURE_SHA256_V1, + DEMO_TOWN_FIXTURE_VERSION, + DEMO_TOWN_GENERATION_MODE, + DEMO_TOWN_NAME, + DEMO_TOWN_PROVENANCE_MODES, + DEMO_TOWN_WATERMARK, MOCK_CITY_CHANGED_SINCE, MOCK_CITY_NAME, MOCK_CITY_STAFF_ROLES, + DemoTownFixtureContract, MockCityBackupRetentionCheck, MockCityBackupRetentionContract, MockCityContractCheck, MockCityIdpCheck, MockCityIdpContract, MockCityVendorContract, + assert_demo_town_fixture_safe, assert_secret_free_report, + demo_town_fixture, + demo_town_fixture_contract, mock_city_backup_retention_contract, mock_city_idp_contract, mock_city_report, @@ -18,19 +31,33 @@ run_mock_city_backup_retention_suite, run_mock_city_contract_suite, run_mock_city_idp_contract_suite, + validate_demo_town_fixture, ) __all__ = [ + "DEMO_TOWN_CONTRACT_SCHEMA_VERSION", + "DEMO_TOWN_DEFAULT_SEED", + "DEMO_TOWN_FIXTURE_ID", + "DEMO_TOWN_FIXTURE_SHA256_V1", + "DEMO_TOWN_FIXTURE_VERSION", + "DEMO_TOWN_GENERATION_MODE", + "DEMO_TOWN_NAME", + "DEMO_TOWN_PROVENANCE_MODES", + "DEMO_TOWN_WATERMARK", "MOCK_CITY_CHANGED_SINCE", "MOCK_CITY_NAME", "MOCK_CITY_STAFF_ROLES", + "DemoTownFixtureContract", "MockCityBackupRetentionCheck", "MockCityBackupRetentionContract", "MockCityContractCheck", "MockCityIdpCheck", "MockCityIdpContract", "MockCityVendorContract", + "assert_demo_town_fixture_safe", "assert_secret_free_report", + "demo_town_fixture", + "demo_town_fixture_contract", "mock_city_backup_retention_contract", "mock_city_idp_contract", "mock_city_report", @@ -38,4 +65,5 @@ "run_mock_city_backup_retention_suite", "run_mock_city_contract_suite", "run_mock_city_idp_contract_suite", + "validate_demo_town_fixture", ] diff --git a/civiccore/testing/mock_city.py b/civiccore/testing/mock_city.py index e2953f1..9ecc9d4 100644 --- a/civiccore/testing/mock_city.py +++ b/civiccore/testing/mock_city.py @@ -2,7 +2,9 @@ from __future__ import annotations +import hashlib import json +import re from dataclasses import dataclass from datetime import UTC, datetime, timedelta from functools import lru_cache @@ -14,7 +16,6 @@ from civiccore.connectors.delta import plan_vendor_delta_request from civiccore.connectors.imports import SUPPORTED_CONNECTORS, import_meeting_payload - MOCK_CITY_NAME = "City of Brookfield" MOCK_CITY_CHANGED_SINCE = datetime(2026, 5, 1, 12, 0, tzinfo=UTC) MOCK_CITY_STAFF_ROLES = frozenset({"clerk_admin", "meeting_editor", "city_attorney"}) @@ -24,6 +25,66 @@ "vendor-gated-contract", } +DEMO_TOWN_CONTRACT_SCHEMA_VERSION = "1.0.0" +DEMO_TOWN_FIXTURE_VERSION = "1.0.0" +DEMO_TOWN_FIXTURE_ID = "redstone-valley-records-demo" +DEMO_TOWN_NAME = "Town of Redstone Valley (Fictional)" +# The v1 generator is intentionally static, not pseudo-random. This value is a +# canonical recipe identifier and is validated with the generation mode and +# golden hash so downstream tools can reject a different fixture selection. +DEMO_TOWN_DEFAULT_SEED = "townlight-records-demo-v1" +DEMO_TOWN_GENERATED_AT = "2026-08-17T00:00:00Z" +DEMO_TOWN_WATERMARK = "SYNTHETIC DEMONSTRATION DATA - NOT A REAL MUNICIPAL RECORD" +DEMO_TOWN_PROVENANCE_MODES = ("fully-synthetic",) +DEMO_TOWN_GENERATION_MODE = "static-canonical-v1" +DEMO_TOWN_FIXTURE_SHA256_V1 = "a9c242a3f2618a69d7effb1d0d17d2df06f6744c8c351bba4065d315c94575b4" +DEMO_TOWN_WORKFLOW_STATUSES = ( + "received", + "assigned", + "searching", + "in_review", + "approved", + "fulfilled", + "closed", +) + +_DEMO_TOWN_PII_PATTERNS = { + "email_address": re.compile(r"\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b", re.IGNORECASE), + "north_american_phone": re.compile( + r"\b(?:\+?1[-. ]?)?(?:\(\d{3}\)|\d{3})[-. ]?\d{3}[-. ]?\d{4}\b" + ), + "postal_address": re.compile( + r"\b\d{1,6}\s+[A-Z0-9][A-Z0-9 .'-]{1,40}\s" + r"(?:STREET|ST|ROAD|RD|AVENUE|AVE|BOULEVARD|BLVD|DRIVE|DR|LANE|LN|COURT|CT|WAY)\b", + re.IGNORECASE, + ), + "social_security_number": re.compile(r"\b\d{3}-\d{2}-\d{4}\b"), +} +_DEMO_TOWN_FORBIDDEN_PERSONAL_FIELDS = { + "address", + "email", + "email_address", + "full_name", + "home_address", + "legal_name", + "mailing_address", + "person_name", + "phone", + "phone_number", + "postal_address", + "requester_name", + "resident_name", + "social_security_number", + "ssn", + "street_address", + "telephone", +} +_DEMO_TOWN_PERSONAL_NAME_FIELD_RE = re.compile( + r"^(?:contact|employee|first|full|last|legal|person|preferred|requester|resident|staff)_name$" +) +_DEMO_TOWN_ID_RE = re.compile(r"^[a-z0-9][a-z0-9-]*$") +_DEMO_TOWN_SHA256_RE = re.compile(r"^[a-f0-9]{64}$") + @dataclass(frozen=True) class MockCityVendorContract: @@ -183,6 +244,624 @@ def public_dict(self) -> dict[str, Any]: } +@dataclass(frozen=True) +class DemoTownFixtureContract: + """Versioned, deterministic and independently authored municipal demo data.""" + + schema_version: str + fixture_id: str + fixture_version: str + deterministic_seed: str + generated_at: str + municipality: dict[str, Any] + sources: tuple[dict[str, Any], ...] + records: tuple[dict[str, Any], ...] + requests: tuple[dict[str, Any], ...] + expected: dict[str, Any] + + def public_dict(self) -> dict[str, Any]: + sources = json.loads(json.dumps(self.sources)) + records = [] + for source_record in json.loads(json.dumps(self.records)): + record = dict(source_record) + record["content_sha256"] = _sha256_text(record["content"]) + records.append(record) + requests = json.loads(json.dumps(self.requests)) + expected = json.loads(json.dumps(self.expected)) + artifact_hashes = { + "sources.json": _sha256_json(sources), + "records.json": _sha256_json(records), + "requests.json": _sha256_json(requests), + "expected.json": _sha256_json(expected), + } + payload = { + "manifest": { + "schema_version": self.schema_version, + "fixture_id": self.fixture_id, + "fixture_version": self.fixture_version, + "deterministic_seed": self.deterministic_seed, + "generation_mode": DEMO_TOWN_GENERATION_MODE, + "generated_at": self.generated_at, + "generator": "civiccore.testing.mock_city", + "municipality": dict(self.municipality), + "synthetic": True, + "watermark": DEMO_TOWN_WATERMARK, + "network_calls": False, + "provenance_modes": list(DEMO_TOWN_PROVENANCE_MODES), + "artifact_hashes": artifact_hashes, + }, + "sources": sources, + "records": records, + "requests": requests, + "expected": expected, + } + payload["fixture_sha256"] = _sha256_json(payload) + return payload + + +def demo_town_fixture_contract() -> DemoTownFixtureContract: + """Return the canonical v1 fictional-town records contract without I/O.""" + + source_id = "townlight-independent-authorship-v1" + request_id = "rv-request-2026-0001" + return DemoTownFixtureContract( + schema_version=DEMO_TOWN_CONTRACT_SCHEMA_VERSION, + fixture_id=DEMO_TOWN_FIXTURE_ID, + fixture_version=DEMO_TOWN_FIXTURE_VERSION, + deterministic_seed=DEMO_TOWN_DEFAULT_SEED, + generated_at=DEMO_TOWN_GENERATED_AT, + municipality={ + "municipality_id": "redstone-valley-fictional", + "name": DEMO_TOWN_NAME, + "state": "CO", + "timezone": "America/Denver", + "population_band": "50,000-100,000", + "fictional": True, + }, + sources=( + { + "source_id": source_id, + "publisher": "Townlight fixture authors", + "canonical_url": None, + "retrieved_at": None, + "content": ( + "Independent authorship source for Redstone Valley fixture version 1.0.0" + ), + "content_sha256": _sha256_text( + "Independent authorship source for Redstone Valley fixture version 1.0.0" + ), + "acquisition_method": "independent-authorship", + "provenance_mode": "fully-synthetic", + "license_or_permission": "Apache-2.0", + "allowed_uses": ["testing", "demonstration", "redistribution"], + "redistributable": True, + "contains_personal_data": False, + "notes": ( + "Project-authored fictional content that is not copied or adapted " + "from any real municipality or media organization." + ), + }, + ), + records=( + { + "record_id": "rv-record-council-0001", + "title": "Council action summary for the demonstration calendar", + "department": "Town Clerk", + "record_type": "meeting-summary", + "created_at": "2026-08-04T20:00:00Z", + "retention_class": "fictional-permanent", + "access_class": "public", + "source_refs": [source_id], + "derivation": "independently-authored", + "synthetic": True, + "watermark": DEMO_TOWN_WATERMARK, + "contains_personal_data": False, + "content": ( + "The fictional council accepted the prior action summary and approved " + "a demonstration-only trail maintenance schedule. No real vote, person, " + "place, ordinance, or meeting is represented." + ), + "ground_truth_tags": ["council", "trail-maintenance", "public"], + }, + { + "record_id": "rv-record-trails-0001", + "title": "Demonstration trail inspection summary", + "department": "Parks and Open Space", + "record_type": "inspection-summary", + "created_at": "2026-08-05T15:30:00Z", + "retention_class": "fictional-operational", + "access_class": "public", + "source_refs": [source_id], + "derivation": "independently-authored", + "synthetic": True, + "watermark": DEMO_TOWN_WATERMARK, + "contains_personal_data": False, + "content": ( + "A synthetic inspection found two demonstration markers needing " + "replacement and one invented drainage segment scheduled for review." + ), + "ground_truth_tags": ["parks", "inspection", "trail-maintenance"], + }, + { + "record_id": "rv-record-water-0001", + "title": "Synthetic utility sampling summary", + "department": "Utilities", + "record_type": "sampling-summary", + "created_at": "2026-08-06T14:15:00Z", + "retention_class": "fictional-operational", + "access_class": "public", + "source_refs": [source_id], + "derivation": "independently-authored", + "synthetic": True, + "watermark": DEMO_TOWN_WATERMARK, + "contains_personal_data": False, + "content": ( + "All invented samples in this software fixture matched its fictional " + "test thresholds. This statement is not regulatory or safety guidance." + ), + "ground_truth_tags": ["utilities", "sampling", "public"], + }, + ), + requests=( + { + "request_id": request_id, + "requester_category": "synthetic-public-requester", + "description": ( + "Provide the fictional trail maintenance schedule and its related " + "demonstration inspection summary." + ), + "received_at": "2026-08-17T16:00:00Z", + "target_record_ids": [ + "rv-record-council-0001", + "rv-record-trails-0001", + ], + "policy_basis": "fictional-demo-policy-v1", + "synthetic": True, + "watermark": DEMO_TOWN_WATERMARK, + "contains_personal_data": False, + }, + ), + expected={ + "search": [ + { + "query": "trail maintenance schedule", + "record_ids": [ + "rv-record-council-0001", + "rv-record-trails-0001", + ], + }, + { + "query": "utility sampling", + "record_ids": ["rv-record-water-0001"], + }, + ], + "pii": { + "expected_findings": [], + "record_ids": [ + "rv-record-council-0001", + "rv-record-trails-0001", + "rv-record-water-0001", + ], + }, + "workflow": { + "request_id": request_id, + "status_sequence": [*DEMO_TOWN_WORKFLOW_STATUSES], + "human_approval_required": True, + }, + "counts": {"sources": 1, "records": 3, "requests": 1}, + }, + ) + + +def demo_town_fixture() -> dict[str, Any]: + """Build the canonical v1 public fixture and verify its safety contract.""" + + fixture = demo_town_fixture_contract().public_dict() + assert_demo_town_fixture_safe(fixture) + return fixture + + +def _closed_object_errors( + value: dict[str, Any], expected_fields: set[str], context: str +) -> list[str]: + errors = [ + f"{context}.{field_name} is required" + for field_name in sorted(expected_fields - set(value)) + ] + errors.extend( + f"{context}.{field_name} is not allowed" + for field_name in sorted(set(value) - expected_fields) + ) + return errors + + +def _reference_list_errors( + value: Any, + valid_ids: set[str], + context: str, + *, + allow_empty: bool = False, +) -> list[str]: + if not isinstance(value, list): + return [f"{context} must be a list"] + errors: list[str] = [] + if not value and not allow_empty: + errors.append(f"{context} must not be empty") + seen: set[str] = set() + for item in value: + if not isinstance(item, str): + errors.append(f"{context} contains a non-string reference") + continue + if item in seen: + errors.append(f"{context} contains duplicate reference {item!r}") + seen.add(item) + if item not in valid_ids: + errors.append(f"{context} contains unknown reference {item!r}") + return errors + + +def validate_demo_town_fixture(fixture: dict[str, Any]) -> tuple[str, ...]: + """Return deterministic contract errors for a candidate demo-town fixture.""" + + errors: list[str] = [] + root_fields = {"manifest", "sources", "records", "requests", "expected", "fixture_sha256"} + errors.extend(_closed_object_errors(fixture, root_fields, "fixture")) + manifest = fixture.get("manifest") + if not isinstance(manifest, dict): + return ("manifest must be an object",) + + manifest_fields = { + "schema_version", + "fixture_id", + "fixture_version", + "deterministic_seed", + "generation_mode", + "generated_at", + "generator", + "municipality", + "synthetic", + "watermark", + "network_calls", + "provenance_modes", + "artifact_hashes", + } + errors.extend(_closed_object_errors(manifest, manifest_fields, "manifest")) + expected_manifest_values = { + "schema_version": DEMO_TOWN_CONTRACT_SCHEMA_VERSION, + "fixture_id": DEMO_TOWN_FIXTURE_ID, + "fixture_version": DEMO_TOWN_FIXTURE_VERSION, + "deterministic_seed": DEMO_TOWN_DEFAULT_SEED, + "generation_mode": DEMO_TOWN_GENERATION_MODE, + "generated_at": DEMO_TOWN_GENERATED_AT, + "generator": "civiccore.testing.mock_city", + "synthetic": True, + "watermark": DEMO_TOWN_WATERMARK, + "network_calls": False, + } + for field_name, expected_value in expected_manifest_values.items(): + if manifest.get(field_name) != expected_value: + errors.append(f"manifest.{field_name} must equal {expected_value!r}") + + municipality = manifest.get("municipality") + expected_municipality = { + "municipality_id": "redstone-valley-fictional", + "name": DEMO_TOWN_NAME, + "state": "CO", + "timezone": "America/Denver", + "population_band": "50,000-100,000", + "fictional": True, + } + if not isinstance(municipality, dict): + errors.append("manifest.municipality must be an object") + elif municipality != expected_municipality: + errors.append("manifest.municipality must equal the canonical fictional municipality") + + if manifest.get("provenance_modes") != list(DEMO_TOWN_PROVENANCE_MODES): + errors.append("manifest.provenance_modes must contain only fully-synthetic for v1") + + sources = fixture.get("sources") + records = fixture.get("records") + requests = fixture.get("requests") + expected = fixture.get("expected") + if not isinstance(sources, list): + errors.append("sources must be a list") + sources = [] + if not isinstance(records, list): + errors.append("records must be a list") + records = [] + if not isinstance(requests, list): + errors.append("requests must be a list") + requests = [] + if not isinstance(expected, dict): + errors.append("expected must be an object") + expected = {} + + source_ids: set[str] = set() + source_fields = { + "source_id", + "publisher", + "canonical_url", + "retrieved_at", + "content", + "content_sha256", + "acquisition_method", + "provenance_mode", + "license_or_permission", + "allowed_uses", + "redistributable", + "contains_personal_data", + "notes", + } + for index, source in enumerate(sources): + prefix = f"sources[{index}]" + if not isinstance(source, dict): + errors.append(f"{prefix} must be an object") + continue + errors.extend(_closed_object_errors(source, source_fields, prefix)) + source_id = source.get("source_id") + if not isinstance(source_id, str) or not _DEMO_TOWN_ID_RE.fullmatch(source_id): + errors.append(f"{prefix}.source_id must be a stable identifier") + elif source_id in source_ids: + errors.append(f"{prefix}.source_id duplicates {source_id!r}") + else: + source_ids.add(source_id) + if source.get("provenance_mode") != "fully-synthetic": + errors.append(f"{prefix}.provenance_mode must be 'fully-synthetic' in v1") + if source.get("acquisition_method") != "independent-authorship": + errors.append(f"{prefix}.acquisition_method must be independent-authorship") + if source.get("contains_personal_data") is not False: + errors.append(f"{prefix} must declare contains_personal_data false") + if source.get("redistributable") is not True: + errors.append(f"{prefix} must be redistributable") + if source.get("canonical_url") is not None: + errors.append(f"{prefix}.canonical_url must be null for fully synthetic content") + if source.get("retrieved_at") is not None: + errors.append(f"{prefix}.retrieved_at must be null for independently authored content") + if not isinstance(source.get("license_or_permission"), str) or not source.get( + "license_or_permission" + ): + errors.append(f"{prefix}.license_or_permission must be explicit") + if source.get("allowed_uses") != ["testing", "demonstration", "redistribution"]: + errors.append(f"{prefix}.allowed_uses must authorize the canonical public fixture uses") + content = source.get("content") + if not isinstance(content, str) or not content: + errors.append(f"{prefix}.content must materialize the independently authored source") + elif source.get("content_sha256") != _sha256_text(content): + errors.append(f"{prefix}.content_sha256 does not match content") + + record_ids: set[str] = set() + record_fields = { + "record_id", + "title", + "department", + "record_type", + "created_at", + "retention_class", + "access_class", + "source_refs", + "derivation", + "synthetic", + "watermark", + "contains_personal_data", + "content", + "content_sha256", + "ground_truth_tags", + } + for index, record in enumerate(records): + prefix = f"records[{index}]" + if not isinstance(record, dict): + errors.append(f"{prefix} must be an object") + continue + errors.extend(_closed_object_errors(record, record_fields, prefix)) + record_id = record.get("record_id") + if not isinstance(record_id, str) or not _DEMO_TOWN_ID_RE.fullmatch(record_id): + errors.append(f"{prefix}.record_id must be a stable identifier") + elif record_id in record_ids: + errors.append(f"{prefix}.record_id duplicates {record_id!r}") + else: + record_ids.add(record_id) + errors.extend( + _reference_list_errors(record.get("source_refs"), source_ids, f"{prefix}.source_refs") + ) + if record.get("synthetic") is not True: + errors.append(f"{prefix} must be explicitly synthetic") + if record.get("watermark") != DEMO_TOWN_WATERMARK: + errors.append(f"{prefix} must carry the canonical synthetic watermark") + if record.get("contains_personal_data") is not False: + errors.append(f"{prefix} must declare contains_personal_data false") + content = record.get("content") + if not isinstance(content, str) or not content: + errors.append(f"{prefix}.content must be non-empty text") + elif record.get("content_sha256") != _sha256_text(content): + errors.append(f"{prefix}.content_sha256 does not match content") + + request_ids: set[str] = set() + request_fields = { + "request_id", + "requester_category", + "description", + "received_at", + "target_record_ids", + "policy_basis", + "synthetic", + "watermark", + "contains_personal_data", + } + for index, request in enumerate(requests): + prefix = f"requests[{index}]" + if not isinstance(request, dict): + errors.append(f"{prefix} must be an object") + continue + errors.extend(_closed_object_errors(request, request_fields, prefix)) + request_id = request.get("request_id") + if not isinstance(request_id, str) or not _DEMO_TOWN_ID_RE.fullmatch(request_id): + errors.append(f"{prefix}.request_id must be a stable identifier") + elif request_id in request_ids: + errors.append(f"{prefix}.request_id duplicates {request_id!r}") + else: + request_ids.add(request_id) + errors.extend( + _reference_list_errors( + request.get("target_record_ids"), record_ids, f"{prefix}.target_record_ids" + ) + ) + if request.get("synthetic") is not True: + errors.append(f"{prefix} must be explicitly synthetic") + if request.get("watermark") != DEMO_TOWN_WATERMARK: + errors.append(f"{prefix} must carry the canonical synthetic watermark") + if request.get("contains_personal_data") is not False: + errors.append(f"{prefix} must declare contains_personal_data false") + + expected_fields = {"search", "pii", "workflow", "counts"} + errors.extend(_closed_object_errors(expected, expected_fields, "expected")) + searches = expected.get("search") + if not isinstance(searches, list): + errors.append("expected.search must be a list") + else: + seen_queries: set[str] = set() + for index, search in enumerate(searches): + prefix = f"expected.search[{index}]" + if not isinstance(search, dict): + errors.append(f"{prefix} must be an object") + continue + errors.extend(_closed_object_errors(search, {"query", "record_ids"}, prefix)) + query = search.get("query") + if not isinstance(query, str) or not query.strip(): + errors.append(f"{prefix}.query must be non-empty") + elif query in seen_queries: + errors.append(f"{prefix}.query duplicates {query!r}") + else: + seen_queries.add(query) + errors.extend( + _reference_list_errors(search.get("record_ids"), record_ids, f"{prefix}.record_ids") + ) + + pii = expected.get("pii") + if not isinstance(pii, dict): + errors.append("expected.pii must be an object") + else: + errors.extend( + _closed_object_errors(pii, {"expected_findings", "record_ids"}, "expected.pii") + ) + if pii.get("expected_findings") != []: + errors.append("expected.pii.expected_findings must be empty for the v1 fixture") + errors.extend( + _reference_list_errors(pii.get("record_ids"), record_ids, "expected.pii.record_ids") + ) + if ( + isinstance(pii.get("record_ids"), list) + and all(isinstance(item, str) for item in pii["record_ids"]) + and set(pii["record_ids"]) != record_ids + ): + errors.append("expected.pii.record_ids must cover every fixture record") + + workflow = expected.get("workflow") + if not isinstance(workflow, dict): + errors.append("expected.workflow must be an object") + else: + errors.extend( + _closed_object_errors( + workflow, + {"request_id", "status_sequence", "human_approval_required"}, + "expected.workflow", + ) + ) + if workflow.get("request_id") not in request_ids: + errors.append("expected.workflow.request_id must reference a fixture request") + if workflow.get("status_sequence") != list(DEMO_TOWN_WORKFLOW_STATUSES): + errors.append("expected.workflow.status_sequence must equal the canonical sequence") + if workflow.get("human_approval_required") is not True: + errors.append("expected.workflow.human_approval_required must be true") + + counts = expected.get("counts") + expected_counts = {"sources": len(sources), "records": len(records), "requests": len(requests)} + if counts != expected_counts: + errors.append(f"expected.counts must equal {expected_counts!r}") + + expected_artifact_hashes = { + "sources.json": _sha256_json(sources), + "records.json": _sha256_json(records), + "requests.json": _sha256_json(requests), + "expected.json": _sha256_json(expected), + } + if manifest.get("artifact_hashes") != expected_artifact_hashes: + errors.append("manifest.artifact_hashes do not match fixture artifacts") + + fixture_without_hash = dict(fixture) + fixture_without_hash.pop("fixture_sha256", None) + if fixture.get("fixture_sha256") != _sha256_json(fixture_without_hash): + errors.append("fixture_sha256 does not match fixture content") + if fixture.get("fixture_sha256") != DEMO_TOWN_FIXTURE_SHA256_V1: + errors.append("fixture_sha256 does not match the pinned v1 golden fixture") + return tuple(errors) + + +def assert_demo_town_fixture_safe(fixture: dict[str, Any]) -> None: + """Apply conservative privacy heuristics and the closed public fixture contract. + + These checks reduce accidental disclosure risk; they do not claim to identify + every possible person name or item of personal data in arbitrary free text. + """ + + forbidden_fields = sorted( + field_name + for field_name in _nested_field_names(fixture) + if _is_forbidden_personal_field(field_name) + ) + if forbidden_fields: + raise ValueError( + "demo town fixture contains forbidden personal-data fields: " + + ", ".join(forbidden_fields) + ) + + serialized = json.dumps(fixture, sort_keys=True, ensure_ascii=False) + pii_matches = sorted( + name for name, pattern in _DEMO_TOWN_PII_PATTERNS.items() if pattern.search(serialized) + ) + if pii_matches: + raise ValueError( + "demo town fixture contains forbidden PII patterns: " + ", ".join(pii_matches) + ) + + assert_secret_free_report(fixture) + errors = validate_demo_town_fixture(fixture) + if errors: + raise ValueError("invalid demo town fixture: " + "; ".join(errors)) + + +def _is_forbidden_personal_field(field_name: str) -> bool: + normalized = re.sub(r"[^a-z0-9]+", "_", field_name.lower()).strip("_") + return normalized in _DEMO_TOWN_FORBIDDEN_PERSONAL_FIELDS or bool( + _DEMO_TOWN_PERSONAL_NAME_FIELD_RE.fullmatch(normalized) + ) + + +def _nested_field_names(value: Any) -> list[str]: + names: list[str] = [] + if isinstance(value, dict): + for key, child in value.items(): + names.append(str(key)) + names.extend(_nested_field_names(child)) + elif isinstance(value, list | tuple): + for child in value: + names.extend(_nested_field_names(child)) + return names + + +def _sha256_json(value: Any) -> str: + encoded = json.dumps( + value, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + ).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +def _sha256_text(value: str) -> str: + return hashlib.sha256(value.encode("utf-8")).hexdigest() + + def mock_city_vendor_contracts() -> list[MockCityVendorContract]: """Return reusable vendor contracts for mock-city integration tests.""" @@ -575,16 +1254,29 @@ def assert_secret_free_report(report: dict[str, Any]) -> None: __all__ = [ + "DEMO_TOWN_CONTRACT_SCHEMA_VERSION", + "DEMO_TOWN_DEFAULT_SEED", + "DEMO_TOWN_FIXTURE_ID", + "DEMO_TOWN_FIXTURE_SHA256_V1", + "DEMO_TOWN_FIXTURE_VERSION", + "DEMO_TOWN_GENERATION_MODE", + "DEMO_TOWN_NAME", + "DEMO_TOWN_PROVENANCE_MODES", + "DEMO_TOWN_WATERMARK", "MOCK_CITY_CHANGED_SINCE", "MOCK_CITY_NAME", "MOCK_CITY_STAFF_ROLES", + "DemoTownFixtureContract", "MockCityBackupRetentionCheck", "MockCityBackupRetentionContract", "MockCityContractCheck", "MockCityIdpCheck", "MockCityIdpContract", "MockCityVendorContract", + "assert_demo_town_fixture_safe", "assert_secret_free_report", + "demo_town_fixture", + "demo_town_fixture_contract", "mock_city_backup_retention_contract", "mock_city_idp_contract", "mock_city_report", @@ -592,4 +1284,5 @@ def assert_secret_free_report(report: dict[str, Any]) -> None: "run_mock_city_backup_retention_suite", "run_mock_city_contract_suite", "run_mock_city_idp_contract_suite", + "validate_demo_town_fixture", ] diff --git a/pyproject.toml b/pyproject.toml index 0a6028d..fd2f888 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -81,7 +81,7 @@ dev = [ "pytest>=9.0.0", "pytest-asyncio>=1.0.0", "respx>=0.22.0", - "ruff>=0.11.0", + "ruff==0.15.22", "testcontainers[postgres]>=4.9.0", # LLM provider SDKs included in dev so test_llm_providers runs without # skips per Hard Rule 4a. End users still install [openai]/[anthropic] diff --git a/tests/test_demo_town_contract.py b/tests/test_demo_town_contract.py new file mode 100644 index 0000000..f905452 --- /dev/null +++ b/tests/test_demo_town_contract.py @@ -0,0 +1,258 @@ +from __future__ import annotations + +import hashlib +import json +import socket +from copy import deepcopy + +import pytest + +from civiccore.testing import ( + DEMO_TOWN_CONTRACT_SCHEMA_VERSION, + DEMO_TOWN_DEFAULT_SEED, + DEMO_TOWN_FIXTURE_ID, + DEMO_TOWN_FIXTURE_SHA256_V1, + DEMO_TOWN_FIXTURE_VERSION, + DEMO_TOWN_GENERATION_MODE, + DEMO_TOWN_NAME, + DEMO_TOWN_PROVENANCE_MODES, + DEMO_TOWN_WATERMARK, + assert_demo_town_fixture_safe, + demo_town_fixture, + demo_town_fixture_contract, + validate_demo_town_fixture, +) + + +def _canonical_sha256(value: object) -> str: + payload = json.dumps( + value, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + ).encode("utf-8") + return hashlib.sha256(payload).hexdigest() + + +def _reseal(fixture: dict[str, object]) -> None: + fixture["manifest"]["artifact_hashes"] = { + artifact_name: _canonical_sha256(fixture[key]) + for artifact_name, key in { + "sources.json": "sources", + "records.json": "records", + "requests.json": "requests", + "expected.json": "expected", + }.items() + } + fixture.pop("fixture_sha256", None) + fixture["fixture_sha256"] = _canonical_sha256(fixture) + + +def test_demo_town_contract_is_versioned_and_explicitly_fictional() -> None: + fixture = demo_town_fixture() + manifest = fixture["manifest"] + + assert manifest["schema_version"] == DEMO_TOWN_CONTRACT_SCHEMA_VERSION == "1.0.0" + assert manifest["fixture_id"] == DEMO_TOWN_FIXTURE_ID + assert manifest["fixture_version"] == DEMO_TOWN_FIXTURE_VERSION == "1.0.0" + assert manifest["deterministic_seed"] == DEMO_TOWN_DEFAULT_SEED + assert manifest["generation_mode"] == DEMO_TOWN_GENERATION_MODE == "static-canonical-v1" + assert manifest["municipality"] == { + "municipality_id": "redstone-valley-fictional", + "name": DEMO_TOWN_NAME, + "state": "CO", + "timezone": "America/Denver", + "population_band": "50,000-100,000", + "fictional": True, + } + assert manifest["synthetic"] is True + assert manifest["watermark"] == DEMO_TOWN_WATERMARK + assert manifest["network_calls"] is False + assert all(record["synthetic"] is True for record in fixture["records"]) + assert all(record["watermark"] == DEMO_TOWN_WATERMARK for record in fixture["records"]) + assert all(request["watermark"] == DEMO_TOWN_WATERMARK for request in fixture["requests"]) + + +def test_demo_town_fixture_and_artifact_hashes_are_deterministic() -> None: + first = demo_town_fixture() + second = demo_town_fixture_contract().public_dict() + + assert first == second + assert validate_demo_town_fixture(first) == () + for artifact_name, value in { + "sources.json": first["sources"], + "records.json": first["records"], + "requests.json": first["requests"], + "expected.json": first["expected"], + }.items(): + assert first["manifest"]["artifact_hashes"][artifact_name] == _canonical_sha256(value) + fixture_without_hash = dict(first) + fixture_without_hash.pop("fixture_sha256") + assert first["fixture_sha256"] == _canonical_sha256(fixture_without_hash) + assert first["fixture_sha256"] == DEMO_TOWN_FIXTURE_SHA256_V1 + assert DEMO_TOWN_FIXTURE_SHA256_V1 == ( + "a9c242a3f2618a69d7effb1d0d17d2df06f6744c8c351bba4065d315c94575b4" + ) + assert len(first["fixture_sha256"]) == 64 + + tampered = deepcopy(first) + tampered["records"][0]["content"] += " tampered" + errors = validate_demo_town_fixture(tampered) + assert "records[0].content_sha256 does not match content" in errors + assert "manifest.artifact_hashes do not match fixture artifacts" in errors + assert "fixture_sha256 does not match fixture content" in errors + + +def test_demo_town_provenance_is_explicit_and_redistributable() -> None: + fixture = demo_town_fixture() + source = fixture["sources"][0] + + assert fixture["manifest"]["provenance_modes"] == list(DEMO_TOWN_PROVENANCE_MODES) + assert DEMO_TOWN_PROVENANCE_MODES == ("fully-synthetic",) + assert source["provenance_mode"] == "fully-synthetic" + assert source["acquisition_method"] == "independent-authorship" + assert source["canonical_url"] is None + assert source["retrieved_at"] is None + assert source["redistributable"] is True + assert source["contains_personal_data"] is False + assert source["allowed_uses"] == ["testing", "demonstration", "redistribution"] + serialized = json.dumps(fixture).lower() + assert "longmont" not in serialized + assert "public media" not in serialized + + +def test_demo_town_defaults_are_secret_and_pii_free() -> None: + fixture = demo_town_fixture() + + assert_demo_town_fixture_safe(fixture) + assert all(record["contains_personal_data"] is False for record in fixture["records"]) + assert all(request["contains_personal_data"] is False for request in fixture["requests"]) + assert fixture["expected"]["pii"]["expected_findings"] == [] + + personal_field = deepcopy(fixture) + personal_field["requests"][0]["email"] = "synthetic-requester" + with pytest.raises(ValueError, match="forbidden personal-data fields: email"): + assert_demo_town_fixture_safe(personal_field) + + pii_pattern = deepcopy(fixture) + pii_pattern["records"][0]["content"] += " contact person@example.invalid" + with pytest.raises(ValueError, match="forbidden PII patterns: email_address"): + assert_demo_town_fixture_safe(pii_pattern) + + personal_identity = deepcopy(fixture) + personal_identity["requests"][0]["requester_name"] = "Jane Doe" + personal_identity["requests"][0]["home_address"] = "123 Main Street" + with pytest.raises( + ValueError, + match="forbidden personal-data fields: home_address, requester_name", + ): + assert_demo_town_fixture_safe(personal_identity) + + address_in_text = deepcopy(fixture) + address_in_text["records"][0]["content"] += " Deliver to 123 Main Street." + with pytest.raises(ValueError, match="forbidden PII patterns: postal_address"): + assert_demo_town_fixture_safe(address_in_text) + + +def test_demo_town_generation_performs_no_network_calls(monkeypatch: pytest.MonkeyPatch) -> None: + attempted_connections: list[object] = [] + + def reject_connection(*args: object, **kwargs: object) -> None: + attempted_connections.append((args, kwargs)) + raise AssertionError("fixture generation attempted a network connection") + + monkeypatch.setattr(socket, "create_connection", reject_connection) + + fixture = demo_town_fixture() + + assert attempted_connections == [] + assert fixture["manifest"]["network_calls"] is False + assert all(source["canonical_url"] is None for source in fixture["sources"]) + + +def test_demo_town_contract_supplies_records_workflow_ground_truth() -> None: + fixture = demo_town_fixture() + + assert fixture["expected"]["counts"] == {"sources": 1, "records": 3, "requests": 1} + assert fixture["expected"]["workflow"] == { + "request_id": "rv-request-2026-0001", + "status_sequence": [ + "received", + "assigned", + "searching", + "in_review", + "approved", + "fulfilled", + "closed", + ], + "human_approval_required": True, + } + assert fixture["expected"]["search"][0]["record_ids"] == [ + "rv-record-council-0001", + "rv-record-trails-0001", + ] + + +def test_demo_town_rejects_unimplemented_provenance_modes() -> None: + fixture = deepcopy(demo_town_fixture()) + source = fixture["sources"][0] + source["provenance_mode"] = "licensed-adaptation" + source.pop("license_or_permission") + source.pop("allowed_uses") + _reseal(fixture) + + errors = validate_demo_town_fixture(fixture) + + assert "sources[0].provenance_mode must be 'fully-synthetic' in v1" in errors + assert "sources[0].license_or_permission is required" in errors + assert "sources[0].allowed_uses is required" in errors + + +def test_demo_town_rejects_duplicate_ids_and_broken_references() -> None: + fixture = deepcopy(demo_town_fixture()) + duplicate = deepcopy(fixture["records"][0]) + duplicate["source_refs"] = ["missing-source"] + fixture["records"].append(duplicate) + fixture["requests"][0]["target_record_ids"] = ["missing-record"] + fixture["expected"]["search"][0]["record_ids"] = ["missing-record"] + fixture["expected"]["counts"] = {"sources": 99, "records": 99, "requests": 99} + _reseal(fixture) + + errors = validate_demo_town_fixture(fixture) + + assert "records[3].record_id duplicates 'rv-record-council-0001'" in errors + assert "records[3].source_refs contains unknown reference 'missing-source'" in errors + assert "requests[0].target_record_ids contains unknown reference 'missing-record'" in errors + assert "expected.search[0].record_ids contains unknown reference 'missing-record'" in errors + assert any(error.startswith("expected.counts must equal") for error in errors) + + +def test_demo_town_rejects_workflow_and_pii_ground_truth_drift() -> None: + fixture = deepcopy(demo_town_fixture()) + fixture["expected"]["workflow"]["request_id"] = "missing-request" + fixture["expected"]["workflow"]["status_sequence"] = ["received", "closed"] + fixture["expected"]["workflow"]["human_approval_required"] = False + fixture["expected"]["pii"]["record_ids"] = ["rv-record-water-0001"] + fixture["expected"]["pii"]["expected_findings"] = ["unexpected-person"] + _reseal(fixture) + + errors = validate_demo_town_fixture(fixture) + + assert "expected.workflow.request_id must reference a fixture request" in errors + assert "expected.workflow.status_sequence must equal the canonical sequence" in errors + assert "expected.workflow.human_approval_required must be true" in errors + assert "expected.pii.expected_findings must be empty for the v1 fixture" in errors + assert "expected.pii.record_ids must cover every fixture record" in errors + + +def test_demo_town_v1_golden_hash_requires_a_versioned_content_change() -> None: + fixture = deepcopy(demo_town_fixture()) + fixture["records"][0]["content"] += " Canonical content drift." + fixture["records"][0]["content_sha256"] = _canonical_sha256( + fixture["records"][0]["content"] + ) + _reseal(fixture) + + errors = validate_demo_town_fixture(fixture) + + assert "fixture_sha256 does not match the pinned v1 golden fixture" in errors