import os import time import pytest from PrimeControls import calc H = 3600000 M = 60000 T0 = 1784200000000 # fixed epoch anchor def row(state, ts, src="prov:default:/tag:Sim/AreaA/T1:/alm:Alm1", eid=None, prio=3, dp="", system=False, ack_user=None): lvl, name = calc.normalize_priority(prio) return {"event_id": eid, "source": src, "display_path": dp, "priority": lvl, "priority_name": name, "state": state, "ts": ts, "is_system": system, "ack_user": ack_user} def lifecycle(eid, src, t_active, dur_ms=None, ack_ms=None, prio=3): rows = [row("active", t_active, src, eid, prio)] if ack_ms is not None: rows.append(row("ack", t_active + ack_ms, src, eid, prio, ack_user="op1")) if dur_ms is not None: rows.append(row("clear", t_active + dur_ms, src, eid, prio)) return rows # ---------------- parse_source / normalize_priority ---------------- def test_parse_source_from_source_path(): p = calc.parse_source("prov:default:/tag:Plant/TankFarm/T101_Hi:/alm:T101_Hi", "") assert p["area"] == "TankFarm" assert p["label"] == "TankFarm/T101_Hi" def test_parse_source_display_path_wins_label_but_area_from_tag(): p = calc.parse_source("prov:default:/tag:Plant/Boiler/DrumLo:/alm:DrumLo", "North/Drum Level") assert p["label"] == "North/Drum Level" assert p["area"] == "Boiler" def test_parse_source_system_row(): p = calc.parse_source("evt:System Startup", "") assert p["label"] == "evt:System Startup" def test_normalize_priority(): assert calc.normalize_priority(3) == (3, "High") assert calc.normalize_priority("Critical") == (4, "Critical") assert calc.normalize_priority("AlarmPriority.Low") == (1, "Low") lvl, _ = calc.normalize_priority("weird") assert lvl == -1 # ---------------- correlation ---------------- def test_correlate_full_lifecycle_uuid(): rows = lifecycle("e1", "s1", T0, dur_ms=5 * M, ack_ms=2 * M) insts, mode = calc.correlate_instances(rows) assert mode == "uuid" assert len(insts) == 1 i = insts[0] assert i["active_ms"] == T0 and i["duration_ms"] == 5 * M and i["tta_ms"] == 2 * M assert not i["open"] and i["ack_user"] == "op1" def test_correlate_ack_after_clear(): rows = [row("active", T0, eid="e1"), row("clear", T0 + M, eid="e1"), row("ack", T0 + 2 * M, eid="e1")] i = calc.correlate_instances(rows)[0][0] assert i["duration_ms"] == M and i["tta_ms"] == 2 * M def test_correlate_straddler_and_open(): rows = [row("clear", T0, eid="old"), row("active", T0 + M, eid="new")] insts, _ = calc.correlate_instances(rows) by_id = dict((i["event_id"], i) for i in insts) assert by_id["old"]["active_ms"] is None and by_id["old"]["duration_ms"] is None assert by_id["new"]["open"] and by_id["new"]["clear_ms"] is None def test_correlate_negative_skew_none(): rows = [row("active", T0, eid="e1"), row("ack", T0 - M, eid="e1")] i = calc.correlate_instances(rows)[0][0] assert i["tta_ms"] is None def test_correlate_system_dropped_and_empty(): assert calc.correlate_instances([row("active", T0, system=True)])[0] == [] assert calc.correlate_instances([])[0] == [] def test_correlate_source_sequence_fallback(): rows = [row("active", T0, src="sA"), row("clear", T0 + M, src="sA"), row("active", T0 + 2 * M, src="sA"), row("active", T0 + 3 * M, src="sB")] insts, mode = calc.correlate_instances(rows) assert mode == "source-sequence" assert len(insts) == 3 closed = [i for i in insts if i["source"] == "sA" and not i["open"]] assert len(closed) == 1 and closed[0]["duration_ms"] == M def test_partition_boundary(): insts = [dict(active_ms=T0, source="a"), dict(active_ms=T0 - 1, source="b"), dict(active_ms=None, source="c")] cur, prior, orphans = calc.partition_instances(insts, T0, T0 + H) assert len(cur) == 1 and len(prior) == 1 and len(orphans) == 1 # ---------------- bins / floods ---------------- def make_insts(times, src="s1", prio=3): out = [] for i, t in enumerate(times): rows = lifecycle("e%d_%s" % (i, src), src, t, dur_ms=30000, prio=prio) out.extend(rows) insts, _ = calc.correlate_instances(out) return insts def test_rate_bins_placement_and_edges(): start = (T0 // 600000) * 600000 insts = make_insts([start, start + 600000 - 1, start + 600000]) r = calc.rate_bins(insts, start, start + 2 * 600000) assert r["bins"][0]["count"] == 2 and r["bins"][1]["count"] == 1 def test_rate_bins_empty_and_reversed(): assert calc.rate_bins([], T0, T0)["bins"] == [] assert calc.rate_bins([], T0 + 1, T0)["bins"] == [] def test_flood_hysteresis_merge_and_split(): start = (T0 // 600000) * 600000 times = [] times += [start + i for i in range(12)] # bin0: 12 (flood) times += [start + 600000 + i for i in range(7)] # bin1: 7 (>=5 continues) times += [start + 2 * 600000 + i for i in range(12)] # bin2: 12 insts = make_insts(times) r = calc.rate_bins(insts, start, start + 3 * 600000) f = calc.flood_episodes(r, insts, start, start + 3 * 600000) assert len(f["episodes"]) == 1 times2 = [start + i for i in range(12)] + [start + 2 * 600000 + i for i in range(12)] insts2 = make_insts(times2) r2 = calc.rate_bins(insts2, start, start + 3 * 600000) f2 = calc.flood_episodes(r2, insts2, start, start + 3 * 600000) assert len(f2["episodes"]) == 2 assert f2["episodes"][0]["event_count"] == 12 assert f2["pct_time_in_flood"] > 0 # ---------------- detections ---------------- def test_chattering_detects_sim_shape_not_slow(): start = T0 end = T0 + 2 * H chatter_times = [start + i * 45000 for i in range(160)] # every 45s slow_times = [start + i * 5 * M for i in range(24)] # every 5min -> 12/hr slow insts = make_insts(chatter_times, src="chat") + make_insts(slow_times, src="slow") out = calc.chattering(insts, start, end) srcs = [r["source"] for r in out] assert "chat" in srcs and "slow" not in srcs def test_fleeting_and_standing(): rows = [] for i in range(4): rows += lifecycle("f%d" % i, "fleety", T0 + i * M, dur_ms=3000) rows += lifecycle("n1", "normal", T0, dur_ms=5 * M) insts, _ = calc.correlate_instances(rows) fl = calc.fleeting(insts) assert fl["total"] == 4 and fl["sources"][0]["source"] == "fleety" now = T0 + 30 * H status = [{"source": "prov:default:/tag:A/B/C:/alm:C", "display_path": "", "priority": 3, "priority_name": "High", "active_ms": T0, "unacked": True}] st = calc.standing(status, [], now) assert st["mode"] == "status" and st["count"] == 1 st2 = calc.standing([], [i for i in insts if i["open"]], now) assert st2["mode"] == "journal" def test_mtta_mttr_exclusions_and_median(): rows = (lifecycle("a", "s", T0, dur_ms=2 * M, ack_ms=M) + lifecycle("b", "s", T0 + M, dur_ms=4 * M, ack_ms=3 * M) + lifecycle("c", "s", T0 + 2 * M, dur_ms=100 * M) + # no ack lifecycle("d", "s", T0 + 3 * M)) # open, no clear insts, _ = calc.correlate_instances(rows) mm = calc.mtta_mttr(insts, T0, T0 + H) assert mm["mtta"]["count"] == 2 and mm["mtta"]["median_ms"] == 2 * M assert mm["mttr"]["count"] == 3 and mm["mttr"]["median_ms"] == 4 * M def test_pareto_cum_pct_vs_total(): insts = make_insts([T0 + i * M for i in range(6)], src="hot") insts += make_insts([T0 + i * M for i in range(4)], src="warm") p = calc.pareto(insts) assert p["total_activations"] == 10 assert p["rows"][0]["source"] == "hot" and abs(p["rows"][0]["cum_pct"] - 60.0) < 0.01 assert abs(p["rows"][1]["cum_pct"] - 100.0) < 0.01 assert calc.top_sources(p, 5) == p["rows"][:5] def test_heatmap_placement(): os.environ["TZ"] = "UTC" time.tzset() # 2026-07-16 is a Thursday; 10:00 UTC ts = 1784196000000 # Thu Jul 16 2026 10:00:00 UTC insts = make_insts([ts, ts, ts + H]) hm = calc.heatmap(insts) assert hm["rows"][3][10] == 2 and hm["rows"][3][11] == 1 assert hm["max_count"] == 2 and hm["total"] == 3 def test_priority_distribution_extremes(): insts = make_insts([T0 + i * M for i in range(8)], src="lo", prio=1) insts += make_insts([T0 + i * M for i in range(2)], src="hi", prio=4) d = calc.priority_distribution(insts) assert d["buckets"]["low"] == 8 and d["buckets"]["high"] == 2 assert d["sum_abs_dev"] is not None assert calc.priority_distribution([])["sum_abs_dev"] is None # ---------------- health / delta ---------------- def test_health_anchors_and_grades(): h = calc.health_score(6.0, 0.0, 0, 0, 0.0, True) assert h["score"] == 100 and h["grade"] == "A" assert calc.health_score(60.0, None, 0, 0, None, True)["subs"][0]["score"] == 0 h2 = calc.health_score(None, 1.0, None, None, None, True) flood_sub = [s for s in h2["subs"] if s["key"] == "flood"][0] assert flood_sub["score"] == 80 h3 = calc.health_score(None, None, None, None, None, True) assert h3["score"] is None and h3["grade"] is None def test_health_quiet_plant(): h = calc.health_score(None, None, 0, 0, None, False) assert h["score"] is not None and h["grade"] in ("A", "B") def test_delta_semantics(): assert calc.delta(10, 0)["dir"] == "new" assert calc.delta(0, None)["dir"] == "none" assert calc.delta(102, 100)["dir"] == "flat" d = calc.delta(200, 100) assert d["dir"] == "up" and d["good"] is False d2 = calc.delta(50, 100, higher_is_worse=True) assert d2["dir"] == "down" and d2["good"] is True # ---------------- bundle ---------------- def synth_rows(): rows = [] base = T0 for i in range(20): # baseline across areas/priorities src = "prov:default:/tag:Plant/Area%d/Tag%d:/alm:A%d" % (i % 4, i, i) rows += lifecycle("b%d" % i, src, base + i * 8 * M, dur_ms=3 * M, ack_ms=M, prio=(i % 3) + 1) chat_src = "prov:default:/tag:Plant/Area1/Chatty:/alm:Chatty" for i in range(100): # every 50s for ~83 min -> ~25/hr over the 4h window rows += lifecycle("c%d" % i, chat_src, base + i * 50000, dur_ms=8000, prio=3) flood_t = base + 3 * H for i in range(18): rows += lifecycle("fl%d" % i, "prov:default:/tag:Plant/Area2/F%d:/alm:F%d" % (i, i), flood_t + i * 20000, dur_ms=M, prio=2) for i in range(5): rows += lifecycle("fle%d" % i, "prov:default:/tag:Plant/Area3/Blip:/alm:Blip", base + i * 12 * M, dur_ms=4000, prio=1) return rows def test_build_bundle_golden_and_empty(): start, end = T0, T0 + 4 * H status = [{"source": "prov:default:/tag:Plant/Area0/Stand:/alm:Stand", "display_path": "", "priority": 3, "priority_name": "High", "active_ms": T0 - 30 * H, "unacked": True}] b = calc.build_bundle(synth_rows(), status, 0, start, end, end + M) for key in ("meta", "kpis", "health", "rate", "floods", "priority", "heatmap", "mtta_mttr", "pareto", "top_sources", "chattering", "fleeting", "standing", "active_now", "insights"): assert key in b, key assert b["meta"]["activation_count"] > 0 assert len(b["floods"]["episodes"]) >= 1 assert any(r["source"].endswith("Chatty") for r in b["chattering"]) assert b["standing"]["count"] == 1 assert b["fleeting"]["total"] >= 5 assert b["health"]["grade"] is not None assert b["kpis"]["activations"]["value"] > 0 assert len(b["meta"]["areas"]) >= 4 empty = calc.build_bundle([], [], None, start, end, end) assert empty["meta"]["activation_count"] == 0 assert empty["health"]["grade"] is None assert empty["pareto"]["rows"] == [] and empty["rate"]["bins"] != [] def test_build_bundle_filters(): start, end = T0, T0 + 4 * H b = calc.build_bundle(synth_rows(), [], None, start, end, end, {"filters": {"areas": ["Area2"]}}) assert all(a == "Area2" for a in b["meta"]["areas"]) # Area2 = 5 baseline rows (i % 4 == 2) + 18 flood rows assert b["meta"]["activation_count"] == 23