Add run_summary_export/tests/test_core.py

run_summary_export/tests/test_core.py

@@ -0,0 +1,88 @@
+import json
+import csv
+import os
+import io
+import pytest
+from pathlib import Path
+
+import run_summary_export.core as core
+
+
+@pytest.fixture
+def sample_run_summary_list():
+    return [
+        core.RunSummary(
+            run_id="run_001",
+            p95=10.5,
+            p99=15.2,
+            retry_free_count=100,
+            publish_reorder_count=2,
+        ),
+        core.RunSummary(
+            run_id="run_002",
+            p95=9.8,
+            p99=14.7,
+            retry_free_count=120,
+            publish_reorder_count=1,
+        ),
+    ]
+
+
+def test_run_summary_fields(sample_run_summary_list):
+    rs = sample_run_summary_list[0]
+    assert isinstance(rs.run_id, str)
+    assert isinstance(rs.p95, float)
+    assert isinstance(rs.p99, float)
+    assert isinstance(rs.retry_free_count, int)
+    assert isinstance(rs.publish_reorder_count, int)
+
+
+def test_export_summary_json(tmp_path, sample_run_summary_list):
+    out_file = tmp_path / "summary.json"
+    result_path = core.export_summary(sample_run_summary_list, format="json")
+    # Datei existiert
+    assert result_path is not None
+    assert os.path.exists(result_path)
+
+    with open(result_path, 'r', encoding='utf-8') as f:
+        data = json.load(f)
+    assert isinstance(data, list)
+    assert all(set(d.keys()) >= {"run_id", "p95", "p99", "retry_free_count", "publish_reorder_count"} for d in data)
+
+
+def test_export_summary_csv(tmp_path, sample_run_summary_list):
+    result_path = core.export_summary(sample_run_summary_list, format="csv")
+    assert result_path is not None
+    assert os.path.exists(result_path)
+
+    with open(result_path, 'r', encoding='utf-8') as f:
+        reader = csv.DictReader(f)
+        rows = list(reader)
+
+    assert len(rows) == len(sample_run_summary_list)
+    header = reader.fieldnames
+    assert set(header) >= {"run_id", "p95", "p99", "retry_free_count", "publish_reorder_count"}
+
+    float_fields = ["p95", "p99"]
+    int_fields = ["retry_free_count", "publish_reorder_count"]
+    for row in rows:
+        for f_field in float_fields:
+            assert pytest.approx(float(row[f_field])) == next(rs for rs in sample_run_summary_list if rs.run_id == row["run_id"]).__dict__[f_field]
+        for i_field in int_fields:
+            assert int(row[i_field]) >= 0
+
+
+def test_export_summary_invalid_format(sample_run_summary_list):
+    result = core.export_summary(sample_run_summary_list, format="xml")
+    assert result is None
+
+
+def test_export_summary_single_item(tmp_path):
+    rs = core.RunSummary("run_x", 12.3, 18.7, 50, 3)
+    result_csv = core.export_summary(rs, format="csv")
+    assert result_csv is not None
+    with open(result_csv, newline='', encoding='utf-8') as f:
+        reader = csv.DictReader(f)
+        rows = list(reader)
+    assert len(rows) == 1
+    assert rows[0]["run_id"] == "run_x"