64 lines
2.5 KiB
Python
64 lines
2.5 KiB
Python
|
"""Static contract checks for T-015 Bell immutable event storage."""
|
||
|
|
|
||
|
|
from pathlib import Path
|
||
|
|
import unittest
|
||
|
|
|
||
|
|
|
||
|
|
ROOT = Path(__file__).resolve().parents[1]
|
||
|
|
|
||
|
|
|
||
|
|
def text(path: str) -> str:
|
||
|
|
return (ROOT / path).read_text(encoding="utf-8")
|
||
|
|
|
||
|
|
|
||
|
|
class BellEventContractTests(unittest.TestCase):
|
||
|
|
def test_runtime_schema_copy_matches_frozen_contract(self) -> None:
|
||
|
|
self.assertEqual(
|
||
|
|
(ROOT / "docs/raw/contracts/event-v0.1.schema.json").read_bytes(),
|
||
|
|
(ROOT / "Bell/contracts/event-v0.1.schema.json").read_bytes(),
|
||
|
|
)
|
||
|
|
|
||
|
|
def test_bell_module_freezes_reviewed_dependencies(self) -> None:
|
||
|
|
module = text("Bell/go.mod")
|
||
|
|
self.assertIn("github.com/jackc/pgx/v5 v5.10.0", module)
|
||
|
|
self.assertIn("github.com/oklog/ulid/v2 v2.1.2", module)
|
||
|
|
self.assertIn("github.com/santhosh-tekuri/jsonschema/v6 v6.0.2", module)
|
||
|
|
|
||
|
|
def test_event_migration_is_append_only_and_not_limited_to_sixteen(self) -> None:
|
||
|
|
migration = text("deploy/postgres/012_bell_events.sql").lower()
|
||
|
|
privileges = text("deploy/postgres/013_privileges_bell_events.sql").lower()
|
||
|
|
for marker in (
|
||
|
|
"create table if not exists bell.events",
|
||
|
|
"create table if not exists bell.event_outcomes",
|
||
|
|
"bell.reject_immutable_change",
|
||
|
|
"insert into bell.schema_migrations(version) values (3)",
|
||
|
|
):
|
||
|
|
self.assertIn(marker, migration)
|
||
|
|
self.assertNotIn("limit 16", migration)
|
||
|
|
self.assertIn("grant select, insert on table bell.events, bell.event_outcomes to bell_runtime", privileges)
|
||
|
|
self.assertNotIn("grant update", privileges)
|
||
|
|
self.assertNotIn("grant delete", privileges)
|
||
|
|
self.assertNotIn("grant truncate", privileges)
|
||
|
|
|
||
|
|
def test_factory_owns_id_and_checks_all_semantic_boundaries(self) -> None:
|
||
|
|
source = text("Bell/internal/event/event.go")
|
||
|
|
for marker in (
|
||
|
|
"upstream_id_forbidden",
|
||
|
|
"latency_inconsistent",
|
||
|
|
"confidence_forbidden",
|
||
|
|
"evidence_unsafe",
|
||
|
|
"primary_sensor_invalid",
|
||
|
|
"privacy_unavailable",
|
||
|
|
"privacy_denied",
|
||
|
|
):
|
||
|
|
self.assertIn(marker, source)
|
||
|
|
|
||
|
|
def test_contract_document_no_longer_rejects_non_imaging_empty_snapshot(self) -> None:
|
||
|
|
readme = text("docs/raw/contracts/README.md")
|
||
|
|
self.assertNotIn("`snapshot_uris` 空数组 / 负 latency", readme)
|
||
|
|
self.assertIn("`snapshot_uris` 空数组对非成像事件是合法值", readme)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
unittest.main()
|