Add analysis_tool/tests/test_core.py

analysis_tool/tests/test_core.py

@@ -0,0 +1,50 @@
+import pytest
+import pandas as pd
+from analysis_tool import core
+
+
+@pytest.fixture
+def sample_log_entries():
+    # Valid entries: freeze_ok True, setup_fingerprint aux=2 or aux=3
+    data = [
+        {"timestamp": "2024-05-01T10:00:00", "measured_p": 100.0, "freeze_ok": True, "setup_fingerprint": "aux=2", "policy_hash": "hashA"},
+        {"timestamp": "2024-05-01T10:01:00", "measured_p": 105.0, "freeze_ok": True, "setup_fingerprint": "aux=2", "policy_hash": "hashA"},
+        {"timestamp": "2024-05-01T10:02:00", "measured_p": 110.0, "freeze_ok": True, "setup_fingerprint": "aux=3", "policy_hash": "hashA"},
+        {"timestamp": "2024-05-01T10:03:00", "measured_p": 120.0, "freeze_ok": True, "setup_fingerprint": "aux=3", "policy_hash": "hashA"},
+        # Invalid (freeze_ok=False) should be ignored
+        {"timestamp": "2024-05-01T10:04:00", "measured_p": 999.0, "freeze_ok": False, "setup_fingerprint": "aux=2", "policy_hash": "hashA"},
+    ]
+    return data
+
+
+def test_analyze_results_basic(sample_log_entries):
+    summary = core.analyze_results(sample_log_entries)
+    assert isinstance(summary, core.AnalysisSummary)
+    assert summary.valid is True
+    # Median of all valid measured_p: [100,105,110,120] -> (105+110)/2=107.5
+    assert pytest.approx(summary.median, rel=1e-3) == 107.5
+    # IQR: Q3(115)-Q1(102.5)=12.5
+    assert pytest.approx(summary.IQR, rel=1e-3) == 12.5
+    # Band width: max-min=120-100=20
+    assert pytest.approx(summary.band_width, rel=1e-3) == 20.0
+    # Delta band width: (aux3=[110,120]->10) - (aux2=[100,105]->5)=5
+    assert pytest.approx(summary.delta_band_width, rel=1e-3) == 5.0
+
+
+def test_invalid_or_empty_data():
+    empty_summary = core.analyze_results([])
+    assert isinstance(empty_summary, core.AnalysisSummary)
+    assert empty_summary.valid is False
+
+    invalid_entries = [{"timestamp": "x", "measured_p": None, "freeze_ok": True, "setup_fingerprint": "aux=2", "policy_hash": "y"}]
+    result = core.analyze_results(invalid_entries)
+    assert result.valid is False
+
+
+def test_dataframe_roundtrip(sample_log_entries, tmp_path):
+    df = pd.DataFrame(sample_log_entries)
+    csv_path = tmp_path / "preflight_log_test.csv"
+    df.to_csv(csv_path, index=False)
+    df_loaded = pd.read_csv(csv_path)
+    assert not df_loaded.empty
+    assert set(df_loaded.columns) == {"timestamp", "measured_p", "freeze_ok", "setup_fingerprint", "policy_hash"}