Add data_analysis_script/tests/test_core.py

data_analysis_script/tests/test_core.py

@@ -0,0 +1,50 @@
+import pytest
+import pandas as pd
+from pathlib import Path
+from data_analysis_script import core
+
+
+@pytest.fixture()
+def sample_log_file(tmp_path: Path):
+    csv_path = tmp_path / "test_run_logs.csv"
+    # Create a CSV with columns from data_models LogEntry
+    df = pd.DataFrame(
+        [
+            {"worker_start_offset": 0.0, "expires_at_dist_hours": 1.0, "retry_total_overhead_ms": 10.0},
+            {"worker_start_offset": 0.5, "expires_at_dist_hours": 1.1, "retry_total_overhead_ms": 12.0},
+            {"worker_start_offset": 1.0, "expires_at_dist_hours": 2.0, "retry_total_overhead_ms": 15.0},
+        ]
+    )
+    df.to_csv(csv_path, index=False)
+    return csv_path
+
+
+def test_analyze_logs_returns_analysisresults(sample_log_file):
+    result = core.analyze_logs(str(sample_log_file))
+    assert isinstance(result, core.AnalysisResults)
+
+    js = result.to_json()
+    assert set(js.keys()) == {"max_outlier", "band_center", "band_width"}
+    assert all(isinstance(v, float) for v in js.values())
+
+
+def test_analyze_logs_invalid_file(tmp_path):
+    invalid_path = tmp_path / "nonexistent.csv"
+    with pytest.raises(FileNotFoundError):
+        core.analyze_logs(str(invalid_path))
+
+
+def test_analyze_logs_with_empty_file(tmp_path):
+    empty_path = tmp_path / "empty.csv"
+    empty_path.write_text("worker_start_offset,expires_at_dist_hours,retry_total_overhead_ms\n")
+    with pytest.raises(ValueError):
+        core.analyze_logs(str(empty_path))
+
+
+def test_to_json_round_trip(sample_log_file):
+    result = core.analyze_logs(str(sample_log_file))
+    js_dict = result.to_json()
+    # reconstruct-like check for field equivalence
+    assert pytest.approx(result.max_outlier, rel=1e-6) == js_dict["max_outlier"]
+    assert pytest.approx(result.band_center, rel=1e-6) == js_dict["band_center"]
+    assert pytest.approx(result.band_width, rel=1e-6) == js_dict["band_width"]