87 lines
3.7 KiB
Python
87 lines
3.7 KiB
Python
"""Static safety checks for the T-012 reconciliation/orphan boundary."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pathlib
|
|
import re
|
|
import unittest
|
|
|
|
|
|
ROOT = pathlib.Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def text(relative: str) -> str:
|
|
return (ROOT / relative).read_text(encoding="utf-8")
|
|
|
|
|
|
def normalized(value: str) -> str:
|
|
return re.sub(r"\s+", " ", value.lower())
|
|
|
|
|
|
class SenseReconcileSafetyTests(unittest.TestCase):
|
|
def test_postgres_v5_has_fencing_ownership_and_private_reports(self) -> None:
|
|
schema = normalized(text("deploy/postgres/010_reconcile_safety.sql"))
|
|
privileges = normalized(
|
|
text("deploy/postgres/011_privileges_reconcile_safety.sql")
|
|
)
|
|
implementation = normalized(text("Sense/internal/store/postgres.go"))
|
|
for table in (
|
|
"media_path_ownership",
|
|
"operational_leases",
|
|
"orphan_scan_runs",
|
|
"orphan_scan_findings",
|
|
"orphan_cleanup_actions",
|
|
):
|
|
self.assertIn(f"sense.{table}", schema)
|
|
self.assertIn(f"sense.{table}", privileges)
|
|
self.assertIn("insert into sense.schema_migrations(version) values (5)", schema)
|
|
self.assertIn("clock_timestamp()", implementation)
|
|
self.assertIn("for update of r skip locked", implementation)
|
|
self.assertIn("and lease_token = $5", implementation)
|
|
self.assertNotIn("endpoint_ref", schema)
|
|
self.assertNotIn("credential_ref", schema)
|
|
self.assertNotIn("source_uri", schema)
|
|
|
|
def test_cleanup_gate_has_no_force_or_unowned_delete_path(self) -> None:
|
|
source = text("Sense/internal/orphan/orphan.go")
|
|
normalized_source = normalized(source)
|
|
self.assertIn("candidates*100 > observed*10", normalized_source)
|
|
self.assertIn("candidates > maxcleanupitems", normalized_source)
|
|
self.assertIn('confirmation != "delete "+scanid', normalized_source)
|
|
self.assertIn(
|
|
"finding.classification != store.orphanownedstale", normalized_source
|
|
)
|
|
self.assertNotRegex(normalized_source, r"\bforce\b|bypass")
|
|
migration = normalized(text("deploy/postgres/010_reconcile_safety.sql"))
|
|
self.assertIn("classification = 'owned_stale'", migration)
|
|
|
|
def test_mediamtx_inventory_discards_sources_and_bounds_pagination(self) -> None:
|
|
source = text("Sense/internal/mtx/client.go")
|
|
self.assertIn("ListPathNames", source)
|
|
self.assertIn("maxPages", source)
|
|
self.assertIn("list paths repeated page", source)
|
|
inventory = source[source.index("func (c *Client) ListPathNames") :]
|
|
inventory = inventory[: inventory.index("func (c *Client) CreatePath")]
|
|
self.assertNotIn("item.Source", inventory)
|
|
|
|
def test_metrics_are_fixed_and_control_contract_is_untouched(self) -> None:
|
|
metrics = text("Sense/internal/metrics/metrics.go")
|
|
main = text("Sense/cmd/sense-api/main.go")
|
|
self.assertIn('mux.Handle("GET /metrics", registry.Handler())', main)
|
|
for forbidden in ("tenant_id", "site_id", "device_id", "path_name"):
|
|
self.assertNotIn(forbidden, metrics)
|
|
control = text("docs/contracts/sense-control-v1.openapi.json")
|
|
self.assertEqual(7, len(re.findall(r'"operationId"\s*:', control)))
|
|
|
|
def test_operator_command_requires_fresh_scan_confirmation(self) -> None:
|
|
command = text("Sense/cmd/sense-orphan/main.go")
|
|
manager = text("Sense/internal/orphan/orphan.go")
|
|
self.assertIn("exact confirmation: DELETE <scan-id>", command)
|
|
self.assertIn("scan.ExpiresAt.After(now)", manager)
|
|
self.assertIn("cleanupCandidates(scan, paths, ownership)", manager)
|
|
self.assertIn("RecordOrphanCleanup", manager)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|