Add audit_drift_script/tests/test_core.py

audit_drift_script/tests/test_core.py

@@ -0,0 +1,87 @@
+import pytest
+import json
+from datetime import datetime
+from pathlib import Path
+from audit_drift_script import core
+
+@pytest.fixture
+def valid_report_dict():
+    return {
+        "timestamp": datetime.utcnow().isoformat(),
+        "run_id": "run_001",
+        "stratum": "pinned",
+        "decision": "PASS",
+        "warn_rate": 0.05,
+        "unknown_rate": 0.01
+    }
+
+@pytest.fixture
+def invalid_report_dict_missing_field():
+    return {
+        "timestamp": datetime.utcnow().isoformat(),
+        # missing run_id
+        "stratum": "pinned",
+        "decision": "PASS",
+        "warn_rate": 0.05,
+        "unknown_rate": 0.01
+    }
+
+@pytest.fixture
+def tmp_reports_dir(tmp_path, valid_report_dict):
+    dir_path = tmp_path / "reports"
+    dir_path.mkdir()
+    sub_dir = dir_path / "sub"
+    sub_dir.mkdir()
+    # create two files
+    for i, strata in enumerate(["pinned", "unpinned"], 1):
+        report_path = sub_dir / f"report_{i}.json"
+        d = valid_report_dict.copy()
+        d["stratum"] = strata
+        d["run_id"] = f"run_{i}"
+        d["decision"] = "PASS" if i == 1 else "FAIL"
+        report_path.write_text(json.dumps(d))
+    return dir_path
+
+def test_validate_report_schema_valid(valid_report_dict):
+    assert core.validate_report_schema(valid_report_dict) is True
+
+def test_validate_report_schema_invalid(invalid_report_dict_missing_field):
+    assert core.validate_report_schema(invalid_report_dict_missing_field) is False
+
+def test_reportdata_from_json(valid_report_dict):
+    report = core.ReportData.from_json(valid_report_dict)
+    assert isinstance(report, core.ReportData)
+    assert report.run_id == "run_001"
+    assert isinstance(report.timestamp, datetime)
+
+def test_aggregate_statistics(valid_report_dict):
+    reports = [core.ReportData.from_json(valid_report_dict)]
+    agg = core.aggregate_statistics(reports)
+    assert isinstance(agg, dict)
+    assert 'pinned' in agg
+    assert 'total' in agg
+
+def test_analyze_reports_directory(tmp_reports_dir):
+    result = core.analyze_reports(str(tmp_reports_dir))
+    assert isinstance(result, dict)
+    assert 'pinned' in result
+    assert 'unpinned' in result
+
+    pinned_stats = result['pinned']
+    unpinned_stats = result['unpinned']
+
+    assert pinned_stats['count'] >= 1
+    assert unpinned_stats['count'] >= 1
+
+def test_analyze_reports_with_invalid_json(tmp_path, valid_report_dict):
+    bad_file = tmp_path / "drift_report.json"
+    bad_file.write_text('{invalid json}')
+    sub_dir = tmp_path / "sub"
+    sub_dir.mkdir()
+    good_file = sub_dir / "drift_report.json"
+    good_file.write_text(json.dumps(valid_report_dict))
+
+    result = core.analyze_reports(str(tmp_path))
+    # only the good one counted
+    assert isinstance(result, dict)
+    assert sum(v['count'] for v in result.values()) == 1