186 lines
6.2 KiB
Python
186 lines
6.2 KiB
Python
"""Conditional clock intervals, never a symmetric-network offset assertion."""
|
|
|
|
from dataclasses import replace
|
|
|
|
import pytest
|
|
|
|
from k1link.perception.streaming_clock import (
|
|
ClockBounds,
|
|
ClockMappingError,
|
|
ClockProbe,
|
|
ClockWindow,
|
|
)
|
|
|
|
BASE = 10**16 # Deliberately above IEEE754 exact-integer range.
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"fault",
|
|
[
|
|
None,
|
|
"float",
|
|
"integer",
|
|
"plus",
|
|
"leading",
|
|
"negative_zero",
|
|
"extra",
|
|
"missing",
|
|
"flags",
|
|
"wide",
|
|
"schema",
|
|
],
|
|
)
|
|
def test_bounds_wire_exact_closed_and_signed(fault):
|
|
bounds = window().add(probe())
|
|
document = bounds.to_dict()
|
|
if fault is None:
|
|
assert ClockBounds.from_dict(document) == bounds
|
|
return
|
|
if fault == "float":
|
|
document["measured_at_ns"] = float(bounds.measured_at_ns)
|
|
elif fault == "integer":
|
|
document["measured_at_ns"] = bounds.measured_at_ns
|
|
elif fault == "plus":
|
|
document["offset_upper_ns"] = "+1"
|
|
elif fault == "leading":
|
|
document["measured_at_ns"] = "0" + document["measured_at_ns"]
|
|
elif fault == "negative_zero":
|
|
document["offset_upper_ns"] = "-0"
|
|
elif fault == "extra":
|
|
document["ready"] = True
|
|
elif fault == "missing":
|
|
del document["samples"]
|
|
elif fault == "flags":
|
|
document["conditional_rate_error_envelope"] = 1
|
|
elif fault == "wide":
|
|
document["offset_upper_ns"] = str(2**63)
|
|
else:
|
|
document["schema_version"] = "old"
|
|
with pytest.raises(ClockMappingError):
|
|
ClockBounds.from_dict(document)
|
|
|
|
|
|
def probe(number=1, *, outbound=1_000_000, inbound=2_000_000, offset=-(10**12)):
|
|
sent = BASE + number * 100_000_000
|
|
return ClockProbe(
|
|
"mac",
|
|
"worker",
|
|
str(number),
|
|
sent,
|
|
sent + offset + outbound,
|
|
sent + offset + outbound + 100_000,
|
|
sent + outbound + inbound + 100_000,
|
|
)
|
|
|
|
|
|
def window(**kwargs):
|
|
return ClockWindow("mac", "worker", rate_ppm=500, timestamp_error_ns=50_000, **kwargs)
|
|
|
|
|
|
@pytest.mark.parametrize("offset", [-(10**12), 0, 10**12])
|
|
@pytest.mark.parametrize("outbound,inbound", [(1, 9_000_000), (9_000_000, 1), (1, 1)])
|
|
def test_contains_true_offset_without_symmetric_path_assumption(offset, outbound, inbound):
|
|
sample = probe(offset=offset, outbound=outbound, inbound=inbound)
|
|
mapping = window().add(sample)
|
|
now = sample.local_receive_ns
|
|
assert mapping.offset_at(now)[0] <= offset <= mapping.offset_at(now)[1]
|
|
assert mapping.uncertainty_ns(now + 1_000_000) > mapping.uncertainty_ns(now)
|
|
value = mapping.to_dict()
|
|
assert value["measured_at_ns"] == str(now)
|
|
assert not value["symmetric_network_assumed"]
|
|
assert value["timestamp_error_budget_ns"] == "50000"
|
|
|
|
|
|
def test_intersection_requires_evidence_and_drift_is_not_averaged_away():
|
|
mapping = window()
|
|
first = mapping.add(probe(outbound=1_000_000, inbound=20_000_000))
|
|
with pytest.raises(ClockMappingError, match="budget"):
|
|
first.require(first.measured_at_ns)
|
|
second = mapping.add(probe(2, outbound=20_000_000, inbound=1_000_000))
|
|
second.require(second.measured_at_ns)
|
|
assert second.uncertainty_ns(second.measured_at_ns) < 2_000_000
|
|
assert (
|
|
second.offset_at(second.measured_at_ns)[0]
|
|
< -(10**12)
|
|
< second.offset_at(second.measured_at_ns)[1]
|
|
)
|
|
|
|
|
|
def test_expired_mapping_waits_and_new_observation_can_restore_it():
|
|
mapping = window()
|
|
first = mapping.add(probe())
|
|
with pytest.raises(ClockMappingError, match="expired"):
|
|
first.require(first.expires_at_ns)
|
|
with pytest.raises(ClockMappingError, match="expired"):
|
|
mapping.current(first.expires_at_ns)
|
|
resumed = mapping.add(probe(30))
|
|
assert resumed.samples == 1
|
|
resumed.require(resumed.measured_at_ns)
|
|
|
|
|
|
def test_contradictory_samples_quarantine_not_cherry_pick():
|
|
mapping = window()
|
|
mapping.add(probe())
|
|
with pytest.raises(ClockMappingError, match="envelope"):
|
|
mapping.add(probe(2, offset=0))
|
|
with pytest.raises(ClockMappingError, match="quarantined"):
|
|
mapping.add(probe(3))
|
|
with pytest.raises(ClockMappingError):
|
|
mapping.current(BASE + 3_000_000_000)
|
|
|
|
|
|
@pytest.mark.parametrize("fault", ["local", "remote", "nonce", "overlap"])
|
|
def test_foreign_or_replayed_observation_cannot_refresh_mapping(fault):
|
|
mapping = window()
|
|
original = probe()
|
|
mapping.add(original)
|
|
second = probe(2)
|
|
changes = {
|
|
"local": {"local_clock_id": "other"},
|
|
"remote": {"remote_clock_id": "rebooted"},
|
|
"nonce": {"nonce": original.nonce},
|
|
"overlap": {"local_send_ns": original.local_receive_ns},
|
|
}
|
|
with pytest.raises(ClockMappingError):
|
|
mapping.add(replace(second, **changes[fault]))
|
|
assert mapping.last_receive_ns == original.local_receive_ns
|
|
|
|
|
|
def test_history_is_bounded_and_exact_values_survive():
|
|
mapping = window(maximum_age_ns=5_000_000_000)
|
|
for number in range(1, 35):
|
|
bounds = mapping.add(probe(number))
|
|
assert len(mapping.samples) == bounds.samples == 16
|
|
assert bounds.to_dict()["measured_at_ns"] == str(probe(34).local_receive_ns)
|
|
|
|
|
|
@pytest.mark.parametrize("value", [-1, 2**63, True, 1.0])
|
|
def test_invalid_monotonic_timestamp_rejected(value):
|
|
with pytest.raises(ClockMappingError):
|
|
replace(probe(), local_send_ns=value)
|
|
|
|
|
|
def test_deadline_remote_order_and_backward_reads():
|
|
sample = probe()
|
|
for changes in (
|
|
{"local_receive_ns": sample.local_send_ns - 1},
|
|
{"local_receive_ns": sample.local_send_ns + 2_000_000_001},
|
|
{"remote_send_ns": sample.remote_receive_ns - 1},
|
|
):
|
|
with pytest.raises(ClockMappingError):
|
|
replace(sample, **changes)
|
|
mapping = window().add(sample)
|
|
with pytest.raises(ClockMappingError):
|
|
mapping.offset_at(mapping.measured_at_ns - 1)
|
|
with pytest.raises(ClockMappingError):
|
|
replace(mapping, offset_lower_ns=mapping.offset_upper_ns + 1)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"rate,error,age",
|
|
[(0, 1, 1), (1001, 1, 1), (True, 1, 1), (1, 0, 1), (1, 1_000_001, 1), (1, 1, 0)],
|
|
)
|
|
def test_controller_must_supply_bounded_envelope(rate, error, age):
|
|
with pytest.raises(ClockMappingError):
|
|
ClockWindow("mac", "worker", rate_ppm=rate, timestamp_error_ns=error, maximum_age_ns=age)
|