Add drift_report_parser/tests/test_core.py

drift_report_parser/tests/test_core.py

@@ -0,0 +1,86 @@
+import json
+import tempfile
+import os
+from pathlib import Path
+import pytest
+
+from drift_report_parser import core
+
+def _create_temp_drift_report(tmp_path: Path, content: dict) -> str:
+    file_path = tmp_path / 'drift_report.json'
+    with open(file_path, 'w', encoding='utf-8') as f:
+        json.dump(content, f)
+    return str(file_path)
+
+def test_parse_drift_report_valid(tmp_path):
+    sample_data = {
+        "runs": [
+            {
+                "timestamp": "2024-05-17T12:00:00",
+                "pinned": True,
+                "unpinned": False,
+                "decision": "PASS",
+                "rolling_warn_rate": 0.05,
+                "counts": "warn=2,fail=0,unknown=0",
+                "label": "exp_1",
+                "comment": "ok"
+            },
+            {
+                "timestamp": "2024-05-18T13:30:00",
+                "pinned": False,
+                "unpinned": True,
+                "decision": "WARN",
+                "rolling_warn_rate": 0.15,
+                "counts": "warn=3,fail=1,unknown=1",
+                "label": "exp_2",
+                "comment": None
+            }
+        ]
+    }
+    report_file = _create_temp_drift_report(tmp_path, sample_data)
+
+    result = core.parse_drift_report(report_file)
+
+    assert isinstance(result, list)
+    assert len(result) == 2
+
+    first = result[0]
+    assert isinstance(first, dict)
+    assert first["timestamp"].startswith("2024-05-17")
+    assert first["decision"] == "PASS"
+    assert first["pinned"] is True
+    assert isinstance(first["rolling_warn_rate"], float)
+    assert "counts" in first
+
+def test_parse_drift_report_missing_field(tmp_path):
+    # Missing some optional fields
+    sample_data = {
+        "runs": [
+            {"timestamp": "2024-05-19T10:00:00", "pinned": False, "unpinned": False, "decision": "FAIL"}
+        ]
+    }
+    report_file = _create_temp_drift_report(tmp_path, sample_data)
+
+    result = core.parse_drift_report(report_file)
+    assert len(result) == 1
+    item = result[0]
+    assert item["decision"] == "FAIL"
+    assert "rolling_warn_rate" in item
+
+def test_parse_drift_report_invalid_json(tmp_path):
+    invalid_file = tmp_path / "invalid.json"
+    invalid_file.write_text("{ not-a-json }", encoding="utf-8")
+
+    with pytest.raises(Exception):
+        core.parse_drift_report(str(invalid_file))
+
+def test_parse_drift_report_file_not_exists():
+    with pytest.raises(FileNotFoundError):
+        core.parse_drift_report("non_existent_file.json")
+
+def test_parse_drift_report_type_validation(tmp_path):
+    sample_data = {"runs": [{"timestamp": 12345, "pinned": "yes", "decision": "PASS"}]}
+    report_file = _create_temp_drift_report(tmp_path, sample_data)
+
+    with pytest.raises(Exception):
+        core.parse_drift_report(report_file)