250 lines
8.7 KiB
Python
250 lines
8.7 KiB
Python
"""Small synthetic memory/clock/pipe checks; no real decode or models on the Mac."""
|
|
|
|
import json
|
|
import socket
|
|
import struct
|
|
from contextlib import contextmanager
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from k1link.perception.streaming_pipe_rpc import BoundedPipeRpc, PipeRpcError
|
|
from k1link.perception.streaming_queue import StreamMailbox
|
|
from k1link.perception.streaming_sensors import CausalSensorWindow, normalized_sensor
|
|
|
|
|
|
def bundle(seq=0, size=10):
|
|
return {"sequence": seq, "payload_bytes": size}
|
|
|
|
|
|
def test_reservation_handoff_is_atomic_and_does_not_double_charge():
|
|
mailbox = StreamMailbox(byte_limit=10)
|
|
ticket = mailbox.reserve_ingress(10)
|
|
value = bundle()
|
|
assert mailbox.put_reserved(value, ticket)
|
|
assert mailbox.bytes == mailbox.peak_bytes == 10
|
|
with pytest.raises(ValueError, match="ownership"):
|
|
ticket.release()
|
|
assert mailbox.take() is value
|
|
mailbox.cancel()
|
|
assert mailbox.bytes == 10 and not mailbox.quiescent
|
|
mailbox.release(value)
|
|
assert mailbox.quiescent
|
|
|
|
|
|
@pytest.mark.parametrize("case", ["closed", "size", "foreign", "sequence"])
|
|
def test_rejected_handoff_keeps_caller_ownership(case):
|
|
mailbox = StreamMailbox(byte_limit=40)
|
|
ticket = mailbox.reserve_ingress(10)
|
|
value = bundle()
|
|
if case == "closed":
|
|
mailbox.cancel()
|
|
assert not mailbox.put_reserved(value, ticket)
|
|
else:
|
|
if case == "size":
|
|
value["payload_bytes"] = 11
|
|
if case == "sequence":
|
|
mailbox.put(bundle(1))
|
|
with pytest.raises(ValueError):
|
|
(StreamMailbox() if case == "foreign" else mailbox).put_reserved(value, ticket)
|
|
assert mailbox.bytes >= 10
|
|
ticket.size = 99999 # Only the stored reservation size is authoritative.
|
|
ticket.release()
|
|
mailbox.cancel()
|
|
assert mailbox.bytes == 0 and mailbox.quiescent
|
|
|
|
|
|
def test_handoff_overflow_drops_pending_only():
|
|
mailbox = StreamMailbox(byte_limit=40)
|
|
mailbox.put(bundle(0))
|
|
active = mailbox.take()
|
|
mailbox.put(bundle(1, 5))
|
|
mailbox.put(bundle(2, 5))
|
|
ticket = mailbox.reserve_ingress(20)
|
|
assert mailbox.put_reserved(bundle(3, 20), ticket)
|
|
assert mailbox.dropped == [{"sequence": 1, "reason": "pending-overflow"}]
|
|
assert mailbox.bytes == 35
|
|
mailbox.cancel()
|
|
assert mailbox.bytes == 10
|
|
mailbox.release(active)
|
|
assert mailbox.quiescent
|
|
|
|
|
|
@contextmanager
|
|
def pipe(response=None, check=lambda: None):
|
|
client, server = socket.socketpair()
|
|
try:
|
|
if response:
|
|
server.sendall(response)
|
|
yield BoundedPipeRpc(client.fileno(), client.fileno(), check), server
|
|
finally:
|
|
client.close()
|
|
server.close()
|
|
|
|
|
|
def message(header, payload=b""):
|
|
raw = json.dumps({**header, "payload_bytes": len(payload)}).encode()
|
|
return struct.pack("<I", len(raw)) + raw + payload
|
|
|
|
|
|
def test_rpc_fills_only_the_callers_preallocated_buffer():
|
|
with pipe(message({"frame_index": 0}, b"1234")) as (rpc, server):
|
|
target = bytearray(4)
|
|
assert rpc.exchange({"op": "decode"}, b"fragment", target) == {"frame_index": 0}
|
|
assert target == b"1234"
|
|
raw = server.recv(1024)
|
|
size = struct.unpack("<I", raw[:4])[0]
|
|
assert json.loads(raw[4 : 4 + size]) == {"op": "decode", "payload_bytes": 8}
|
|
assert raw[4 + size :] == b"fragment"
|
|
|
|
|
|
@pytest.mark.parametrize("case", ["timeout", "size", "header", "duplicate", "truncated", "guard"])
|
|
def test_rpc_failures_are_bounded_and_cannot_be_retried(case):
|
|
raw = message({}, b"x") if case == "size" else None
|
|
if case == "header":
|
|
raw = struct.pack("<I", 4097)
|
|
if case == "duplicate":
|
|
body = b'{"payload_bytes":0,"payload_bytes":0}'
|
|
raw = struct.pack("<I", len(body)) + body
|
|
|
|
def guard():
|
|
if case == "guard":
|
|
raise ValueError("lost epoch")
|
|
|
|
with pipe(raw, guard) as (rpc, server):
|
|
if case == "truncated":
|
|
server.shutdown(socket.SHUT_WR)
|
|
with pytest.raises(ValueError):
|
|
rpc.exchange(None, b"", bytearray(), timeout=0.02)
|
|
assert rpc.failed
|
|
with pytest.raises(PipeRpcError, match="poisoned"):
|
|
rpc.exchange(None, b"", bytearray(), timeout=0.02)
|
|
|
|
|
|
def point(stamp, seq=1):
|
|
raw = struct.pack("<I", 1) + np.array([[1, 2, 3]], "<f8").tobytes() + b"\xff"
|
|
return normalized_sensor("lidar", seq, stamp, raw)
|
|
|
|
|
|
def pose(stamp):
|
|
return normalized_sensor("pose", 1, stamp, np.array([0, 0, 0, 0, 0, 0, 1], "<f8").tobytes())
|
|
|
|
|
|
def test_causal_preroll_is_preserved_without_future_sensor_substitution():
|
|
mailbox = StreamMailbox()
|
|
window = CausalSensorWindow(mailbox)
|
|
window.append(point(500_000_000))
|
|
window.append(pose(990_000_000))
|
|
window.append(point(999_000_000, 2))
|
|
cut = window.bind(1_000_000_000)
|
|
assert cut.available and len(cut.increments) == len(cut.history_only) == 1
|
|
assert len(window.rolling) == 2 and cut.increments[0].sequence == 2
|
|
window.finish_camera(1_000_000_000)
|
|
window.append(pose(1_001_000_000))
|
|
with pytest.raises(ValueError, match="backwards"):
|
|
window.bind(1_000_000_000)
|
|
assert cut.pose_age_ns == 10_000_000 # New pose cannot mutate the existing cut.
|
|
mailbox.cancel()
|
|
assert mailbox.bytes == window.CACHE_BYTES and not mailbox.quiescent
|
|
window.close()
|
|
assert mailbox.quiescent
|
|
|
|
|
|
def test_sensor_cache_cannot_grow_when_camera_stalls():
|
|
mailbox = StreamMailbox()
|
|
window = CausalSensorWindow(mailbox)
|
|
for sequence in range(64):
|
|
window.append(point(sequence, sequence))
|
|
with pytest.raises(ValueError, match="cache exceeds"):
|
|
window.append(point(65, 65))
|
|
window.close()
|
|
mailbox.finish()
|
|
assert mailbox.quiescent
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"modality,raw",
|
|
[
|
|
("lidar", b""),
|
|
("lidar", struct.pack("<I", 50001)),
|
|
("pose", bytes(55)),
|
|
("pose", bytes(56)),
|
|
("unknown", bytes(56)),
|
|
("lidar", struct.pack("<I", 1) + np.array([np.nan, 0, 0], "<f8").tobytes() + b"x"),
|
|
],
|
|
)
|
|
def test_normalized_sensor_layout_is_explicit_and_validated(modality, raw):
|
|
with pytest.raises(ValueError):
|
|
normalized_sensor(modality, 1, 100, raw)
|
|
|
|
|
|
def test_cloud_validation_checks_later_bounded_chunks():
|
|
xyz = np.ones((2000, 3), "<f8")
|
|
xyz[-1, -1] = np.inf
|
|
with pytest.raises(ValueError, match="nonfinite"):
|
|
normalized_sensor("lidar", 1, 100, struct.pack("<I", 2000) + xyz.tobytes() + bytes(2000))
|
|
|
|
|
|
def test_wire_pose_owns_aligned_immutable_quaternion_with_exact_values():
|
|
from k1link.perception.geometry_math import quaternion_xyzw_to_rotation_matrix
|
|
|
|
quaternion = np.array([0.1, -0.2, 0.3, 0.9273618495495703], "<f8")
|
|
raw = np.array([1, 2, 3], "<f8").tobytes() + quaternion.tobytes()
|
|
position, decoded = normalized_sensor("pose", 1, 100, raw).value
|
|
assert decoded.flags.owndata and decoded.ctypes.data % 16 == 0
|
|
assert not decoded.flags.writeable and not position.flags.writeable
|
|
assert decoded.tobytes() == quaternion.tobytes()
|
|
assert np.array_equal(
|
|
quaternion_xyzw_to_rotation_matrix(decoded),
|
|
quaternion_xyzw_to_rotation_matrix(quaternion),
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("failure", ["socket", "receiver-registration", "source-registration"])
|
|
def test_bridge_initialization_failure_releases_unstarted_resources(monkeypatch, failure):
|
|
monkeypatch.syspath_prepend(
|
|
str(
|
|
Path(__file__).resolve().parents[1]
|
|
/ "experiments/perception/worker/streaming_profile_stage1"
|
|
)
|
|
)
|
|
import pilot_binary_bridge as bridge
|
|
|
|
mailbox = StreamMailbox()
|
|
pair = socket.socketpair()
|
|
calls = []
|
|
|
|
def sockets():
|
|
if failure == "socket":
|
|
raise ValueError("injected socket failure")
|
|
return pair
|
|
|
|
def register(thread):
|
|
calls.append(thread)
|
|
if failure == "receiver-registration" or len(calls) == 2:
|
|
raise ValueError("injected registration failure")
|
|
|
|
monkeypatch.setattr(bridge.socket, "socketpair", sockets)
|
|
# Only constructor ownership is under test, without a live lease/thread.
|
|
monkeypatch.setattr("k1link.perception.streaming_ingress.wire.binding", lambda start: {})
|
|
runtime = SimpleNamespace(
|
|
mailbox=mailbox,
|
|
start=None,
|
|
track_thread=register,
|
|
register_ingress=lambda thread, epoch: register(thread),
|
|
)
|
|
source = SimpleNamespace(source_zero=0, wall_zero=0, run=lambda sock: None)
|
|
try:
|
|
with pytest.raises(ValueError, match="injected"):
|
|
bridge.BinaryGraphBridge(runtime, None, source, {})
|
|
mailbox.finish()
|
|
assert mailbox.bytes == 0 and mailbox.quiescent
|
|
if failure != "socket":
|
|
assert all(s.fileno() == -1 for s in pair)
|
|
assert all(t.ident is None for t in calls)
|
|
finally:
|
|
for connection in pair:
|
|
connection.close()
|