304 lines
11 KiB
Python
304 lines
11 KiB
Python
import asyncio
|
|
import queue
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from aiortc import RTCConfiguration, RTCPeerConnection, RTCSessionDescription
|
|
|
|
from k1link.viewer.node_media import MEDIA_PROTOCOL, NodeMediaPeers, admit_sdp
|
|
|
|
SDP_HEADER = "v=0\r\nm=application 9 UDP/DTLS/SCTP webrtc-datachannel\r\n"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"address,kind", [("8.8.8.8", "host"), ("192.168.1.2", "relay"), ("::1", "host")]
|
|
)
|
|
def test_media_rejects_non_private_or_relay_candidates(address, kind):
|
|
with pytest.raises(ValueError):
|
|
admit_sdp(SDP_HEADER + f"a=candidate:1 1 udp 1 {address} 12345 typ {kind}\r\n")
|
|
|
|
|
|
def test_media_accepts_paired_lan_and_tailnet_candidates():
|
|
for address in ("192.168.1.2", "100.80.6.113", "peer.local"):
|
|
admit_sdp(SDP_HEADER + f"a=candidate:1 1 udp 1 {address} 12345 typ host\r\n")
|
|
|
|
|
|
def test_failed_media_offer_retires_peer_without_camera_channel(monkeypatch):
|
|
async def reject(*_):
|
|
raise ValueError("Invalid remote description")
|
|
|
|
monkeypatch.setattr(RTCPeerConnection, "setRemoteDescription", reject)
|
|
|
|
async def run():
|
|
class Camera:
|
|
def snapshot(self):
|
|
return {}
|
|
|
|
peers = NodeMediaPeers(None, Camera())
|
|
try:
|
|
with pytest.raises(ValueError):
|
|
await peers.offer({"sdp": SDP_HEADER, "view_id": str(uuid4())})
|
|
assert peers.items == {}
|
|
finally:
|
|
await peers.close_all()
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_native_webrtc_roundtrip_and_missing_camera_preserve_rrd(monkeypatch):
|
|
"""One bounded loopback peer, no STUN/TURN, device or external network."""
|
|
import aioice.ice
|
|
|
|
class Subscription:
|
|
def __init__(self):
|
|
self.pending = None
|
|
self.output = queue.Queue()
|
|
self.output.put(payload)
|
|
|
|
def next_batch(self, *, wait=True):
|
|
try:
|
|
self.pending = (1, self.output.get(timeout=0.1), None)
|
|
return self.pending
|
|
except queue.Empty:
|
|
return b""
|
|
|
|
def snapshot(self, frame=None):
|
|
return {"type": "lidar-state", "sequence": 1, "age_ms": 0, "points": 5000}
|
|
|
|
def acknowledge(self, sequence):
|
|
self.pending = None
|
|
|
|
def release(self):
|
|
pass
|
|
|
|
class Hub:
|
|
def subscribe(self, view_id, after):
|
|
return Subscription()
|
|
|
|
class Camera:
|
|
def snapshot(self):
|
|
return {"generation": None, "phase": "error",
|
|
"recording": {"source_end_expected": True}}
|
|
|
|
# The old test's sub-16KB fake payload missed the actual fragmentation bug.
|
|
import numpy as np
|
|
import rerun as rr
|
|
recording = rr.RecordingStream("synthetic-node-media")
|
|
binary = recording.binary_stream()
|
|
recording.log("points", rr.Points3D(np.random.default_rng(42).random((5000, 3))))
|
|
payload = binary.read()
|
|
assert payload.startswith(b"RRF2") and len(payload) > 16384
|
|
|
|
async def run():
|
|
peers = NodeMediaPeers(Hub(), Camera())
|
|
monkeypatch.setattr(aioice.ice, "get_host_addresses", lambda **_: ["127.0.0.1"])
|
|
client = RTCPeerConnection(RTCConfiguration(iceServers=[]))
|
|
channel = client.createDataChannel("rrd", ordered=True)
|
|
client.createDataChannel("camera", ordered=True)
|
|
received = asyncio.Event()
|
|
payloads = []
|
|
|
|
@channel.on("message")
|
|
def message(data):
|
|
if isinstance(data, str):
|
|
return
|
|
payloads.append(data)
|
|
if len(payloads) > 1 and sum(map(len, payloads[1:])) == len(payload):
|
|
channel.send('{"ack":1}')
|
|
received.set()
|
|
|
|
try:
|
|
await client.setLocalDescription(await client.createOffer())
|
|
answer = await peers.offer({
|
|
"sdp": client.localDescription.sdp, "view_id": str(uuid4()),
|
|
})
|
|
await client.setRemoteDescription(
|
|
RTCSessionDescription(sdp=answer["sdp"], type="answer")
|
|
)
|
|
await asyncio.wait_for(received.wait(), timeout=8)
|
|
assert answer["media_protocol"] == MEDIA_PROTOCOL
|
|
assert payloads[0] == b"MCF1" + len(payload).to_bytes(4, "big")
|
|
assert b"".join(payloads[1:]) == payload
|
|
assert all(len(part) <= 16384 for part in payloads[1:])
|
|
assert answer["peer_id"] in peers.items
|
|
assert channel.readyState == "open"
|
|
finally:
|
|
await client.close()
|
|
await peers.close_all()
|
|
assert not peers.items
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_camera_waits_for_post_calibration_producer_and_delivers_metadata():
|
|
class Camera:
|
|
calls = 0
|
|
opened = []
|
|
|
|
def snapshot(self):
|
|
self.calls += 1
|
|
return {"generation": 3, "delivery": {"media_type": "video/mp4"},
|
|
"recording": {"active": True, "producer_alive": self.calls >= 2,
|
|
"media_ready": self.calls >= 3, "active_epoch": 3}}
|
|
|
|
def open_delivery(self, generation, *, require_recording=False):
|
|
assert require_recording
|
|
self.opened.append(generation)
|
|
return "lease"
|
|
|
|
class Channel:
|
|
messages = []
|
|
|
|
def send(self, data):
|
|
self.messages.append(data)
|
|
|
|
async def run():
|
|
camera, channel = Camera(), Channel()
|
|
peers = NodeMediaPeers(None, camera)
|
|
peers.items["synthetic"] = {}
|
|
assert await peers.camera_delivery("synthetic", channel) == "lease"
|
|
assert camera.opened == [3]
|
|
import json
|
|
assert json.loads(channel.messages[0]) == {"type": "camera-ready", "mime": "video/mp4"}
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_installed_camera_uses_declared_os_ffmpeg():
|
|
from pathlib import Path
|
|
root = Path(__file__).resolve().parents[1]
|
|
unit = (root / "plugins/xgrids-k1/packaging/mission-core-k1.service").read_text()
|
|
assert "Environment=MISSIONCORE_FFMPEG_BINARY=/usr/bin/ffmpeg" in unit
|
|
assert "iproute2, ffmpeg" in (root / "plugins/xgrids-k1/packaging/build_deb.py").read_text()
|
|
|
|
|
|
def test_native_rrd_idle_does_not_close_camera_or_peer(monkeypatch):
|
|
"""Regression: the real binary sink, two media channels, idle then resume."""
|
|
import json
|
|
import time
|
|
from types import SimpleNamespace
|
|
|
|
from k1link.data_plane import ConsumerFrameContext, DecodedPointCloudView
|
|
from k1link.viewer.node_rerun import NodeRerunBridge
|
|
|
|
class Segments:
|
|
def __init__(self):
|
|
self.queue = queue.Queue()
|
|
|
|
def get(self, timeout):
|
|
return self.queue.get(timeout=timeout)
|
|
|
|
class Camera:
|
|
def __init__(self):
|
|
self.lease = SimpleNamespace(segments=Segments())
|
|
self.released = []
|
|
|
|
def snapshot(self):
|
|
return {"generation": 1, "delivery": {"media_type": "video/mp4"},
|
|
"recording": {"active": True, "producer_alive": True,
|
|
"media_ready": True, "active_epoch": 1}}
|
|
|
|
def open_delivery(self, generation, *, require_recording=False):
|
|
assert require_recording
|
|
assert generation == 1
|
|
return self.lease
|
|
|
|
def release_delivery(self, lease, **_):
|
|
self.released.append(lease)
|
|
|
|
def mark_streaming(self, _):
|
|
pass
|
|
|
|
async def run():
|
|
import aioice.ice
|
|
monkeypatch.setattr(aioice.ice, "get_host_addresses", lambda **_: ["127.0.0.1"])
|
|
hub, camera = NodeRerunBridge(), Camera()
|
|
peers = NodeMediaPeers(hub, camera)
|
|
client = RTCPeerConnection(RTCConfiguration(iceServers=[]))
|
|
rrd = client.createDataChannel("rrd", ordered=True)
|
|
video = client.createDataChannel("camera", ordered=True)
|
|
ready, resumed, camera_frame = asyncio.Event(), asyncio.Event(), asyncio.Event()
|
|
|
|
batch_sequence, remaining = 0, 0
|
|
|
|
@rrd.on("message")
|
|
def rrd_message(data):
|
|
nonlocal batch_sequence, remaining
|
|
if isinstance(data, str):
|
|
value = json.loads(data)
|
|
batch_sequence = value["sequence"]
|
|
if value["lidar"]["sequence"] == 1:
|
|
resumed.set()
|
|
elif data.startswith(b"MCF1") and remaining == 0:
|
|
remaining = int.from_bytes(data[4:], "big")
|
|
else:
|
|
remaining -= len(data)
|
|
if remaining == 0:
|
|
rrd.send(json.dumps({"ack": batch_sequence}))
|
|
ready.set()
|
|
|
|
@video.on("message")
|
|
def camera_message(data):
|
|
if data == b"synthetic-camera-fragment":
|
|
camera_frame.set()
|
|
|
|
try:
|
|
await client.setLocalDescription(await client.createOffer())
|
|
answer = await peers.offer({
|
|
"sdp": client.localDescription.sdp, "view_id": str(uuid4()),
|
|
})
|
|
await client.setRemoteDescription(
|
|
RTCSessionDescription(sdp=answer["sdp"], type="answer")
|
|
)
|
|
await asyncio.wait_for(ready.wait(), 8)
|
|
await asyncio.sleep(1.5)
|
|
assert answer["peer_id"] in peers.items
|
|
assert rrd.readyState == video.readyState == "open"
|
|
assert not camera.released
|
|
now = time.monotonic_ns()
|
|
hub.process(DecodedPointCloudView(
|
|
ConsumerFrameContext(1, time.time_ns(), now, now, 1, True),
|
|
"world", ((1.0, 2.0, 3.0),), b"\x01",
|
|
))
|
|
camera.lease.segments.queue.put(("media", b"synthetic-camera-fragment"))
|
|
await asyncio.wait_for(resumed.wait(), 5)
|
|
await asyncio.wait_for(camera_frame.wait(), 5)
|
|
assert rrd.readyState == video.readyState == "open"
|
|
finally:
|
|
await client.close()
|
|
await peers.close_all()
|
|
hub.close()
|
|
assert not peers.items
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_network_backpressure_longer_than_two_seconds_is_a_pause():
|
|
"""Bounded synthetic scheduler pause; no sockets, scanner or load generation."""
|
|
import time
|
|
|
|
class Channel:
|
|
readyState = "open"
|
|
sent = []
|
|
blocked = True
|
|
|
|
@property
|
|
def bufferedAmount(self):
|
|
return 1024 * 1024 if self.blocked else 0
|
|
|
|
def send(self, value):
|
|
self.sent.append(value)
|
|
|
|
async def run():
|
|
channel = Channel()
|
|
peers = object.__new__(NodeMediaPeers)
|
|
task = asyncio.create_task(peers.send(channel, b"small-synthetic-payload", lambda: True))
|
|
started = time.monotonic()
|
|
await asyncio.sleep(2.1)
|
|
assert not task.done() and channel.sent == []
|
|
channel.blocked = False
|
|
await asyncio.wait_for(task, 1)
|
|
assert time.monotonic() - started >= 2
|
|
assert channel.sent == [b"MCF1\x00\x00\x00\x17", b"small-synthetic-payload"]
|
|
asyncio.run(run())
|