Add mess_log_processing/tests/test_core.py

mess_log_processing/tests/test_core.py

@@ -0,0 +1,72 @@
+import os
+import io
+import json
+import tempfile
+from pathlib import Path
+
+import pandas as pd
+import pytest
+
+from src.mess_log_processing import core
+
+
+@pytest.fixture
+def sample_jsonl(tmp_path: Path) -> Path:
+    data = [
+        {"t_publish": 1.0, "t_gate_read": 2.0, "t_index_visible": 3.0, "pinned_flag": True, "timeouts": 0, "drift_signature": "A"},
+        {"t_publish": 2.0, "t_gate_read": 3.5, "t_index_visible": 5.0, "pinned_flag": False, "timeouts": 1, "drift_signature": "B"},
+        {"t_publish": 1.5, "t_gate_read": 2.8, "t_index_visible": 4.3, "pinned_flag": True, "timeouts": 0, "drift_signature": "A"},
+    ]
+    file_path = tmp_path / "mess_log.jsonl"
+    with open(file_path, "w", encoding="utf-8") as f:
+        for entry in data:
+            f.write(json.dumps(entry) + "\n")
+    return file_path
+
+
+@pytest.fixture
+def sample_csv(tmp_path: Path) -> Path:
+    df = pd.DataFrame(
+        [
+            {"t_publish": 1.2, "t_gate_read": 2.1, "t_index_visible": 3.3, "pinned_flag": True, "timeouts": 0, "drift_signature": "A"},
+            {"t_publish": 2.3, "t_gate_read": 3.3, "t_index_visible": 5.2, "pinned_flag": False, "timeouts": 2, "drift_signature": "B"},
+        ]
+    )
+    csv_path = tmp_path / "summary.csv"
+    df.to_csv(csv_path, index=False)
+    return csv_path
+
+
+def test_process_logs_returns_dataframe(sample_jsonl: Path, sample_csv: Path):
+    result = core.process_logs(str(sample_jsonl), str(sample_csv))
+    assert isinstance(result, pd.DataFrame)
+    assert not result.empty
+    # Validate required columns for aggregated stats
+    expected_cols = {"pinned_flag", "metric", "p50", "p95", "p99", "max"}
+    assert expected_cols.issubset(set(result.columns))
+
+
+def test_p99_calculation_consistency(sample_jsonl: Path, sample_csv: Path):
+    result = core.process_logs(str(sample_jsonl), str(sample_csv))
+    pinned_stats = result[result["pinned_flag"] == True]
+    unpinned_stats = result[result["pinned_flag"] == False]
+    # There should be different statistical profiles between pinned and unpinned
+    assert all(pinned_stats["p99"] >= pinned_stats["p50"])
+    assert all(unpinned_stats["p99"] >= unpinned_stats["p50"])
+
+
+def test_invalid_input_raises(tmp_path: Path):
+    # Passing non-existent files should raise FileNotFoundError
+    json_path = tmp_path / "missing.jsonl"
+    csv_path = tmp_path / "missing.csv"
+    with pytest.raises((FileNotFoundError, OSError)):
+        _ = core.process_logs(str(json_path), str(csv_path))
+
+
+def test_typing_and_validation(sample_jsonl: Path, sample_csv: Path):
+    # Test that invalid CSV (non-numeric fields) raises an exception on computation
+    csv_path = sample_csv
+    with open(csv_path, "a", encoding="utf-8") as f:
+        f.write("not_a_number,not_a_number,not_a_number,True,0,A\n")
+    with pytest.raises(Exception):
+        _ = core.process_logs(str(sample_jsonl), str(csv_path))