111 lines
4.4 KiB
Python
111 lines
4.4 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from tools.build_api_registry import build as build_api_registry
|
|
from tools.build_architecture_inventory import build as build_architecture_inventory
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
CONFIG = ROOT / "config"
|
|
|
|
|
|
def load(name: str) -> dict:
|
|
return json.loads((CONFIG / name).read_text(encoding="utf-8"))
|
|
|
|
|
|
class GovernanceRegistryTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.features = load("features.config.json")
|
|
self.pages = load("pages.config.json")
|
|
self.api = load("api.config.json")
|
|
self.data = load("data-fields.config.json")
|
|
self.quality = load("data-quality.config.json")
|
|
self.jobs = load("jobs.config.json")
|
|
|
|
def test_features_are_unique_and_use_declared_roles(self) -> None:
|
|
roles = set(self.features["roles"])
|
|
items = self.features["features"]
|
|
ids = [item["id"] for item in items]
|
|
self.assertEqual(len(ids), len(set(ids)))
|
|
self.assertTrue(all(item["access"] in roles for item in items))
|
|
self.assertTrue(all(isinstance(item["enabled"], bool) for item in items))
|
|
|
|
def test_page_registry_matches_the_current_primary_navigation(self) -> None:
|
|
inventory_pages = build_architecture_inventory()["pages"]
|
|
registered = self.pages["pages"]
|
|
self.assertEqual(
|
|
[(item["id"], item["title"]) for item in registered],
|
|
[(item["id"], item["title"]) for item in inventory_pages],
|
|
)
|
|
feature_ids = {item["id"] for item in self.features["features"]}
|
|
self.assertTrue(all(page["feature"] in feature_ids for page in registered))
|
|
self.assertEqual(sum(bool(page.get("default")) for page in registered), 1)
|
|
|
|
def test_api_registry_is_current_and_owned(self) -> None:
|
|
self.assertEqual(self.api, build_api_registry())
|
|
feature_ids = {item["id"] for item in self.features["features"]}
|
|
roles = set(self.features["roles"])
|
|
keys = []
|
|
for route in self.api["routes"]:
|
|
keys.append((route["method"], route["path"], route["match"]))
|
|
self.assertIn(route["feature"], feature_ids)
|
|
self.assertIn(route["access"], roles)
|
|
self.assertEqual(len(keys), len(set(keys)))
|
|
|
|
def test_public_routes_are_explicitly_limited(self) -> None:
|
|
public = {
|
|
(item["method"], item["path"])
|
|
for item in self.api["routes"]
|
|
if item["access"] == "public"
|
|
}
|
|
self.assertEqual(
|
|
public,
|
|
{
|
|
("GET", "/api/health"),
|
|
("POST", "/api/auth/login"),
|
|
("POST", "/api/auth/register"),
|
|
},
|
|
)
|
|
|
|
def test_calculation_datasets_use_approved_providers(self) -> None:
|
|
providers = self.data["providers"]
|
|
ids = []
|
|
for dataset in self.data["datasets"]:
|
|
ids.append(dataset["id"])
|
|
self.assertIn(dataset["primary"], providers)
|
|
for fallback in dataset.get("fallbacks", []):
|
|
self.assertIn(fallback, providers)
|
|
if dataset["usage"] == "calculation":
|
|
self.assertTrue(providers[dataset["primary"]]["calculation_allowed"])
|
|
if dataset["primary"] == "unresolved":
|
|
self.assertEqual(dataset["usage"], "blocked")
|
|
self.assertEqual(len(ids), len(set(ids)))
|
|
|
|
def test_every_dataset_has_one_quality_rule_and_known_unit_profile(self) -> None:
|
|
dataset_ids = {item["id"] for item in self.data["datasets"]}
|
|
rules = self.quality["datasets"]
|
|
self.assertEqual(set(rules), dataset_ids)
|
|
profiles = set(self.quality["unit_profiles"])
|
|
self.assertTrue(
|
|
all(str(rule.get("unit_profile") or "none") in profiles for rule in rules.values())
|
|
)
|
|
|
|
def test_background_jobs_have_complete_unique_runtime_contracts(self) -> None:
|
|
jobs = self.jobs["jobs"]
|
|
ids = [item["id"] for item in jobs]
|
|
self.assertEqual(len(ids), len(set(ids)))
|
|
for item in jobs:
|
|
self.assertTrue(item["schedule"])
|
|
self.assertTrue(item["input_date_policy"])
|
|
self.assertTrue(item["lock_key"])
|
|
self.assertGreater(int(item["timeout_seconds"]), 0)
|
|
self.assertGreater(int(item["max_attempts"]), 0)
|
|
self.assertTrue(item["output_version"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|