55 lines
2.3 KiB
Python
55 lines
2.3 KiB
Python
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / "tools" / "hil"))
|
|
|
|
from hil_common import read_events
|
|
from report import evaluate
|
|
|
|
|
|
@pytest.mark.hil
|
|
def test_captured_baseline_evidence():
|
|
artifact_text = os.environ.get("HIL_ARTIFACT", "")
|
|
if not artifact_text:
|
|
pytest.skip("set HIL_ARTIFACT to a completed artifacts/hil/<timestamp> directory")
|
|
artifact = Path(artifact_text)
|
|
checks = evaluate(read_events(artifact))
|
|
failures = [f"{check.name}: {check.detail}" for check in checks if check.status == "fail"]
|
|
assert not failures, "\n".join(failures)
|
|
|
|
|
|
@pytest.mark.hil
|
|
def test_captured_fault_recovery_evidence():
|
|
artifact_text = os.environ.get("HIL_ARTIFACT", "")
|
|
requested_fault = os.environ.get("HIL_FAULT", "")
|
|
if not artifact_text or not requested_fault:
|
|
pytest.skip("set HIL_ARTIFACT and HIL_FAULT to a completed fault-run artifact")
|
|
checks = evaluate(read_events(Path(artifact_text)), requested_fault)
|
|
fault_check = next(check for check in checks if check.name == f"fault {requested_fault}")
|
|
assert fault_check.status == "pass", fault_check.detail
|
|
|
|
|
|
@pytest.mark.hil
|
|
def test_meter_outage_is_counted_once_and_not_published():
|
|
artifact_text = os.environ.get("HIL_ARTIFACT", "")
|
|
expect_no_meter = os.environ.get("HIL_EXPECT_NO_METER", "") == "1"
|
|
if not artifact_text or not expect_no_meter:
|
|
pytest.skip("set HIL_ARTIFACT and HIL_EXPECT_NO_METER=1 for a no-meter capture")
|
|
events = read_events(Path(artifact_text))
|
|
sender = [event for event in events if event.get("source") == "sender"]
|
|
samples = [event for event in sender if event.get("event") == "meter_sample"]
|
|
active_faults = [event for event in sender
|
|
if event.get("event") == "meter_fault" and event.get("state") == "active"]
|
|
assert samples, "no meter sample evidence captured"
|
|
assert all(event.get("ok") is False and event.get("classification") == "no_data"
|
|
for event in samples)
|
|
assert len(active_faults) == 1, active_faults
|
|
assert not [event for event in sender if event.get("event") == "batch_created"]
|
|
assert not [event for event in sender
|
|
if event.get("event") == "queue" and event.get("action") == "enqueue"]
|