140 lines
7.9 KiB
Python
140 lines
7.9 KiB
Python
"""Small deterministic checks for passive readiness evidence, without a database."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from datetime import datetime
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from hockey_data import broadcast_readiness as readiness
|
|
|
|
|
|
def hub(generation=2):
|
|
return SimpleNamespace(_mapping_generations={"DEV-ONE": generation}, _broadcast_mapping_results={})
|
|
|
|
|
|
def result(**overrides):
|
|
return {"ok": True, "match_id": "match-A", "assignment_id": "opaque-assignment",
|
|
"profile_id": 9, "profile_version": 3, "applied": 2, "rules_applied": 1,
|
|
"unchanged": 4, "rules_unchanged": 1, "errors": [], "missing": [],
|
|
"rule_missing": [], "warnings": [], **overrides}
|
|
|
|
|
|
def test_full_result_counts_confirmed_cache_without_retaining_values_or_queries():
|
|
state = hub()
|
|
readiness.record_mapping_result(state, "DEV-ONE", 2, result(score_snapshot={"private": "secret"},
|
|
sql_text="secret-sql", value="private-value"), full=True)
|
|
stored = state._broadcast_mapping_results["DEV-ONE"]
|
|
assert stored["confirmed"] == 8 and stored["outcome"] == "ready" and stored["full_profile"] is True
|
|
assert stored["completed_at"].endswith("Z")
|
|
assert "secret" not in repr(stored) and "private-value" not in repr(stored)
|
|
|
|
|
|
@pytest.mark.parametrize("missing", ["errors", "missing", "rule_missing", "warnings"])
|
|
def test_partial_incomplete_result_demotes_full_success_and_partial_success_cannot_restore_it(missing):
|
|
state = hub()
|
|
readiness.record_mapping_result(state, "DEV-ONE", 2, result(), full=True)
|
|
readiness.record_mapping_result(state, "DEV-ONE", 2, result(**{missing: [{"error": "private-sql-error"}]}), full=False)
|
|
failed = dict(state._broadcast_mapping_results["DEV-ONE"])
|
|
assert failed["outcome"] == "error" and failed["errors_count"] == 1 and failed["full_profile"] is False
|
|
assert "private-sql-error" not in repr(failed)
|
|
readiness.record_mapping_result(state, "DEV-ONE", 2, result(), full=False)
|
|
assert state._broadcast_mapping_results["DEV-ONE"] == failed
|
|
readiness.record_mapping_result(state, "DEV-ONE", 2, result(), full=True)
|
|
assert state._broadcast_mapping_results["DEV-ONE"]["outcome"] == "ready"
|
|
|
|
|
|
def test_late_old_generation_success_or_failure_cannot_change_current_evidence():
|
|
state = hub()
|
|
readiness.record_mapping_result(state, "DEV-ONE", 2, result(), full=True)
|
|
current = dict(state._broadcast_mapping_results["DEV-ONE"])
|
|
readiness.begin_mapping_attempt(state, "DEV-ONE", 1, full=True)
|
|
readiness.record_mapping_result(state, "DEV-ONE", 1, result(ok=False), full=False)
|
|
readiness.record_mapping_exception(state, "DEV-ONE", 1, full=False)
|
|
readiness.record_mapping_result(state, "DEV-ONE", 1, result(profile_version=1), full=True)
|
|
assert state._broadcast_mapping_results["DEV-ONE"] == current
|
|
|
|
|
|
def test_interrupted_full_apply_cannot_reuse_old_success_and_partial_exception_demotes():
|
|
state = hub()
|
|
readiness.record_mapping_result(state, "DEV-ONE", 2, result(), full=True)
|
|
readiness.record_mapping_exception(state, "DEV-ONE", 2, full=False)
|
|
assert state._broadcast_mapping_results["DEV-ONE"]["outcome"] == "error"
|
|
readiness.record_mapping_result(state, "DEV-ONE", 2, result(), full=True)
|
|
readiness.begin_mapping_attempt(state, "DEV-ONE", 2, full=True)
|
|
readiness.record_mapping_exception(state, "DEV-ONE", 2, full=True)
|
|
assert not state._broadcast_mapping_results
|
|
|
|
|
|
def test_evidence_is_bounded_and_optional_instrumentation_state_initializes_safely(monkeypatch):
|
|
monkeypatch.setattr(readiness, "MAX_MAPPING_RESULTS", 3)
|
|
state = SimpleNamespace(_mapping_generations={})
|
|
for index in range(4):
|
|
readiness.record_mapping_result(state, f"DEVICE-{index}", 0, result(), full=True)
|
|
assert list(state._broadcast_mapping_results) == ["DEVICE-1", "DEVICE-2", "DEVICE-3"]
|
|
|
|
|
|
def test_target_validation_uses_exact_titles_and_fields_and_rejects_ambiguous_or_positional_refs():
|
|
fields = [{"input": title, "field": name} for title, name in [
|
|
("Valid", "A.Text"), ("Valid", "A.Text"), ("Renamed", "A.Text"),
|
|
("Duplicate", "A.Text"), ("12", "A.Text"), ("Valid", "a.text")]]
|
|
inputs = [{"title": title, "key": "old-key", "number": "12", "fields": [{"name": "A.Text"}]} for title in ("Valid", "Duplicate", "Duplicate")]
|
|
assert readiness._target_counts(fields, inputs) == {"enabled_links": 6, "target_count": 5,
|
|
"matched_count": 1, "missing_count": 2, "ambiguous_count": 1, "invalid_count": 1}
|
|
|
|
|
|
def test_unknown_and_errors_never_become_green_from_partial_positive_checks():
|
|
checks = [readiness._check("selection", "ready", "selected", "ready"),
|
|
readiness._check("delivery", "unknown", "unconfirmed", "unknown"),
|
|
readiness._check("pending", "warning", "pending", "warning")]
|
|
assert readiness._response("DEV-ONE", "match-A", checks, datetime.utcnow())["status"] == "unknown"
|
|
checks.append(readiness._check("targets", "error", "missing", "error"))
|
|
assert readiness._response("DEV-ONE", "match-A", checks, datetime.utcnow())["status"] == "error"
|
|
|
|
|
|
def test_empty_selection_never_loads_metadata_or_falls_back_and_db_failure_is_sanitized(monkeypatch):
|
|
async def forbidden(*_args, **_kwargs):
|
|
raise AssertionError("metadata must not be loaded")
|
|
monkeypatch.setattr(readiness, "run_db", forbidden)
|
|
blank = asyncio.run(readiness.broadcast_readiness(SimpleNamespace(), SimpleNamespace(id="user"), game_id="match-A"))
|
|
assert blank["status"] == "unknown" and blank["device_id"] == ""
|
|
assert blank["limitations"] == {"source_sql_executed": False, "upstream_freshness_checked": False}
|
|
|
|
async def broken(*_args, **_kwargs):
|
|
raise RuntimeError("postgres://credential-secret private SQL")
|
|
monkeypatch.setattr(readiness, "run_db", broken)
|
|
state = SimpleNamespace(normalise_device_id=lambda value: value, _mapping_generations={}, _live={})
|
|
failed = asyncio.run(readiness.broadcast_readiness(state, SimpleNamespace(id="user"), device_id="DEV-ONE", game_id="match-A"))
|
|
assert failed["status"] == "unknown" and failed["checks"][0]["reason"] == "state_unavailable"
|
|
assert "credential-secret" not in repr(failed) and "private SQL" not in repr(failed)
|
|
|
|
|
|
def test_old_configuration_result_cannot_restore_or_demote_new_configuration_evidence():
|
|
state = hub()
|
|
old_token = state._broadcast_configuration_token = object()
|
|
readiness.record_mapping_result(state, "DEV-ONE", 2, result(), full=True, configuration_token=old_token)
|
|
new_token = state._broadcast_configuration_token = object()
|
|
readiness.record_mapping_result(state, "DEV-ONE", 2, result(), full=True, configuration_token=new_token)
|
|
current = dict(state._broadcast_mapping_results["DEV-ONE"])
|
|
readiness.begin_mapping_attempt(state, "DEV-ONE", 2, full=True, configuration_token=old_token)
|
|
readiness.record_mapping_result(state, "DEV-ONE", 2, result(ok=False), full=False, configuration_token=old_token)
|
|
readiness.record_mapping_exception(state, "DEV-ONE", 2, full=False, configuration_token=old_token)
|
|
readiness.record_mapping_result(state, "DEV-ONE", 2, result(), full=True, configuration_token=old_token)
|
|
assert state._broadcast_mapping_results["DEV-ONE"] == current
|
|
assert current["configuration_token"] is new_token
|
|
|
|
|
|
def test_configuration_observer_fences_readiness_without_resetting_transport_cache_or_generation():
|
|
from hockey_data.agent_bridge import VmixAgentHub
|
|
|
|
calls = []
|
|
original_token = object()
|
|
state = SimpleNamespace(_broadcast_configuration_token=original_token,
|
|
_mapping_generations={"DEV-ONE": 2}, _mapping_value_cache={"confirmed": "private-value"},
|
|
invalidate_refresh_plan=lambda: calls.append("refresh"))
|
|
VmixAgentHub._mapping_configuration_changed(state)
|
|
assert state._broadcast_configuration_token is not original_token and calls == ["refresh"]
|
|
assert state._mapping_generations == {"DEV-ONE": 2}
|
|
assert state._mapping_value_cache == {"confirmed": "private-value"}
|