From 0ab49078cc51cf6ad9d0bbe1198a580ddec47b03 Mon Sep 17 00:00:00 2001 From: Martino Ferrari Date: Thu, 25 Jun 2026 08:56:57 +0200 Subject: [PATCH] test(e2e-chain): scenario model + starter scenarios Co-Authored-By: Claude Opus 4.6 --- Test/E2E/chain/scenarios.py | 209 ++++++++++++++++++++++++++++++++++++ 1 file changed, 209 insertions(+) create mode 100644 Test/E2E/chain/scenarios.py diff --git a/Test/E2E/chain/scenarios.py b/Test/E2E/chain/scenarios.py new file mode 100644 index 0000000..2adbb4c --- /dev/null +++ b/Test/E2E/chain/scenarios.py @@ -0,0 +1,209 @@ +#!/usr/bin/env python3 +""" +scenarios.py — Declarative scenario matrix for the streaming-chain E2E suite. + +Each scenario describes one full-chain run: + MARTe2 App (UDPStreamer) -> StreamHub -> mock client + +The matrix is a *curated covering set*: every configurable option value of the +UDPStreamer appears in at least one scenario, plus deliberately chosen +high-risk interactions. gen_data.py / gen_cfg.py / client / validate_waveform.py +all consume the scenario dict schema documented below. + +Scenario dict schema +-------------------- +{ + "id": str # unique slug, also the artifact basename + "desc": str # human-readable one-liner for the report + "network": "unicast" | "multicast" + "publishing": "Strict" | "Accumulate" | "Decimate" + "ratio": int | None # Decimate only (>=1) + "min_refresh_hz": float | None # Accumulate only (>0) + "max_payload": int # MaxPayloadSize bytes + "ws_port": int # StreamHub WSPort (unique per scenario) + "sources": [ # one or more UDPStreamer blocks + { + "id": str # source/session id (hub Source key) + "udp_port": int # UDPStreamer Port (unique per scenario) + "data_port": int | None # multicast DATA port + "multicast_group": str | None # e.g. "239.0.0.1" + "signals": [ + { + "name": str + "type": one of TYPE_CODES keys + "elements": int # NumberOfElements (1 = scalar) + "time_mode": "PacketTime"|"FullArray"|"FirstSample"|"LastSample" + "time_signal": str | None # name of the TimeSignal (when needed) + "sampling_rate": float | None # Hz, for First/LastSample + "quant": "none"|"uint8"|"int8"|"uint16"|"int16" + "range_min": float | None # quant only + "range_max": float | None # quant only + "unit": str | None # e.g. "V", "us", "ns" + "formula": "sine"|"ramp"|"counter"|"time_ns"|"time_us" + "freq": float | None # sine frequency (Hz over row index) + "is_time": bool # True if this signal is a TimeSignal + }, ... + ] + }, ... + ], + "oracle": "analytic" | "fed" | "both" + "client_checks": subset of ["live","zoom","window","trigger"] + "trig_signal": "src:sig" | None # which signal to trigger on +} +""" + +# MARTe2 TypeDescriptor.all codes: all = (type<<2) | (bits<<6), +# BasicType SignedInteger=0, UnsignedInteger=1, Float=2 (verified against +# MARTe2 L0Types/BasicType.h and L1Portability/TypeDescriptor.h). +TYPE_CODES = { + "int8": 0x0200, + "uint8": 0x0204, + "int16": 0x0400, + "uint16": 0x0404, + "int32": 0x0800, + "uint32": 0x0804, + "int64": 0x1000, + "uint64": 0x1004, + "float32": 0x0808, + "float64": 0x1008, +} + +# numpy dtype per MARTe type (little-endian). +NP_DTYPE = { + "int8": "= 1): + errs.append("Decimate needs ratio>=1") + if s["publishing"] == "Accumulate" and not (s.get("min_refresh_hz") and + s["min_refresh_hz"] > 0): + errs.append("Accumulate needs min_refresh_hz>0") + if not s.get("sources"): + errs.append("no sources") + for src in s.get("sources", []): + if s["network"] == "multicast": + if not src.get("multicast_group") or not src.get("data_port"): + errs.append(f"src {src['id']}: multicast needs group+data_port") + names = {sig["name"] for sig in src["signals"]} + for sig in src["signals"]: + if sig["type"] not in TYPE_CODES: + errs.append(f"{sig['name']}: bad type {sig['type']}") + if sig["quant"] not in QUANT_TYPES: + errs.append(f"{sig['name']}: bad quant {sig['quant']}") + if sig["quant"] != "none": + if sig["type"] not in FLOAT_TYPES: + errs.append(f"{sig['name']}: quant only on float types") + if sig["range_min"] is None or sig["range_max"] is None: + errs.append(f"{sig['name']}: quant needs range_min/max") + if sig["time_mode"] not in TIME_MODES: + errs.append(f"{sig['name']}: bad time_mode") + if sig["time_mode"] != "PacketTime": + if not sig["time_signal"] or sig["time_signal"] not in names: + errs.append(f"{sig['name']}: time_mode needs valid time_signal") + if sig["time_mode"] == "FullArray": + ts = next((x for x in src["signals"] + if x["name"] == sig["time_signal"]), None) + if ts and ts["elements"] != sig["elements"]: + errs.append(f"{sig['name']}: FullArray time_signal length mismatch") + if sig["time_mode"] in ("FirstSample", "LastSample"): + if not sig["sampling_rate"] or sig["sampling_rate"] <= 0: + errs.append(f"{sig['name']}: First/LastSample needs sampling_rate>0") + return errs + + +# ── Starter set (expanded to the full covering matrix in Task 8) ────────────── +SCENARIOS = [ + { + "id": "s01_scalar_uint32", + "desc": "Single uint32 scalar counter, Strict unicast (type fidelity)", + "network": "unicast", "publishing": "Strict", + "ratio": None, "min_refresh_hz": None, "max_payload": 1400, + "ws_port": 8101, + "sources": [{ + "id": "src", "udp_port": 44610, "data_port": None, + "multicast_group": None, + "signals": [ + _sig("Counter", "uint32", 1, formula="counter"), + _sig("Sine", "float32", 1, formula="sine", freq=2.0, unit="V"), + ], + }], + "oracle": "analytic", + "client_checks": ["live", "zoom", "window", "trigger"], + "trig_signal": "src:Sine", + }, + { + "id": "s02_array_float32_fullarray", + "desc": "100-elem float32 array, FullArray time mode, uint64 ns time array", + "network": "unicast", "publishing": "Strict", + "ratio": None, "min_refresh_hz": None, "max_payload": 65507, + "ws_port": 8102, + "sources": [{ + "id": "src", "udp_port": 44612, "data_port": None, + "multicast_group": None, + "signals": [ + _sig("TimeArr", "uint64", 100, unit="ns", + formula="time_ns", is_time=True), + _sig("Wave", "float32", 100, time_mode="FullArray", + time_signal="TimeArr", formula="sine", freq=5.0, unit="V"), + ], + }], + "oracle": "both", + "client_checks": ["live", "zoom"], + "trig_signal": None, + }, + { + "id": "s03_quant_uint16", + "desc": "float32 scalar quantised to uint16 over [-5,5], Strict unicast", + "network": "unicast", "publishing": "Strict", + "ratio": None, "min_refresh_hz": None, "max_payload": 1400, + "ws_port": 8103, + "sources": [{ + "id": "src", "udp_port": 44614, "data_port": None, + "multicast_group": None, + "signals": [ + _sig("Sine", "float32", 1, quant="uint16", + range_min=-5.0, range_max=5.0, formula="sine", + freq=3.0, unit="V"), + ], + }], + "oracle": "analytic", + "client_checks": ["live", "zoom"], + "trig_signal": "src:Sine", + }, +] + + +if __name__ == "__main__": + ok = True + for s in SCENARIOS: + errs = validate_scenario(s) + print(f"{s['id']:32s} {'OK' if not errs else errs}") + ok = ok and not errs + print(f"\n{len(SCENARIOS)} scenarios, {'ALL VALID' if ok else 'INVALID PRESENT'}")