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)