From f08301a40b9ca1ff025398e872a472d06e79612e Mon Sep 17 00:00:00 2001 From: Mika Date: Thu, 19 Mar 2026 13:57:22 +0000 Subject: [PATCH] Add artifact_1_bandwidth_analysis/tests/test_core.py --- .../tests/test_core.py | 63 +++++++++++++++++++ 1 file changed, 63 insertions(+) create mode 100644 artifact_1_bandwidth_analysis/tests/test_core.py diff --git a/artifact_1_bandwidth_analysis/tests/test_core.py b/artifact_1_bandwidth_analysis/tests/test_core.py new file mode 100644 index 0000000..9114e6d --- /dev/null +++ b/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}])