stroblme
f52d9c3077
_serve_once treated a cancelled call future as finished, so an event loop cancelling all tasks at once (shutdown) never sent the worker a cancel and the call's thread kept asyncio.run waiting for SILENCE_S. The cameras audio test now removes its voice flow and device, the unpaired-frame test waits for its first frame before the unpair, and test_connector no longer inherits the controller an app test bound to demand.
514 lines
20 KiB
Python
514 lines
20 KiB
Python
"""Cameras on a device: frames over its socket, demand, and what a screen fetches.
|
|
|
|
A device reports a ``camera`` port that sends only while
|
|
somebody wants it. Demand is what the viewer sockets want plus what running
|
|
flows read, mapped onto the device's on-demand ports and sent to its socket.
|
|
"""
|
|
|
|
import json
|
|
import time
|
|
from collections.abc import Callable
|
|
from typing import Any
|
|
|
|
import orjson
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from fluksio.api.routes import oauth
|
|
from fluksio.core.config import settings
|
|
from fluksio.flow.demand import demand
|
|
from fluksio.flow.events import RECENT_DIGESTS, event_bus
|
|
from tests.api.routes.test_devices import (
|
|
FLOWS,
|
|
_device_node,
|
|
_pair,
|
|
_panels,
|
|
_publish_flow,
|
|
)
|
|
|
|
PREFIX = f"{settings.API_V1_STR}/devices"
|
|
SOCKET = f"{settings.API_V1_STR}/flows/ws"
|
|
ARTIFACTS = f"{settings.API_V1_STR}/artifacts"
|
|
|
|
#: What the NANO reports, as C1 fixes it.
|
|
REPORT = {
|
|
"schema": 1,
|
|
"platform": "esp32p4",
|
|
"board": "nano",
|
|
"firmware": "0.2.0",
|
|
"ports": [
|
|
{"port": "camera", "direction": "in", "dtype": "image", "on_demand": True},
|
|
{"port": "person", "direction": "in", "dtype": "bool"},
|
|
{"port": "detections", "direction": "in", "dtype": "json"},
|
|
{"port": "snapshot", "direction": "in", "dtype": "image"},
|
|
],
|
|
}
|
|
|
|
CAMERA_NODE = _device_node(
|
|
"cam1",
|
|
node="hall",
|
|
provides=[
|
|
{"name": "camera", "port": "camera", "dtype": "image"},
|
|
{"name": "person", "port": "person", "dtype": "bool"},
|
|
{"name": "snapshot", "port": "snapshot", "dtype": "image"},
|
|
],
|
|
)
|
|
|
|
#: A node reading the camera's every frame, which keeps it on for good.
|
|
READER = {
|
|
"id": "watch",
|
|
"type": "debounce",
|
|
"params": {"wait": 5},
|
|
"requires": [{"name": "camera", "dtype": "image"}],
|
|
"provides": [{"name": "moving", "dtype": "bool"}],
|
|
}
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _pairing_unlimited() -> None:
|
|
"""Every test pairs from the test client's one address."""
|
|
for bucket in [b for b in oauth._hits if b.startswith("pair:")]:
|
|
del oauth._hits[bucket]
|
|
|
|
|
|
def frame(name: str, data: bytes, media_type: str = "image/jpeg") -> bytes:
|
|
"""A binary frame as a device sends it."""
|
|
header = orjson.dumps(
|
|
{"type": "frame", "name": name, "media_type": media_type, "ts": 1234}
|
|
)
|
|
return len(header).to_bytes(4, "big") + header + data
|
|
|
|
|
|
def wait_for(check: Callable[[], Any], timeout: float = 5.0) -> Any:
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
found = check()
|
|
if found:
|
|
return found
|
|
time.sleep(0.02)
|
|
raise AssertionError("timed out")
|
|
|
|
|
|
@pytest.fixture
|
|
def camera(client: TestClient, superuser_token_headers: dict[str, str]):
|
|
"""A paired device with a camera, wired in flow ``cams``. Yields its header."""
|
|
_panels(client, superuser_token_headers, {"panels": [{"id": "cam1"}]})
|
|
device = _pair(client, superuser_token_headers, "cam1")
|
|
reported = client.put(f"{PREFIX}/cam1/report", headers=device, json=REPORT)
|
|
assert reported.status_code == 200, reported.text
|
|
_publish_flow(client, superuser_token_headers, "cams", [CAMERA_NODE])
|
|
yield device
|
|
client.delete(f"{FLOWS}/cams", headers=superuser_token_headers)
|
|
client.delete(f"{PREFIX}/cam1", headers=superuser_token_headers)
|
|
|
|
|
|
@pytest.fixture
|
|
def microphone(client: TestClient, superuser_token_headers: dict[str, str]):
|
|
"""A paired device with a microphone, wired in flow ``voice``. Yields its header.
|
|
|
|
Both go again afterwards: a published flow outlives the module in the
|
|
flow store, and the next module's app would load it without its device.
|
|
"""
|
|
_panels(client, superuser_token_headers, {"panels": [{"id": "mic1"}]})
|
|
device = _pair(client, superuser_token_headers, "mic1")
|
|
port = {"port": "mic", "direction": "in", "dtype": "audio"}
|
|
reported = client.put(
|
|
f"{PREFIX}/mic1/report", headers=device, json={**REPORT, "ports": [port]}
|
|
)
|
|
assert reported.status_code == 200, reported.text
|
|
node = _device_node(
|
|
"mic1",
|
|
node="panel",
|
|
provides=[{"name": "mic", "port": "mic", "dtype": "audio"}],
|
|
)
|
|
_publish_flow(client, superuser_token_headers, "voice", [node])
|
|
yield device
|
|
client.delete(f"{FLOWS}/voice", headers=superuser_token_headers)
|
|
client.delete(f"{PREFIX}/mic1", headers=superuser_token_headers)
|
|
|
|
|
|
def _ticket(client: TestClient, headers: dict[str, str]) -> str:
|
|
answer = client.post(f"{settings.API_V1_STR}/flows/ws-ticket", headers=headers)
|
|
return f"{SOCKET}?ticket={answer.json()['ticket']}"
|
|
|
|
|
|
def _next_demand(ws: Any) -> list[str]:
|
|
"""The next demand message on a device's socket, skipping values."""
|
|
while True:
|
|
message = ws.receive_json()
|
|
if message.get("type") == "demand":
|
|
return message["ports"]
|
|
|
|
|
|
def test_a_tile_asking_for_the_camera_turns_it_on_and_off(
|
|
client: TestClient, superuser_token_headers: dict[str, str], camera
|
|
) -> None:
|
|
"""Demand follows what viewers draw, and only on a port sent on demand."""
|
|
with client.websocket_connect(SOCKET, headers=camera) as device:
|
|
snapshot = device.receive_json()
|
|
assert snapshot["demand"] == []
|
|
assert snapshot["uplink"] == "binary"
|
|
|
|
with client.websocket_connect(
|
|
_ticket(client, superuser_token_headers)
|
|
) as viewer:
|
|
viewer.receive_json()
|
|
# `person` is drawn too, but it sends all the time anyway.
|
|
viewer.send_json({"type": "media", "names": ["cams.camera", "cams.person"]})
|
|
assert _next_demand(device) == ["camera"]
|
|
# The tile scrolled away: the page asks for nothing.
|
|
viewer.send_json({"type": "media", "names": []})
|
|
assert _next_demand(device) == []
|
|
viewer.send_json({"type": "media", "names": ["cams.camera"]})
|
|
assert _next_demand(device) == ["camera"]
|
|
# The viewer closed its socket.
|
|
assert _next_demand(device) == []
|
|
|
|
# A snapshot taken while someone watches says so.
|
|
demand.want("someone", {"cams.camera"})
|
|
try:
|
|
assert _next_demand(device) == ["camera"]
|
|
with client.websocket_connect(SOCKET, headers=camera) as again:
|
|
assert again.receive_json()["demand"] == ["camera"]
|
|
finally:
|
|
demand.drop("someone")
|
|
|
|
|
|
@pytest.mark.usefixtures("camera")
|
|
def test_a_flow_reading_the_camera_keeps_it_on(
|
|
client: TestClient, superuser_token_headers: dict[str, str]
|
|
) -> None:
|
|
"""A running flow that reads frames wants every one, watched or not."""
|
|
assert demand.ports("cam1") == []
|
|
_publish_flow(client, superuser_token_headers, "cams", [CAMERA_NODE, READER])
|
|
assert demand.ports("cam1") == ["camera"]
|
|
|
|
stopped = client.post(f"{FLOWS}/cams/stop", headers=superuser_token_headers)
|
|
assert stopped.status_code == 200, stopped.text
|
|
assert demand.ports("cam1") == []
|
|
client.post(f"{FLOWS}/cams/start", headers=superuser_token_headers)
|
|
assert demand.ports("cam1") == ["camera"]
|
|
|
|
# A port the device does not send on demand never has any.
|
|
_publish_flow(client, superuser_token_headers, "cams", [CAMERA_NODE])
|
|
demand.want("someone", {"cams.snapshot", "cams.person"})
|
|
try:
|
|
assert demand.ports("cam1") == []
|
|
finally:
|
|
demand.drop("someone")
|
|
|
|
|
|
def test_a_device_sends_frames_on_its_socket(
|
|
client: TestClient, superuser_token_headers: dict[str, str], camera
|
|
) -> None:
|
|
"""Held and published in one step, credited to its node, pushed to tiles."""
|
|
with (
|
|
client.websocket_connect(SOCKET, headers=camera) as device,
|
|
client.websocket_connect(_ticket(client, superuser_token_headers)) as viewer,
|
|
):
|
|
device.receive_json()
|
|
viewer.receive_json()
|
|
viewer.send_json({"type": "media", "names": ["cams.camera"]})
|
|
_next_demand(device)
|
|
|
|
device.send_bytes(frame("cams.camera", b"\xff\xd8 a jpeg"))
|
|
|
|
pushed = viewer.receive_bytes()
|
|
length = int.from_bytes(pushed[:4], "big")
|
|
header = orjson.loads(pushed[4 : 4 + length])
|
|
assert header["name"] == "cams.camera"
|
|
assert header["media_type"] == "image/jpeg"
|
|
assert pushed[4 + length :] == b"\xff\xd8 a jpeg"
|
|
|
|
value = viewer.receive_json()
|
|
assert value["type"] == "message_value"
|
|
assert value["value"]["digest"] == header["digest"]
|
|
assert value["source"]["kind"] == "node"
|
|
assert value["source"]["id"] == "cams.hall"
|
|
store = client.app.state.artifact_store
|
|
assert store.volatile.get(header["digest"]) == b"\xff\xd8 a jpeg"
|
|
# In memory, and not on the volume.
|
|
assert store.path(header["digest"]) is None
|
|
|
|
# The device is not sent its own camera's values back, only the rest.
|
|
from fluksio.api.routes.flows import panel_scope
|
|
|
|
scope = panel_scope(camera["Authorization"][7:], client.app)
|
|
assert scope == {"cams.person"}
|
|
|
|
|
|
def test_a_frame_the_device_may_not_send_is_dropped(
|
|
client: TestClient,
|
|
superuser_token_headers: dict[str, str],
|
|
camera,
|
|
caplog: pytest.LogCaptureFixture,
|
|
) -> None:
|
|
"""Refused, logged and dropped, and the socket carries on."""
|
|
controller = client.app.state.flow_controller
|
|
before = controller.values().get("cams.person")
|
|
with client.websocket_connect(SOCKET, headers=camera) as device:
|
|
device.receive_json()
|
|
for refused in (
|
|
frame("cams.person", b"\xff\xd8"), # a bool port
|
|
frame("cams.elsewhere", b"\xff\xd8"), # not one of its ports
|
|
frame("cams.camera", b"<svg/>", "text/html"), # not an image
|
|
b"\x00\x00\x00\x09not json!",
|
|
b"\x01",
|
|
):
|
|
device.send_bytes(refused)
|
|
# Still open: a frame it may send goes through after them.
|
|
device.send_bytes(frame("cams.camera", b"\xff\xd8 after"))
|
|
wait_for(lambda: (controller.values().get("cams.camera") or {}).get("value"))
|
|
assert controller.values().get("cams.person") == before
|
|
|
|
# A person's socket is no camera.
|
|
with client.websocket_connect(_ticket(client, superuser_token_headers)) as ws:
|
|
ws.receive_json()
|
|
ws.send_bytes(frame("cams.camera", b"\xff\xd8 forged"))
|
|
ws.send_json({"type": "media", "names": []})
|
|
current = controller.values()["cams.camera"]["value"]
|
|
data = client.app.state.artifact_store.volatile.get(current["digest"])
|
|
assert data == b"\xff\xd8 after"
|
|
from fluksio.api.routes.flows import FRAME_REFUSALS
|
|
|
|
# By fixed keys, never in the words the device chose.
|
|
said = [r.getMessage() for r in caplog.records if "Dropped a frame" in r.message]
|
|
for key in ("not_media", "not_wired", "media_type", "not_a_frame"):
|
|
assert f"Dropped a frame from cam1: {FRAME_REFUSALS[key]}" in said
|
|
assert any(FRAME_REFUSALS["not_a_device"] in line for line in said)
|
|
assert not any("elsewhere" in line or "text/html" in line for line in said)
|
|
|
|
|
|
def test_an_unpaired_device_is_closed_and_its_frames_stop(
|
|
client: TestClient, superuser_token_headers: dict[str, str], camera
|
|
) -> None:
|
|
from starlette.websockets import WebSocketDisconnect
|
|
|
|
with client.websocket_connect(SOCKET, headers=camera) as device:
|
|
device.receive_json()
|
|
client.post(f"{PREFIX}/cam1/unpair", headers=superuser_token_headers)
|
|
with pytest.raises(WebSocketDisconnect) as closed:
|
|
while True:
|
|
device.receive_json()
|
|
assert closed.value.code == 1008
|
|
|
|
|
|
def test_a_device_streams_audio_chunks_on_its_socket(
|
|
client: TestClient, microphone
|
|
) -> None:
|
|
"""A microphone's chunks take the camera's path, in their own family."""
|
|
controller = client.app.state.flow_controller
|
|
with client.websocket_connect(SOCKET, headers=microphone) as ws:
|
|
ws.receive_json()
|
|
ws.send_bytes(frame("voice.mic", b"\x00\x01")) # an image on an audio port
|
|
ws.send_bytes(frame("voice.mic", b"\x00\x02", "audio/L16;rate=16000"))
|
|
value = wait_for(
|
|
lambda: (controller.values().get("voice.mic") or {}).get("value")
|
|
)
|
|
assert value["media_type"] == "audio/L16;rate=16000"
|
|
assert client.app.state.artifact_store.volatile.get(value["digest"]) == b"\x00\x02"
|
|
|
|
|
|
def test_a_frame_past_the_cap_is_dropped(
|
|
client: TestClient, camera, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
controller = client.app.state.flow_controller
|
|
monkeypatch.setattr(settings, "MAX_FRAME_BYTES", 1000)
|
|
with client.websocket_connect(SOCKET, headers=camera) as device:
|
|
device.receive_json()
|
|
device.send_bytes(frame("cams.camera", b"\xff" * 2000))
|
|
device.send_bytes(frame("cams.camera", b"\xff\xd8 small"))
|
|
wait_for(lambda: (controller.values().get("cams.camera") or {}).get("value"))
|
|
digest = controller.values()["cams.camera"]["value"]["digest"]
|
|
store = client.app.state.artifact_store
|
|
assert store.volatile.get(digest) == b"\xff\xd8 small"
|
|
|
|
|
|
def test_a_frame_is_named_after_its_message(client: TestClient, camera) -> None:
|
|
controller = client.app.state.flow_controller
|
|
with client.websocket_connect(SOCKET, headers=camera) as device:
|
|
device.receive_json()
|
|
device.send_bytes(frame("cams.camera", b"\xff\xd8 named"))
|
|
value = wait_for(
|
|
lambda: (controller.values().get("cams.camera") or {}).get("value")
|
|
)
|
|
assert value["name"] == "cams.camera"
|
|
|
|
|
|
def test_an_unpaired_device_is_closed_at_its_next_frame(
|
|
client: TestClient, camera
|
|
) -> None:
|
|
"""Without the event an unpair sends, which a full queue may have lost."""
|
|
from starlette.websockets import WebSocketDisconnect
|
|
|
|
controller = client.app.state.flow_controller
|
|
with client.websocket_connect(SOCKET, headers=camera) as device:
|
|
device.receive_json()
|
|
device.send_bytes(frame("cams.camera", b"\xff\xd8 one"))
|
|
# The credential is asked as each frame is read, so a bump before the
|
|
# server has read the first one would refuse that one too.
|
|
wait_for(lambda: (controller.values().get("cams.camera") or {}).get("value"))
|
|
client.app.state.device_store.bump("cam1")
|
|
device.send_bytes(frame("cams.camera", b"\xff\xd8 two"))
|
|
with pytest.raises(WebSocketDisconnect) as closed:
|
|
while True:
|
|
device.receive_json()
|
|
assert closed.value.code == 1008
|
|
store = client.app.state.artifact_store
|
|
current = controller.values()["cams.camera"]["value"]
|
|
assert store.volatile.get(current["digest"]) == b"\xff\xd8 one"
|
|
|
|
|
|
def test_a_device_that_lost_events_catches_up(
|
|
client: TestClient, camera, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""Its demand and its unpairing come again with a snapshot, after the
|
|
events it did get."""
|
|
from starlette.websockets import WebSocketDisconnect
|
|
|
|
controller = client.app.state.flow_controller
|
|
with client.websocket_connect(SOCKET, headers=camera) as device:
|
|
assert device.receive_json()["demand"] == []
|
|
# Demand changes while the bus drops what the socket has queued.
|
|
monkeypatch.setattr(event_bus, "dropped", lambda queue: True)
|
|
demand.want("someone", {"cams.camera"})
|
|
try:
|
|
seen = []
|
|
while not seen or seen[-1].get("type") != "snapshot":
|
|
seen.append(device.receive_json())
|
|
assert seen[-1]["demand"] == ["camera"]
|
|
finally:
|
|
demand.drop("someone")
|
|
|
|
# And an unpair it never heard of closes it at the next batch.
|
|
client.app.state.device_store.bump("cam1")
|
|
controller.publish_message("cams.person", True)
|
|
with pytest.raises(WebSocketDisconnect) as closed:
|
|
while True:
|
|
device.receive_json()
|
|
assert closed.value.code == 1008
|
|
|
|
|
|
def test_a_report_is_bounded_once_its_schemas_are_in_place(
|
|
client: TestClient, camera
|
|
) -> None:
|
|
"""Thirty pins naming one big schema are thirty copies of it, kept."""
|
|
from fluksio.api.routes.devices import REPORT_MAX, REPORT_STORED_MAX
|
|
|
|
schema = {
|
|
"type": "object",
|
|
"properties": {f"p{i}": {"type": "integer"} for i in range(250)},
|
|
}
|
|
pins = [
|
|
{"port": f"gpio.{i}", "direction": "in", "dtype": "bool", "config": "pin"}
|
|
for i in range(60)
|
|
]
|
|
report = {**REPORT, "config_schemas": {"pin": schema}, "ports": pins}
|
|
raw = orjson.dumps(report)
|
|
assert len(raw) < REPORT_MAX
|
|
answer = client.put(f"{PREFIX}/cam1/report", headers=camera, content=raw)
|
|
assert answer.status_code == 413, answer.text
|
|
kept = client.app.state.device_store.state("cam1").report
|
|
assert len(orjson.dumps(kept)) < REPORT_STORED_MAX
|
|
assert kept["board"] == "nano"
|
|
|
|
|
|
def test_a_screen_may_fetch_the_last_few_frames(client: TestClient, camera) -> None:
|
|
"""A remote screen fetching frame by frame is often one behind."""
|
|
controller = client.app.state.flow_controller
|
|
store = client.app.state.artifact_store
|
|
refs = []
|
|
for index in range(RECENT_DIGESTS + 2):
|
|
ref = store.put([b"frame %d" % index], media_type="image/jpeg", volatile=True)
|
|
controller.publish_message("cams.camera", ref)
|
|
refs.append(ref)
|
|
|
|
def fetch(ref: dict[str, Any]) -> int:
|
|
return client.get(f"{ARTIFACTS}/{ref['digest']}", headers=camera).status_code
|
|
|
|
assert [fetch(ref) for ref in refs[-RECENT_DIGESTS:]] == [200] * RECENT_DIGESTS
|
|
# Older than that, it is not what the screen is drawing.
|
|
assert fetch(refs[0]) == 403
|
|
assert list(event_bus.digests["cams.camera"]) == [
|
|
ref["digest"] for ref in refs[-RECENT_DIGESTS:]
|
|
]
|
|
|
|
|
|
def test_a_viewer_asks_for_a_few_cameras_at_most() -> None:
|
|
from fluksio.api.routes.flows import MAX_MEDIA_NAMES, wanted_names
|
|
|
|
names = [f"cams.c{i}" for i in range(MAX_MEDIA_NAMES + 3)]
|
|
asked = wanted_names(json.dumps({"type": "media", "names": names}), None)
|
|
# The newest, which a page lists last: a tile scrolled back to keeps working.
|
|
assert asked == set(names[-MAX_MEDIA_NAMES:])
|
|
|
|
|
|
def test_viewers_through_the_portal_count_too(
|
|
client: TestClient, superuser_token_headers: dict[str, str], camera
|
|
) -> None:
|
|
"""The tunnel carries text only, so a device there uploads over HTTP, and a
|
|
browser there asks for media like any other."""
|
|
from fluksio.cloud.connector import CloudConnector
|
|
|
|
connector = CloudConnector(client.app)
|
|
sent: dict[str, list[dict[str, Any]]] = {"dev": [], "view": []}
|
|
|
|
class Socket:
|
|
def __init__(self, stream: str) -> None:
|
|
self.stream = stream
|
|
|
|
async def send(self, text: str) -> None:
|
|
sent[self.stream].append(json.loads(text))
|
|
|
|
def texts(stream: str) -> list[dict[str, Any]]:
|
|
return [
|
|
json.loads(frame["text"])
|
|
for frame in sent[stream]
|
|
if frame["op"] == "ws_msg"
|
|
]
|
|
|
|
def demands() -> list[list[str]]:
|
|
return [t["ports"] for t in texts("dev") if t.get("type") == "demand"]
|
|
|
|
token = camera["Authorization"][7:]
|
|
device = client.portal.start_task_soon(
|
|
connector._stream,
|
|
Socket("dev"),
|
|
{"id": "dev", "path": "/api/v1/flows/ws", "query": f"token={token}"},
|
|
)
|
|
try:
|
|
snapshot = wait_for(lambda: texts("dev"))[0]
|
|
assert snapshot["uplink"] == "http"
|
|
assert snapshot["demand"] == []
|
|
|
|
ticket = _ticket(client, superuser_token_headers).split("?", 1)[1]
|
|
viewer = client.portal.start_task_soon(
|
|
connector._stream,
|
|
Socket("view"),
|
|
{"id": "view", "path": "/api/v1/flows/ws", "query": ticket},
|
|
)
|
|
wait_for(lambda: texts("view"))
|
|
asking = json.dumps({"type": "media", "names": ["cams.camera"]})
|
|
client.portal.call(_say, connector, "view", asking)
|
|
assert wait_for(demands)[-1] == ["camera"]
|
|
|
|
viewer.cancel()
|
|
wait_for(lambda: len(demands()) == 2)
|
|
assert demands()[-1] == []
|
|
finally:
|
|
device.cancel()
|
|
|
|
|
|
async def _say(connector: Any, stream: str, text: str) -> None:
|
|
"""What the hub delivers when the browser behind a stream speaks."""
|
|
connector._listeners[stream](text)
|
|
|
|
|
|
def test_a_zero_bound_turns_the_ring_off(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from fluksio.main import _volatile_ring
|
|
|
|
monkeypatch.setattr(settings, "ARTIFACT_VOLATILE_BYTES", 0)
|
|
assert _volatile_ring() is None
|
|
monkeypatch.setattr(settings, "ARTIFACT_VOLATILE_BYTES", 1000)
|
|
assert _volatile_ring() is not None
|