77 lines
3.0 KiB
Python
77 lines
3.0 KiB
Python
"""Bounded synthetic per-cell freshness tests, with no model/source dependency."""
|
|
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from k1link.perception.costmap_freshness import CostmapCellEvidence
|
|
from k1link.perception.realtime_contract import RealtimeContractError
|
|
|
|
NOW = 9_007_199_254_740_993
|
|
|
|
|
|
def check(stamps, states, actions, *, elapsed_ms=0, uncertainty_ms=0):
|
|
return CostmapCellEvidence(tuple(stamps)).assess(
|
|
states=tuple(states),
|
|
actions=tuple(actions),
|
|
source_time_ns=NOW,
|
|
observed_source_time_ns=NOW + elapsed_ms * 1_000_000,
|
|
maximum_age_ms=250,
|
|
clock_uncertainty_ms=uncertainty_ms,
|
|
)
|
|
|
|
|
|
def test_expiry_is_addressed_and_cannot_clear_occupied_or_unknown_cells():
|
|
stamps = (NOW - 200_000_000, NOW, NOW - 999_000_000, None, None)
|
|
result = check(stamps, [1, 1, 2, 0, 1], [0, 1, 2, 2, 0], elapsed_ms=51)
|
|
assert result.states == (3, 1, 2, 0, 3)
|
|
assert result.actions == (2, 1, 2, 2, 2)
|
|
assert result.expired_cells == 1 and result.missing_support_cells == 1
|
|
assert result.suppressed_permissions == 2
|
|
assert result.oldest_permissive_source_time_ns == NOW
|
|
# Even an earlier assessment cannot restore a suppressed permission.
|
|
repeated = check(stamps, result.states, result.actions)
|
|
assert repeated.states == result.states and repeated.actions == result.actions
|
|
|
|
|
|
def test_ttl_boundary_includes_mapping_uncertainty_without_refreshing_support():
|
|
stamp = NOW - 200_000_000
|
|
assert check([stamp], [1], [0], elapsed_ms=45, uncertainty_ms=5).actions == (0,)
|
|
assert check([stamp], [1], [0], elapsed_ms=46, uncertainty_ms=5).actions == (2,)
|
|
assert check([stamp], [1], [0]).oldest_permissive_source_time_ns == stamp
|
|
|
|
|
|
def test_cell_wire_is_exact_int64_and_has_a_bounded_shape():
|
|
evidence = CostmapCellEvidence((NOW, None))
|
|
assert evidence.to_dict()["observed_source_time_ns"] == [str(NOW), None]
|
|
assert CostmapCellEvidence.from_dict(json.loads(json.dumps(evidence.to_dict()))) == evidence
|
|
with pytest.raises(RealtimeContractError):
|
|
CostmapCellEvidence((NOW,) * 8193)
|
|
|
|
|
|
@pytest.mark.parametrize("stamp", [NOW, "01", "-1", str(2**63), True, {}, "NaN"])
|
|
def test_invalid_wire_timestamp_is_rejected(stamp):
|
|
value = CostmapCellEvidence((NOW,)).to_dict()
|
|
value["observed_source_time_ns"] = [stamp]
|
|
with pytest.raises(RealtimeContractError):
|
|
CostmapCellEvidence.from_dict(value)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"stamps,states,actions,elapsed,uncertainty",
|
|
[
|
|
([NOW + 1], [1], [0], 0, 0),
|
|
([NOW], [1], [0], -1, 0),
|
|
([NOW], [1], [0], 0, float("nan")),
|
|
([NOW], [1], [0], 0, -1),
|
|
([NOW], [True], [0], 0, 0),
|
|
([NOW], [1], [True], 0, 0),
|
|
([NOW], [2], [0], 0, 0),
|
|
([None], [0], [1], 0, 0),
|
|
([NOW], [], [], 0, 0),
|
|
],
|
|
)
|
|
def test_invalid_evidence_never_grants_permission(stamps, states, actions, elapsed, uncertainty):
|
|
with pytest.raises(RealtimeContractError):
|
|
check(stamps, states, actions, elapsed_ms=elapsed, uncertainty_ms=uncertainty)
|