74 lines
2.3 KiB
Python
74 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
|
|
from cosma_log_analyzer.bus import StdoutPublisher
|
|
from cosma_log_analyzer.main import analyze_mcap, emit
|
|
|
|
|
|
EXPECTED_RULES = {
|
|
"imu_outliers",
|
|
"watchdog_imu",
|
|
"usbl_snr_low",
|
|
"usbl_distance_spike",
|
|
"battery_low",
|
|
}
|
|
|
|
|
|
def test_analyze_fake_mcap_produces_all_rule_types(fake_mcap: Path) -> None:
|
|
anomalies = analyze_mcap(fake_mcap, subject="AUV206")
|
|
fired = {a.rule for a in anomalies}
|
|
assert EXPECTED_RULES <= fired, f"missing rules: {EXPECTED_RULES - fired}"
|
|
|
|
|
|
def test_analyze_fake_mcap_subject_propagated(fake_mcap: Path) -> None:
|
|
anomalies = analyze_mcap(fake_mcap, subject="AUV206")
|
|
assert anomalies
|
|
assert all(a.subject == "AUV206" for a in anomalies)
|
|
|
|
|
|
def test_watchdog_detects_the_gap(fake_mcap: Path) -> None:
|
|
anomalies = analyze_mcap(fake_mcap, subject="AUV206")
|
|
watchdog = [a for a in anomalies if a.rule == "watchdog_imu"]
|
|
# The fixture inserts exactly one 3s gap.
|
|
assert len(watchdog) == 1
|
|
assert watchdog[0].context["gap_s"] > 2.9
|
|
|
|
|
|
def test_battery_low_fires_once(fake_mcap: Path) -> None:
|
|
anomalies = analyze_mcap(fake_mcap, subject="AUV206")
|
|
bat = [a for a in anomalies if a.rule == "battery_low"]
|
|
assert len(bat) == 1
|
|
assert bat[0].severity == "critical"
|
|
assert bat[0].value < 13.5
|
|
|
|
|
|
def test_usbl_distance_spike_fires_once(fake_mcap: Path) -> None:
|
|
anomalies = analyze_mcap(fake_mcap, subject="AUV206")
|
|
spikes = [a for a in anomalies if a.rule == "usbl_distance_spike"]
|
|
assert len(spikes) == 1
|
|
assert spikes[0].context["delta_m"] > 50.0
|
|
|
|
|
|
def test_usbl_snr_low_fires(fake_mcap: Path) -> None:
|
|
anomalies = analyze_mcap(fake_mcap, subject="AUV206")
|
|
snr = [a for a in anomalies if a.rule == "usbl_snr_low"]
|
|
assert len(snr) >= 1
|
|
assert all(a.value < 5.0 for a in snr)
|
|
|
|
|
|
def test_stdout_publisher_emits_json_lines(fake_mcap: Path) -> None:
|
|
anomalies = analyze_mcap(fake_mcap, subject="AUV206")
|
|
buf = io.StringIO()
|
|
publisher = StdoutPublisher(stream=buf)
|
|
n = emit(anomalies, publisher)
|
|
publisher.close()
|
|
lines = [ln for ln in buf.getvalue().splitlines() if ln]
|
|
assert n == len(lines) == len(anomalies)
|
|
for line in lines:
|
|
obj = json.loads(line)
|
|
assert obj["subject"] == "AUV206"
|
|
assert obj["rule"] in EXPECTED_RULES
|