diff --git a/analysis_tool/tests/test_core.py b/analysis_tool/tests/test_core.py new file mode 100644 index 0000000..391b977 --- /dev/null +++ b/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"}