Add trace_aggpy/tests/test_main.py

trace_aggpy/tests/test_main.py

@@ -0,0 +1,55 @@
+import json
+import pytest
+from trace_aggpy import main
+
+@pytest.fixture
+def sample_read_data():
+    return [
+        {"corr_id": "c1", "timestamp": 1.0, "cpu": 0, "field_tag": "mult", "retry_free": True},
+        {"corr_id": "c1", "timestamp": 1.5, "cpu": 1, "field_tag": "shift", "retry_free": False},
+        {"corr_id": "c1", "timestamp": 2.0, "cpu": 0, "field_tag": "mult", "retry_free": True},
+        {"corr_id": "c1", "timestamp": 3.5, "cpu": 1, "field_tag": "nsec_base", "retry_free": True}
+    ]
+
+@pytest.fixture
+def retry_free_only_data():
+    return [
+        {"corr_id": "c2", "timestamp": 0.5, "cpu": 0, "field_tag": "mult", "retry_free": True},
+        {"corr_id": "c2", "timestamp": 1.5, "cpu": 0, "field_tag": "shift", "retry_free": True}
+    ]
+
+def test_analyze_reads_basic(sample_read_data):
+    result = main.analyze_reads("c1", sample_read_data)
+    assert isinstance(result, main.ReadAnalysisResult)
+    as_json = result.to_json()
+    assert as_json["corr_id"] == "c1"
+    assert isinstance(as_json["retry_free_count"], int)
+    assert isinstance(as_json["duration_metrics"], dict)
+    assert isinstance(as_json["summary_table"], list)
+
+def test_retry_free_count(retry_free_only_data):
+    result = main.analyze_reads("c2", retry_free_only_data)
+    j = result.to_json()
+    assert j["retry_free_count"] == 2
+    assert len(j["summary_table"]) >= 1
+
+def test_invalid_input_types():
+    with pytest.raises(AssertionError):
+        main.analyze_reads(123, [])
+    with pytest.raises(AssertionError):
+        main.analyze_reads("cid", "notalist")
+
+@pytest.mark.parametrize("events", [[], [{"corr_id": "x", "timestamp": 1.0, "cpu": 0, "field_tag": "f", "retry_free": False}]])
+def test_empty_and_single_event(events):
+    result = main.analyze_reads("x", events)
+    data = result.to_json()
+    assert "corr_id" in data
+    assert isinstance(data["duration_metrics"], dict)
+    assert isinstance(data["summary_table"], list)
+
+def test_to_json_structure(sample_read_data):
+    result = main.analyze_reads("c1", sample_read_data)
+    json_data = result.to_json()
+    serialized = json.dumps(json_data)
+    parsed = json.loads(serialized)
+    assert parsed["corr_id"] == json_data["corr_id"]