Add clocksource_switch_analysis/tests/test_core.py

clocksource_switch_analysis/tests/test_core.py

@@ -0,0 +1,38 @@
+import pytest
+from clocksource_switch_analysis import core
+
+@pytest.fixture
+def sample_events():
+    return [
+        {"timestamp": 0.1, "cpu_id": 0, "seqcount_retry_count": 0, "clocksource": "tsc"},
+        {"timestamp": 0.2, "cpu_id": 1, "seqcount_retry_count": 1, "clocksource": "tsc"},
+        {"timestamp": 0.15, "cpu_id": 0, "seqcount_retry_count": 0, "clocksource": "tsc"},
+        {"timestamp": 0.3, "cpu_id": 2, "seqcount_retry_count": 3, "clocksource": "tsc"}
+    ]
+
+
+def test_aggregate_clocksource_data_basic(sample_events):
+    result = core.aggregate_clocksource_data(sample_events)
+    assert isinstance(result, dict)
+    expected_keys = {"mean", "p50", "p95", "p99", "retry_free_rate"}
+    assert expected_keys == set(result.keys())
+    # Mittelwert zwischen min 0.1 und max 0.3 sollte zwischen liegen
+    assert 0.1 <= result["mean"] <= 0.3
+    # retry_free_rate: 2 von 4 Events haben keine Retries
+    assert pytest.approx(result["retry_free_rate"], 0.5)
+
+
+def test_empty_input_returns_zero_metrics():
+    result = core.aggregate_clocksource_data([])
+    assert all(result[k] == 0 for k in result), "Alle Ergebnisse sollten 0 bei leerer Eingabe sein"
+
+
+def test_invalid_data_raises():
+    with pytest.raises((TypeError, KeyError, ValueError)):
+        core.aggregate_clocksource_data([{"timestamp": "invalid"}])
+
+
+def test_zero_retry_events(sample_events):
+    all_zero = [{**ev, "seqcount_retry_count": 0} for ev in sample_events]
+    result = core.aggregate_clocksource_data(all_zero)
+    assert result["retry_free_rate"] == 1.0