Add artifact_1_bandwidth_analysis/tests/test_core.py

artifact_1_bandwidth_analysis/tests/test_core.py

@@ -0,0 +1,63 @@
+import pytest
+import math
+from statistics import median, quantiles
+
+from src.artifact_1_bandwidth_analysis import core
+
+
+@pytest.fixture
+def synthetic_data_enforced():
+    # artificial distribution centered around 10.0
+    return [
+        {"worker_id": f"w{i}", "timestamp": t}
+        for i, t in enumerate([9.9, 10.0, 10.1, 9.8, 10.2])
+    ]
+
+
+@pytest.fixture
+def synthetic_data_randomized():
+    # broader, more scattered timestamps centered around 5.0
+    return [
+        {"worker_id": f"r{i}", "timestamp": t}
+        for i, t in enumerate([4.0, 5.0, 6.0, 3.0, 7.0])
+    ]
+
+
+def test_analyze_bandwidth_basic(synthetic_data_enforced, synthetic_data_randomized):
+    result = core.analyze_bandwidth(synthetic_data_enforced, synthetic_data_randomized)
+
+    # structure validation
+    assert isinstance(result, dict)
+    assert set(result.keys()) == {"affinity_enforced", "affinity_off"}
+
+    for mode_key in result:
+        mode_val = result[mode_key]
+        assert isinstance(mode_val, dict)
+        assert set(mode_val.keys()) == {"band_center", "band_width"}
+        assert isinstance(mode_val["band_center"], (int, float))
+        assert isinstance(mode_val["band_width"], (int, float))
+
+
+def test_analyze_bandwidth_correctness(synthetic_data_enforced, synthetic_data_randomized):
+    result = core.analyze_bandwidth(synthetic_data_enforced, synthetic_data_randomized)
+
+    enforced_center_expected = median([9.9, 10.0, 10.1, 9.8, 10.2])
+    enforced_q1, enforced_q3 = quantiles([9.9, 10.0, 10.1, 9.8, 10.2], n=4)[0], quantiles([9.9, 10.0, 10.1, 9.8, 10.2], n=4)[2]
+    enforced_width_expected = enforced_q3 - enforced_q1
+
+    off_center_expected = median([4.0, 5.0, 6.0, 3.0, 7.0])
+    off_q1, off_q3 = quantiles([4.0, 5.0, 6.0, 3.0, 7.0], n=4)[0], quantiles([4.0, 5.0, 6.0, 3.0, 7.0], n=4)[2]
+    off_width_expected = off_q3 - off_q1
+
+    assert math.isclose(result["affinity_enforced"]["band_center"], enforced_center_expected, rel_tol=1e-9)
+    assert math.isclose(result["affinity_enforced"]["band_width"], enforced_width_expected, rel_tol=1e-9)
+    assert math.isclose(result["affinity_off"]["band_center"], off_center_expected, rel_tol=1e-9)
+    assert math.isclose(result["affinity_off"]["band_width"], off_width_expected, rel_tol=1e-9)
+
+
+def test_analyze_bandwidth_invalid_input():
+    with pytest.raises((ValueError, TypeError)):
+        core.analyze_bandwidth([], None)
+
+    with pytest.raises((ValueError, TypeError)):
+        core.analyze_bandwidth([{"timestamp": "invalid"}], [{"timestamp": 1.0}])