#!/usr/bin/env python3 """ tests_py.py — Unit tests for the streaming-chain E2E Python framework. Run directly (``python3 -m unittest tests_py``) or, for coverage, via collect.py which wraps it in ``coverage run``. These exercise the pure logic of the generators, the waveform oracle, the config emitter, the RCV1 reader and the /proc perf sampler — i.e. everything the orchestrator depends on, without needing a live MARTe/StreamHub stack. """ import os import struct import sys import tempfile import unittest import numpy as np sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import scenarios as S import gen_data as G import gen_cfg as C import validate_waveform as V import proc_perf as P class TestScenarios(unittest.TestCase): def test_all_starter_scenarios_valid(self): for s in S.SCENARIOS: self.assertEqual(S.validate_scenario(s), [], f"{s['id']} invalid") def test_seamless_loop_constraint(self): s = { "id": "x", "network": "unicast", "publishing": "Strict", "ratio": None, "min_refresh_hz": None, "max_payload": 1400, "ws_port": 9000, "sources": [{"id": "s", "udp_port": 1, "data_port": None, "multicast_group": None, "signals": [S._sig("Bad", "float32", freq=3.0)]}], "oracle": "analytic", "client_checks": ["live"], "trig_signal": None, } errs = S.validate_scenario(s) self.assertTrue(any("multiple of LOOP_HZ" in e for e in errs), errs) s["sources"][0]["signals"][0]["freq"] = S.LOOP_HZ * 2 self.assertEqual(S.validate_scenario(s), []) def test_bad_quant_rejected(self): s = { "id": "x", "network": "unicast", "publishing": "Strict", "ratio": None, "min_refresh_hz": None, "max_payload": 1400, "ws_port": 9000, "sources": [{"id": "s", "udp_port": 1, "data_port": None, "multicast_group": None, "signals": [S._sig("Q", "int32", quant="uint8", range_min=0, range_max=1, formula="ramp")]}], "oracle": "analytic", "client_checks": ["live"], "trig_signal": None, } errs = S.validate_scenario(s) self.assertTrue(any("quant only on float" in e for e in errs), errs) def test_all_scenarios_have_kind(self): for s in S.SCENARIOS: self.assertIn("kind", s, f"{s['id']} missing kind") self.assertIn(s["kind"], ("chain", "direct", "recorder", "debug", "tcplogger", "debug_pause_resume")) def test_direct_and_recorder_scenarios_present(self): kinds = {s["kind"] for s in S.SCENARIOS} self.assertIn("direct", kinds) self.assertIn("recorder", kinds) direct_ids = {s["id"] for s in S.SCENARIOS if s["kind"] == "direct"} self.assertEqual(direct_ids, {"s52_direct_unicast", "s53_direct_multicast"}) recorder_ids = {s["id"] for s in S.SCENARIOS if s["kind"] == "recorder"} self.assertEqual(recorder_ids, {"s54_recorder"}) def test_validate_scenario_accepts_all_kinds(self): for s in S.SCENARIOS: S.validate_scenario(s) # must not raise def test_debug_and_tcplogger_scenarios_present(self): kinds = {s["kind"] for s in S.SCENARIOS} self.assertIn("debug", kinds) self.assertIn("tcplogger", kinds) debug_ids = {s["id"] for s in S.SCENARIOS if s["kind"] == "debug"} self.assertEqual(debug_ids, {"s55_debug_force_trace_break"}) tcplogger_ids = {s["id"] for s in S.SCENARIOS if s["kind"] == "tcplogger"} self.assertEqual(tcplogger_ids, {"s56_tcplogger_delivery"}) def test_build_by_kind_filters_correctly(self): import report_build as R results = {"scenarios": [ {"id": "a", "kind": "direct", "status": "PASS"}, {"id": "b", "kind": "chain", "status": "PASS"}, {"id": "c", "kind": "direct", "status": "FAIL"}, ]} d = R.build_by_kind(results, "direct") self.assertEqual(d["n_pass"], 1) self.assertEqual(d["n_fail"], 1) self.assertEqual(d["n_total"], 2) class TestGenData(unittest.TestCase): def test_ground_truth_shapes(self): gt = G.build_ground_truth(S.SCENARIOS[1]) # s02: TimeArr+Wave, 100 elem wave = gt["src:Wave"] self.assertEqual(wave["elements"], 100) self.assertEqual(wave["v"].size, S.NUM_ROWS * 100) self.assertEqual(wave["t"].size, S.NUM_ROWS * 100) def test_binary_roundtrip(self): sc = S.SCENARIOS[0] # s01: Counter(uint32)+Sine(float32) with tempfile.TemporaryDirectory() as d: p = os.path.join(d, "in.bin") gt = G.write_input(sc, p) cols = V.read_marte_binary(p) self.assertIn("Counter", cols) self.assertIn("Sine", cols) # counter is row index → first/last match ground truth self.assertEqual(cols["Counter"].reshape(-1)[0], gt["src:Counter"]["v"][0]) self.assertEqual(int(cols["Counter"].reshape(-1)[5]), 5) class TestOracle(unittest.TestCase): def _gt(self, **kw): t = np.linspace(0, 1, 200) base = {"v": np.sin(2 * np.pi * 5 * t), "type": "float32", "quant": "none", "formula": "sine", "freq": 5.0, "range_min": -1.0, "range_max": 1.0, "elements": 1, "rows": 200, "is_time": False, "t": t, "dt": t[1] - t[0]} base.update(kw) return base, t def test_quant_tol_is_one_level(self): gt, _ = self._gt(quant="uint16", range_min=-5.0, range_max=5.0) tol, step = V._tol(gt) self.assertAlmostEqual(step, 10.0 / 65535, places=9) self.assertGreaterEqual(tol, step) # one full level, not half def test_good_passes_bad_fails(self): gt, t = self._gt(quant="uint16") _, step = V._tol(gt) good = gt["v"] + (np.random.rand(200) - 0.5) * step bad = gt["v"] + (np.random.rand(200) - 0.5) * step * 50 self.assertTrue(V.compare_signal(gt, t, good)["pass"]) self.assertFalse(V.compare_signal(gt, t, bad)["pass"]) def test_wrong_frequency_fails_shape(self): gt, t = self._gt() wrong = np.sin(2 * np.pi * 50 * t) # 10x frequency m = V.compare_signal(gt, t, wrong) self.assertFalse(m["shape_ok"], m) def test_integer_offgrid_fails(self): gt, t = self._gt(type="uint32", quant="none", v=np.arange(200, dtype=np.float64), formula="counter") self.assertTrue(V.compare_signal(gt, t, np.arange(50, 150, dtype=float))["pass"]) self.assertFalse(V.compare_signal(gt, t, np.array([1.5, 999.0]))["pass"]) class TestRCV1(unittest.TestCase): def test_read_received(self): with tempfile.TemporaryDirectory() as d: p = os.path.join(d, "r.bin") t = [0.0, 0.1, 0.2] v = [1.0, 2.0, 3.0] buf = bytearray(b"RCV1") buf += struct.pack("