Files
app/backend/tests/flow/test_messages.py
T
stroblmeandClaude Opus 5 0ffcabfdb9
Docs / docs (push) Successful in 30s
Playwright Tests / test-playwright (1, 2) (push) Successful in 3m7s
Playwright Tests / test-playwright (2, 2) (push) Successful in 1m54s
pre-commit / pre-commit (push) Failing after 4m24s
Test Backend / test-backend (push) Successful in 3m8s
Compose Smoke Test / test-compose (push) Successful in 40s
Playwright Tests / merge-reports (push) Successful in 1m33s
Media dtypes: image, audio and video as narrowed artifact references
A port may now declare `image`, `audio` or `video`. Each is the artifact
reference the engine already had, narrowed by the `media_type` on it, so a
speech recogniser declares what it eats rather than taking any bytes at all and
finding out. Bytes still never travel as a message and nothing on the wire
stops being JSON: a camera publishes one reference per frame, a microphone one
per chunk, and a reference may carry a `meta` dict nothing here interprets.

Streaming media is therefore an ordinary streaming port — with one change to
what that means. An emission used to journal an item with no payload, so
downstream read whatever was current when the item was claimed; a consumer
slower than its producer saw only the newest chunk and the ones between were
lost. That is right for a training curve and wrong for a second of speech, so
an emission now journals a `kind="emission"` item carrying its values, and the
executor hands them to the nodes reading that message instead of writing them
to state again. The value in state stays the latest, which is what everything
else reads, and the wave is filtered by what actually changed rather than
walking everything reachable. No queue serialization change — the existing
`outputs` field carries it.

Continuous media makes the store's missing GC a real problem, so this closes
it: `sweep_artifacts` runs hourly, keeps every digest a `run_artifact` row
records or a live message holds, spares anything written in the last hour, and
stands aside entirely while a run is in flight, since a node may store a
checkpoint long before it returns the reference to it. That also collects the
orphans a deleted flow has always left behind. `ARTIFACT_GC_INTERVAL_S=0` turns
it off.

Around the edges: `GET /artifacts/{digest}` serves the media type the caller
passes and answers ranged requests, so a browser plays a clip rather than
downloading it; `PUT` spools to disk instead of holding the whole body in
memory, as does `save_artifact` given a path; a Media widget draws whatever its
message points at, and a wall panel may fetch the bytes its own tiles are
showing and nothing else; and a connector gets `save_artifact`, for a device
whose readings are bytes.

What this cannot do is live video: a frame every second or two is a glance, and
the honest answer above that is the camera's own stream, which the widget takes
as a URL and the browser plays from source.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-26 23:44:55 +02:00

204 lines
6.7 KiB
Python

import json
import pytest
from fluksio.flow.messages import DType, MessageSpec, flow_of, qualify
def test_port_defaults_to_last_name_segment():
assert MessageSpec(name="temperature").port == "temperature"
assert MessageSpec(name="heating.temperature").port == "temperature"
assert MessageSpec(name="heating.temperature", port="t").port == "t"
def test_int_rejects_bool():
# bool is a subclass of int, but a flag is not a number here.
spec = MessageSpec(name="count", dtype=DType.INT)
spec.check(3)
with pytest.raises(TypeError):
spec.check(True)
def test_float_accepts_int_but_not_bool():
spec = MessageSpec(name="temp", dtype=DType.FLOAT)
spec.check(21)
spec.check(21.5)
with pytest.raises(TypeError):
spec.check(True)
with pytest.raises(TypeError):
spec.check("21")
def test_json_dtype_round_trips():
spec = MessageSpec(name="payload", dtype=DType.JSON)
value = {"a": [1, 2], "b": None}
spec.check(value)
assert json.loads(json.dumps(value)) == value
def test_record_takes_flat_scalars_only():
spec = MessageSpec(name="notice", dtype=DType.RECORD)
spec.check({"title": "Boiler", "count": 3, "hot": True, "detail": None})
for wrong in ({"a": {"nested": 1}}, {"a": [1]}, [], "x"):
with pytest.raises(TypeError):
spec.check(wrong)
def test_series_carries_lines_and_whatever_else_the_answer_echoes():
spec = MessageSpec(name="temps", dtype=DType.SERIES)
spec.check(
{
"range_s": 3600,
"interval_s": 60,
"lines": [{"label": "living", "points": [[1.0, 21.5], [2.0, 21.6]]}],
}
)
spec.check({"lines": []})
for wrong in (
{"lines": [{"label": "a", "points": [[1.0, 2.0, 3.0]]}]},
{"lines": [{"label": "a", "points": [["1", 2.0]]}]},
{"lines": [{"label": 1, "points": []}]},
{"lines": {}},
{},
):
with pytest.raises(TypeError):
spec.check(wrong)
def test_list_items_follow_the_declared_shape():
records = MessageSpec(name="agenda", dtype=DType.LIST)
records.check([{"title": "Dentist", "ts": 1.0}])
records.check([])
with pytest.raises(TypeError):
records.check([1.0])
numbers = MessageSpec(name="window", dtype=DType.LIST, item=DType.FLOAT)
numbers.check([21.4, 2])
for wrong in ([True], ["1"], 1.0):
with pytest.raises(TypeError):
numbers.check(wrong)
def test_the_failing_item_is_named():
spec = MessageSpec(name="window", dtype=DType.LIST, item=DType.FLOAT)
with pytest.raises(TypeError, match="index 1"):
spec.check([1.0, "x"])
def test_a_list_holds_one_declared_level():
for item in (DType.LIST, DType.SERIES):
with pytest.raises(ValueError):
MessageSpec(name="x", dtype=DType.LIST, item=item)
def _ref(media_type: str) -> dict:
return {
"digest": "sha256:" + "ab" * 32,
"size": 12,
"media_type": media_type,
"name": "chunk",
}
def test_a_media_port_checks_the_family():
spec = MessageSpec(name="speech", dtype=DType.AUDIO)
spec.check(_ref("audio/wav"))
for wrong in (_ref("video/mp4"), _ref(""), {"a": 1}, "sha256:x"):
with pytest.raises(TypeError):
spec.check(wrong)
def test_a_media_mismatch_names_the_media_type():
spec = MessageSpec(name="speech", dtype=DType.AUDIO)
with pytest.raises(TypeError, match="video/mp4"):
spec.check(_ref("video/mp4"))
def test_an_artifact_port_accepts_a_media_reference():
# Media narrows artifact, so the wider port still takes it; the reverse is
# what the family check refuses.
MessageSpec(name="blob", dtype=DType.ARTIFACT).check(_ref("image/png"))
with pytest.raises(TypeError):
MessageSpec(name="frame", dtype=DType.IMAGE).check(
{"digest": "sha256:" + "cd" * 32, "size": 1}
)
def test_a_media_reference_may_carry_meta():
spec = MessageSpec(name="frame", dtype=DType.IMAGE)
spec.check({**_ref("image/png"), "meta": {"width": 640, "seq": 3}})
def test_a_list_refuses_media_items():
for item in (DType.IMAGE, DType.AUDIO, DType.VIDEO):
with pytest.raises(ValueError):
MessageSpec(name="x", dtype=DType.LIST, item=item)
def test_coerce_parses_a_media_reference_from_text():
spec = MessageSpec(name="frame", dtype=DType.IMAGE)
assert spec.coerce(json.dumps(_ref("image/png"))) == _ref("image/png")
def test_coerce_parses_structured_text():
spec = MessageSpec(name="notice", dtype=DType.RECORD)
assert spec.coerce('{"title": "Boiler"}') == {"title": "Boiler"}
assert spec.coerce({"title": "Boiler"}) == {"title": "Boiler"}
def test_coerce_from_text():
assert MessageSpec(name="a", dtype=DType.FLOAT).coerce("2.5") == 2.5
assert MessageSpec(name="a", dtype=DType.INT).coerce("7") == 7
assert MessageSpec(name="a", dtype=DType.BOOL).coerce("yes") is True
assert MessageSpec(name="a", dtype=DType.BOOL).coerce("0") is False
def test_spec_serializes():
spec = MessageSpec(name="heating.temp", dtype=DType.FLOAT)
assert MessageSpec.model_validate_json(spec.model_dump_json()) == spec
def test_qualify_scopes_bare_names_only():
assert qualify("heating", "temp") == "heating.temp"
assert qualify("heating", "solar.power") == "solar.power"
assert qualify("heating", "") == ""
assert flow_of("heating.temp") == "heating"
# -----------------------------------------------------------------------------
# NaN and infinity
#
# JSON cannot spell either, so one travelling through a port would come back as
# a response nobody can parse, a socket frame that stops a canvas, or a row the
# database rejects — a long way from the node that produced it.
# -----------------------------------------------------------------------------
def test_a_float_port_refuses_nan_and_infinity():
spec = MessageSpec(name="score", dtype=DType.FLOAT)
spec.check(0.5)
for value in (float("nan"), float("inf"), float("-inf")):
with pytest.raises(TypeError, match="score"):
spec.check(value)
def test_a_json_port_refuses_a_nan_nested_in_it():
spec = MessageSpec(name="report", dtype=DType.JSON)
spec.check({"groups": [{"mean": 1.0}]})
with pytest.raises(TypeError, match="JSON"):
spec.check({"groups": [{"mean": float("nan")}]})
def test_a_series_refuses_a_nan_point():
spec = MessageSpec(name="curve", dtype=DType.SERIES)
lines = [{"label": "loss", "points": [[1.0, float("nan")]]}]
with pytest.raises(TypeError):
spec.check({"lines": lines})
def test_a_value_that_refers_to_itself_does_not_hang_the_check():
spec = MessageSpec(name="report", dtype=DType.JSON)
loop: dict = {}
loop["self"] = loop
spec.check(loop)