100 lines
4.5 KiB
Python
100 lines
4.5 KiB
Python
"""Static safety checks for the T-011 Sense Control API implementation."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
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 SenseControlImplementationTests(unittest.TestCase):
|
|
def test_generated_server_covers_all_frozen_operations(self) -> None:
|
|
generated = text("Sense/internal/controlapi/generated.gen.go")
|
|
self.assertIn("Code generated by github.com/oapi-codegen/oapi-codegen/v2", generated)
|
|
for operation in (
|
|
"ListDevices",
|
|
"CreateDevice",
|
|
"GetDevice",
|
|
"UpdateDevice",
|
|
"SetDeviceDesiredState",
|
|
"BatchSetDeviceDesiredState",
|
|
"GetOperation",
|
|
):
|
|
self.assertIn(operation, generated)
|
|
config = text("Sense/internal/controlapi/oapi-codegen.yaml")
|
|
self.assertIn("std-http-server: true", config)
|
|
self.assertIn("./internal/controlapi", text("init.ps1"))
|
|
|
|
def test_control_api_is_feature_gated_to_postgres(self) -> None:
|
|
config = text("Sense/internal/config/config.go")
|
|
main = text("Sense/cmd/sense-api/main.go")
|
|
self.assertIn('SENSE_CONTROL_API_ENABLED', config)
|
|
self.assertIn('Control API requires SENSE_DB_DRIVER=postgres', config)
|
|
self.assertIn('mux.Handle("/api/v1/", controlHandler)', main)
|
|
self.assertNotIn('mux.Handle("/api/v1/"', text("Sense/internal/store/sqlite.go"))
|
|
|
|
def test_static_authentication_stores_only_digests(self) -> None:
|
|
source = text("Sense/internal/auth/auth.go")
|
|
example = json.loads(text("Sense/api/control-auth.example.json"))
|
|
self.assertIn("subtle.ConstantTimeCompare", source)
|
|
self.assertIn("sha256.Sum256([]byte(token))", source)
|
|
self.assertNotIn('json:"token"', source)
|
|
for principal in example["principals"]:
|
|
self.assertRegex(principal["token_sha256"], r"^[0-9a-f]{64}$")
|
|
self.assertNotIn("token", principal.keys() - {"token_sha256"})
|
|
|
|
def test_postgres_receipts_are_hashed_durable_and_at_least_24_hours(self) -> None:
|
|
migration = normalized(text("deploy/postgres/008_control_api.sql"))
|
|
implementation = text("Sense/internal/store/control_postgres.go")
|
|
self.assertIn("create table if not exists sense.control_idempotency_receipts", migration)
|
|
self.assertIn("scope_hash bytea primary key", migration)
|
|
self.assertIn("expires_at >= created_at + interval '24 hours'", migration)
|
|
self.assertNotIn("idempotency_key", migration)
|
|
self.assertIn("pg_advisory_xact_lock", implementation)
|
|
self.assertIn("ConstantTimeCompare", implementation)
|
|
|
|
def test_concurrency_batch_and_limits_are_not_hardcoded_to_16(self) -> None:
|
|
migration = normalized(text("deploy/postgres/008_control_api.sql"))
|
|
server = text("Sense/internal/controlapi/server.go")
|
|
store = text("Sense/internal/store/control_postgres.go")
|
|
self.assertIn("resource_version bigint not null default 1", migration)
|
|
self.assertIn("len(body.Items) > 128", server)
|
|
self.assertIn("limit > 100", server)
|
|
self.assertIn("SAVEPOINT ", store)
|
|
self.assertNotIn("len(body.Items) > 16", server)
|
|
|
|
def test_audit_v1_is_unchanged_and_v2_is_a_strict_superset(self) -> None:
|
|
first = json.loads(text("docs/contracts/sense-device-audit-v1.schema.json"))
|
|
second = json.loads(text("docs/contracts/sense-device-audit-v2.schema.json"))
|
|
v1_events = set(first["properties"]["event_type"]["enum"])
|
|
v2_events = set(second["properties"]["event_type"]["enum"])
|
|
self.assertEqual(
|
|
{"device.created", "device.desired_state.accepted"}, v1_events
|
|
)
|
|
self.assertEqual(v1_events | {"device.configuration.accepted"}, v2_events)
|
|
self.assertIn("configurationData", second["$defs"])
|
|
|
|
def test_disabled_reconciliation_deletes_only_exact_path(self) -> None:
|
|
reconciler = text("Sense/internal/reconcile/reconciler.go")
|
|
media = text("Sense/internal/mtx/client.go")
|
|
self.assertIn("candidate.Device.DesiredState == device.DesiredDisabled", reconciler)
|
|
self.assertIn("r.media.DeletePath(ctx, candidate.Device.PathName)", reconciler)
|
|
self.assertIn("Deletion is an idempotent convergence operation", media)
|
|
self.assertNotIn("ListPaths", reconciler)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|