fix: fixed tests for stable one.

Author: SculptTechProject

tests/test_orchestrator_rates.py

@@ -7,25 +7,31 @@ def test_per_sensor_rate_hz_controls_sampling_frequency():
     devices_cfg = [{
         "id": "A",
         "sensors": [
-            {"kind": "temp",      "count": 1, "priority": 5, "rate_hz": 5},   # every 0.2s
-            {"kind": "vibration", "count": 1, "priority": 5, "rate_hz": 10},  # every 0.1s
+            {"kind": "temp",      "count": 1, "priority": 5, "rate_hz": 5},   # ~every 0.2s
+            {"kind": "vibration", "count": 1, "priority": 5, "rate_hz": 10},  # ~every 0.1s
         ]
     }]
 
     run_stream(
         spec_str="device=A: temp*1,vibration*1",
-        rate_hz=10.0,              # global tick
+        rate_hz=10.0,                 # global tick
         duration_s=None,
-        total_count=20,            # ~2.0s
+        total_count=20,               # ~2.0s
         writer_for_type={"*": cap.router({"*": cap.writer_for_type("*")})},
         partition_by="none",
         devices_cfg=devices_cfg,
     )
 
-    n_temp = len([r for r in cap.per_type["*"] if r["type"] == "temp"]) \
-             if "*" in cap.per_type else len([r for r in cap.per_type.get("temp", [])])
-    n_vib  = len([r for r in cap.per_type["*"] if r["type"] == "vibration"]) \
-             if "*" in cap.per_type else len([r for r in cap.per_type.get("vibration", [])])
+    # Count how many samples we collected for each type
+    all_recs = cap.per_type.get("*")
+    if all_recs is not None:
+        n_temp = sum(1 for r in all_recs if r["type"] == "temp")
+        n_vib  = sum(1 for r in all_recs if r["type"] == "vibration")
+    else:
+        n_temp = len(cap.per_type.get("temp", []))
+        n_vib  = len(cap.per_type.get("vibration", []))
 
-    # vibration should have noticeably more samples than temp (timing tolerance)
-    assert n_vib >= n_temp + 5, f"expected more vib samples than temp; got temp={n_temp}, vib={n_vib}"
+    # Expect significantly more vibration samples (10 Hz) than temperature (5 Hz).
+    # Leave tolerance for CI timing jitter: require at least 1.5× more vib or +3 samples.
+    assert n_vib >= max(int(1.5 * n_temp), n_temp + 3), \
+        f"expected clearly more vib samples than temp; got temp={n_temp}, vib={n_vib}"

tests/test_vibration.py

@@ -1,5 +1,35 @@
+from __future__ import annotations
+import math
 from dummysensors import VibrationSensor
+
 def test_vibration_basic():
-    s = VibrationSensor(base_hz=10.0, amp=1.0)
-    xs = [s.read(t_s=i*0.01) for i in range(100)]
-    assert max(xs) <= 1.0 + 1e-9 and min(xs) >= -1.0 - 1e-9
+    import math
+    amp = 1.0
+    vib = VibrationSensor(
+        base_hz=8.0,
+        amp=amp,
+        noise_theta=2.0,
+        noise_sigma=0.05,
+        spike_prob=0.0,      # ← disable spikes for a stable test
+        spike_scale=4.0,
+    )
+    dt = 0.001
+    n = 1500
+    vals = [vib.read(i * dt) for i in range(n)]
+
+    assert all(math.isfinite(v) for v in vals)
+    mean = sum(vals) / len(vals)
+    assert abs(mean) < 0.2
+
+    var = sum((v - mean) ** 2 for v in vals) / len(vals)
+    std = math.sqrt(var)
+    assert 0.4 <= std <= 1.2
+
+    # allow for OU noise, but not spikes
+    mx, mn = max(vals), min(vals)
+    max_allowed = amp + 4 * 0.05
+    min_allowed = -amp - 4 * 0.05
+    assert mn >= min_allowed and mx <= max_allowed, (
+        f"values exceeded expected bounds: "
+        f"min={mn}, max={mx}, allowed=({min_allowed}, {max_allowed})"
+    )