Martino Ferrari
207 lines
8.3 KiB
Python
207 lines
8.3 KiB
Python
#!/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"))
|
|
|
|
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
|
|
|
|
|
|
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("<I", 1)
|
|
key = b"src:Sig"
|
|
buf += struct.pack("<H", len(key)) + key + struct.pack("<I", len(t))
|
|
for x in t:
|
|
buf += struct.pack("<d", x)
|
|
for x in v:
|
|
buf += struct.pack("<d", x)
|
|
open(p, "wb").write(buf)
|
|
r = V.read_received(p)
|
|
self.assertIn("src:Sig", r)
|
|
tt, vv = r["src:Sig"]
|
|
self.assertEqual(list(vv), v)
|
|
|
|
|
|
class TestGenCfg(unittest.TestCase):
|
|
def test_tap_routes_through_ddb(self):
|
|
sc = S.SCENARIOS[1] # s02, oracle=both
|
|
with tempfile.TemporaryDirectory() as d:
|
|
mc = os.path.join(d, "m.cfg")
|
|
C.write_marte_cfg(sc, mc, os.path.join(d, "in.bin"),
|
|
tap_bin=os.path.join(d, "tap.bin"))
|
|
cfg = open(mc).read()
|
|
self.assertIn("StreamGAM_src", cfg)
|
|
self.assertIn("TapGAM", cfg)
|
|
# TapGAM must read from the DDB (prefixed name), not the FileReader
|
|
self.assertIn("src_Wave", cfg)
|
|
tap_block = cfg[cfg.index("+TapGAM"):]
|
|
tap_in = tap_block[tap_block.index("InputSignals"):tap_block.index("OutputSignals")]
|
|
self.assertNotIn("FileReaderDS", tap_in)
|
|
|
|
def test_no_tap_direct(self):
|
|
sc = S.SCENARIOS[0] # s01, oracle=analytic
|
|
with tempfile.TemporaryDirectory() as d:
|
|
mc = os.path.join(d, "m.cfg")
|
|
C.write_marte_cfg(sc, mc, os.path.join(d, "in.bin"))
|
|
cfg = open(mc).read()
|
|
self.assertNotIn("TapGAM", cfg)
|
|
self.assertIn("ReaderGAM_src", cfg)
|
|
|
|
def test_hub_cfg_has_ports(self):
|
|
sc = S.SCENARIOS[0]
|
|
with tempfile.TemporaryDirectory() as d:
|
|
hc = os.path.join(d, "h.cfg")
|
|
C.write_hub_cfg(sc, hc)
|
|
cfg = open(hc).read()
|
|
self.assertIn(f"WSPort = {sc['ws_port']}", cfg)
|
|
self.assertIn(f"Port = {sc['sources'][0]['udp_port']}", cfg)
|
|
|
|
|
|
class TestProcPerf(unittest.TestCase):
|
|
def test_snapshot_self(self):
|
|
rec = P.snapshot(os.getpid())
|
|
self.assertTrue(rec["avail"])
|
|
self.assertIn("cpu_s", rec)
|
|
self.assertGreaterEqual(rec["cpu_s"], 0.0)
|
|
|
|
def test_snapshot_missing(self):
|
|
self.assertFalse(P.snapshot(2 ** 30).get("avail", False))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|