319 lines
14 KiB
Python
319 lines
14 KiB
Python
import copy
|
|
import json
|
|
import re
|
|
import unittest
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
|
|
TEST_DIR = Path(__file__).resolve().parent
|
|
CONTRACT_DIR = TEST_DIR.parents[1] / "runtime-status" / "v1"
|
|
SCHEMA = json.loads((CONTRACT_DIR / "runtime-status.schema.json").read_text(encoding="utf-8"))
|
|
VALID_DIR = CONTRACT_DIR / "examples" / "valid"
|
|
INVALID_DIR = CONTRACT_DIR / "examples" / "invalid"
|
|
|
|
|
|
def parse_datetime(value):
|
|
if not isinstance(value, str):
|
|
raise ValueError("not a string")
|
|
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
if parsed.tzinfo is None:
|
|
raise ValueError("timezone is required")
|
|
return parsed.astimezone(timezone.utc)
|
|
|
|
|
|
def matches_type(value, expected):
|
|
if expected == "null":
|
|
return value is None
|
|
if expected == "object":
|
|
return isinstance(value, dict)
|
|
if expected == "array":
|
|
return isinstance(value, list)
|
|
if expected == "string":
|
|
return isinstance(value, str)
|
|
if expected == "integer":
|
|
return isinstance(value, int) and not isinstance(value, bool)
|
|
if expected == "number":
|
|
return isinstance(value, (int, float)) and not isinstance(value, bool)
|
|
if expected == "boolean":
|
|
return isinstance(value, bool)
|
|
raise AssertionError(f"unsupported schema type in test validator: {expected}")
|
|
|
|
|
|
def resolve_ref(ref):
|
|
if not ref.startswith("#/"):
|
|
raise AssertionError(f"external refs are not supported: {ref}")
|
|
node = SCHEMA
|
|
for part in ref[2:].split("/"):
|
|
node = node[part.replace("~1", "/").replace("~0", "~")]
|
|
return node
|
|
|
|
|
|
def validate(instance, schema=None, path="$", errors=None):
|
|
schema = SCHEMA if schema is None else schema
|
|
errors = [] if errors is None else errors
|
|
if "$ref" in schema:
|
|
return validate(instance, resolve_ref(schema["$ref"]), path, errors)
|
|
|
|
for subschema in schema.get("allOf", []):
|
|
validate(instance, subschema, path, errors)
|
|
if "if" in schema:
|
|
condition_errors = validate(instance, schema["if"], path, [])
|
|
branch = schema.get("then") if not condition_errors else schema.get("else")
|
|
if branch is not None:
|
|
validate(instance, branch, path, errors)
|
|
|
|
if "type" in schema:
|
|
allowed = schema["type"] if isinstance(schema["type"], list) else [schema["type"]]
|
|
if not any(matches_type(instance, expected) for expected in allowed):
|
|
errors.append(f"{path}: expected {allowed}")
|
|
return errors
|
|
|
|
if "const" in schema and instance != schema["const"]:
|
|
errors.append(f"{path}: expected constant {schema['const']!r}")
|
|
if "enum" in schema and instance not in schema["enum"]:
|
|
errors.append(f"{path}: value is not in enum")
|
|
|
|
if isinstance(instance, dict):
|
|
required = schema.get("required", [])
|
|
for name in required:
|
|
if name not in instance:
|
|
errors.append(f"{path}: missing required property {name}")
|
|
properties = schema.get("properties", {})
|
|
if schema.get("additionalProperties") is False:
|
|
for name in instance:
|
|
if name not in properties:
|
|
errors.append(f"{path}: additional property {name}")
|
|
for name, value in instance.items():
|
|
if name in properties:
|
|
validate(value, properties[name], f"{path}.{name}", errors)
|
|
|
|
if isinstance(instance, list):
|
|
if "maxItems" in schema and len(instance) > schema["maxItems"]:
|
|
errors.append(f"{path}: too many items")
|
|
if schema.get("uniqueItems"):
|
|
encoded = [json.dumps(item, sort_keys=True) for item in instance]
|
|
if len(encoded) != len(set(encoded)):
|
|
errors.append(f"{path}: duplicate items")
|
|
if "items" in schema:
|
|
for index, value in enumerate(instance):
|
|
validate(value, schema["items"], f"{path}[{index}]", errors)
|
|
|
|
if isinstance(instance, str):
|
|
if "minLength" in schema and len(instance) < schema["minLength"]:
|
|
errors.append(f"{path}: string is too short")
|
|
if "maxLength" in schema and len(instance) > schema["maxLength"]:
|
|
errors.append(f"{path}: string is too long")
|
|
if "pattern" in schema and re.fullmatch(schema["pattern"], instance) is None:
|
|
errors.append(f"{path}: pattern mismatch")
|
|
if schema.get("format") == "date-time":
|
|
try:
|
|
parse_datetime(instance)
|
|
except (TypeError, ValueError):
|
|
errors.append(f"{path}: invalid date-time")
|
|
|
|
if isinstance(instance, (int, float)) and not isinstance(instance, bool):
|
|
if "minimum" in schema and instance < schema["minimum"]:
|
|
errors.append(f"{path}: below minimum")
|
|
if "maximum" in schema and instance > schema["maximum"]:
|
|
errors.append(f"{path}: above maximum")
|
|
return errors
|
|
|
|
|
|
def load(path):
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
|
|
|
|
def validate_contract(instance):
|
|
errors = validate(instance)
|
|
configurations = instance.get("configurations") if isinstance(instance, dict) else None
|
|
if isinstance(configurations, list):
|
|
config_ids = [item.get("config_id") for item in configurations if isinstance(item, dict)]
|
|
duplicates = {config_id for config_id in config_ids if config_ids.count(config_id) > 1}
|
|
if duplicates:
|
|
errors.append(f"$.configurations: duplicate config_id {sorted(duplicates)!r}")
|
|
return errors
|
|
|
|
|
|
def freshness(observed_at, evaluation_time):
|
|
age = evaluation_time - parse_datetime(observed_at)
|
|
if age < timedelta(seconds=-30):
|
|
return "future_rejected"
|
|
return "stale" if age > timedelta(seconds=90) else "fresh"
|
|
|
|
|
|
ALLOWED_TRANSITIONS = {
|
|
"unconfigured": {"starting", "stopped"},
|
|
"starting": {"running", "degraded", "failed", "stopped"},
|
|
"running": {"degraded", "failed", "stopped"},
|
|
"degraded": {"running", "failed", "stopped"},
|
|
"failed": {"starting", "stopped"},
|
|
"stopped": {"starting", "unconfigured"},
|
|
}
|
|
|
|
|
|
def may_transition(previous, current):
|
|
return previous == current or current in ALLOWED_TRANSITIONS[previous]
|
|
|
|
|
|
def may_replace(previous, candidate, evaluation_time):
|
|
if candidate["schema_version"] != "yovision.runtime-status/v1":
|
|
return False
|
|
if validate_contract(candidate):
|
|
return False
|
|
if freshness(candidate["observed_at"], evaluation_time) == "future_rejected":
|
|
return False
|
|
if candidate["brain_instance_ref"] != previous["brain_instance_ref"]:
|
|
return False
|
|
if candidate["sequence"] <= previous["sequence"]:
|
|
return False
|
|
return may_transition(previous["runtime"]["state"], candidate["runtime"]["state"])
|
|
|
|
|
|
class RuntimeStatusV1ContractTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.running = load(VALID_DIR / "running.json")
|
|
|
|
def test_schema_declares_frozen_version_and_closed_objects(self):
|
|
self.assertEqual(SCHEMA["$schema"], "https://json-schema.org/draft/2020-12/schema")
|
|
self.assertEqual(SCHEMA["properties"]["schema_version"]["const"], "yovision.runtime-status/v1")
|
|
self.assertFalse(SCHEMA["additionalProperties"])
|
|
for name in ("runtime", "model", "health"):
|
|
self.assertFalse(SCHEMA["properties"][name]["additionalProperties"])
|
|
self.assertFalse(SCHEMA["properties"]["configurations"]["items"]["additionalProperties"])
|
|
|
|
def test_all_valid_examples_satisfy_schema(self):
|
|
paths = sorted(VALID_DIR.glob("*.json"))
|
|
self.assertGreaterEqual(len(paths), 7)
|
|
for path in paths:
|
|
with self.subTest(path=path.name):
|
|
self.assertEqual(validate_contract(load(path)), [])
|
|
|
|
def test_all_invalid_examples_are_rejected(self):
|
|
paths = sorted(INVALID_DIR.glob("*.json"))
|
|
self.assertGreaterEqual(len(paths), 6)
|
|
for path in paths:
|
|
with self.subTest(path=path.name):
|
|
self.assertNotEqual(validate_contract(load(path)), [])
|
|
|
|
def test_every_runtime_state_is_schema_valid(self):
|
|
for state in ALLOWED_TRANSITIONS:
|
|
message = copy.deepcopy(self.running)
|
|
message["runtime"]["state"] = state
|
|
with self.subTest(state=state):
|
|
self.assertEqual(validate_contract(message), [])
|
|
|
|
def test_state_transition_matrix(self):
|
|
self.assertTrue(may_transition("unconfigured", "starting"))
|
|
self.assertTrue(may_transition("starting", "running"))
|
|
self.assertTrue(may_transition("running", "degraded"))
|
|
self.assertTrue(may_transition("degraded", "running"))
|
|
self.assertTrue(may_transition("running", "failed"))
|
|
self.assertTrue(may_transition("failed", "stopped"))
|
|
self.assertFalse(may_transition("unconfigured", "running"))
|
|
self.assertFalse(may_transition("stopped", "running"))
|
|
|
|
def test_stale_and_future_boundaries(self):
|
|
observed = parse_datetime(self.running["observed_at"])
|
|
self.assertEqual(freshness(self.running["observed_at"], observed + timedelta(seconds=90)), "fresh")
|
|
self.assertEqual(freshness(self.running["observed_at"], observed + timedelta(seconds=91)), "stale")
|
|
self.assertEqual(freshness(self.running["observed_at"], observed - timedelta(seconds=30)), "fresh")
|
|
self.assertEqual(freshness(self.running["observed_at"], observed - timedelta(seconds=31)), "future_rejected")
|
|
|
|
def test_offline_keeps_last_known_and_recovery_replaces_it(self):
|
|
last_known = load(VALID_DIR / "offline-last-known.json")
|
|
evaluation = parse_datetime(last_known["observed_at"]) + timedelta(seconds=180)
|
|
self.assertEqual(freshness(last_known["observed_at"], evaluation), "stale")
|
|
self.assertEqual(last_known["runtime"]["state"], "degraded")
|
|
recovered = load(VALID_DIR / "recovered.json")
|
|
self.assertTrue(may_replace(last_known, recovered, parse_datetime(recovered["observed_at"])))
|
|
|
|
def test_unknown_version_and_out_of_order_do_not_replace_projection(self):
|
|
unknown = load(INVALID_DIR / "unknown-major.json")
|
|
evaluation = parse_datetime(self.running["observed_at"])
|
|
self.assertFalse(may_replace(self.running, unknown, evaluation))
|
|
older = copy.deepcopy(self.running)
|
|
older["sequence"] = self.running["sequence"] - 1
|
|
self.assertFalse(may_replace(self.running, older, evaluation))
|
|
duplicate = load(INVALID_DIR / "duplicate-config-id.json")
|
|
self.assertFalse(may_replace(self.running, duplicate, parse_datetime(duplicate["observed_at"])))
|
|
|
|
def test_configuration_revision_mismatch_is_consumer_derived(self):
|
|
message = load(VALID_DIR / "config-mismatch.json")
|
|
desired_revisions = {"gate-primary": 21}
|
|
self.assertEqual(validate_contract(message), [])
|
|
report = message["configurations"][0]
|
|
self.assertNotEqual(report["applied_revision"], desired_revisions[report["config_id"]])
|
|
self.assertNotIn("desired_revision", report)
|
|
|
|
def test_configuration_apply_state_invariants(self):
|
|
valid_cases = [
|
|
("not_configured", None, None),
|
|
("applying", None, None),
|
|
("applying", 1, None),
|
|
("applied", 1, None),
|
|
("rejected", None, "CONFIG_INVALID"),
|
|
("rejected", 1, "CONFIG_INVALID"),
|
|
]
|
|
for apply_state, revision, error_code in valid_cases:
|
|
message = copy.deepcopy(self.running)
|
|
message["configurations"] = [{
|
|
"config_id": "gate-primary",
|
|
"apply_state": apply_state,
|
|
"applied_revision": revision,
|
|
"error_code": error_code,
|
|
}]
|
|
with self.subTest(apply_state=apply_state):
|
|
self.assertEqual(validate_contract(message), [])
|
|
|
|
invalid_cases = [
|
|
("not_configured", 1, None),
|
|
("applied", None, None),
|
|
("rejected", 1, None),
|
|
]
|
|
for apply_state, revision, error_code in invalid_cases:
|
|
message = copy.deepcopy(self.running)
|
|
message["configurations"] = [{
|
|
"config_id": "gate-primary",
|
|
"apply_state": apply_state,
|
|
"applied_revision": revision,
|
|
"error_code": error_code,
|
|
}]
|
|
with self.subTest(invalid_apply_state=apply_state):
|
|
self.assertNotEqual(validate_contract(message), [])
|
|
|
|
def test_empty_multiple_and_duplicate_configuration_streams(self):
|
|
empty = load(VALID_DIR / "empty-configurations.json")
|
|
multiple = load(VALID_DIR / "configuration-states.json")
|
|
duplicate = load(INVALID_DIR / "duplicate-config-id.json")
|
|
self.assertEqual(validate_contract(empty), [])
|
|
self.assertEqual(validate_contract(multiple), [])
|
|
self.assertEqual(len(multiple["configurations"]), 4)
|
|
self.assertTrue(any("duplicate config_id" in error for error in validate_contract(duplicate)))
|
|
|
|
def test_sensitive_and_business_fields_are_rejected_by_name(self):
|
|
for forbidden in ("access_token", "password", "credential", "internal_path", "stack", "user_session", "video", "face", "alert"):
|
|
message = copy.deepcopy(self.running)
|
|
message[forbidden] = "forbidden"
|
|
with self.subTest(forbidden=forbidden):
|
|
self.assertTrue(any("additional property" in error for error in validate_contract(message)))
|
|
|
|
def test_logical_references_reject_paths(self):
|
|
for value in ("C:\\models\\private.pt", "/srv/models/private.pt", "../private.pt"):
|
|
message = copy.deepcopy(self.running)
|
|
message["model"]["model_ref"] = value
|
|
with self.subTest(value=value):
|
|
self.assertNotEqual(validate_contract(message), [])
|
|
|
|
def test_mapper_responsibilities_are_documented(self):
|
|
mapping = (CONTRACT_DIR / "mapping.md").read_text(encoding="utf-8")
|
|
for field in ("schema_version", "status_id", "brain_instance_ref", "sequence", "observed_at", "runtime.*", "model.*", "configurations[]", "health.*", "inputs[]"):
|
|
self.assertIn(f"`{field}`", mapping)
|
|
self.assertIn("Brain", mapping)
|
|
self.assertIn("Sense", mapping)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|