Add measure_latency/tests/test_core.py

measure_latency/tests/test_core.py

@@ -0,0 +1,62 @@
+import pytest
+import statistics
+from datetime import datetime, timedelta
+
+from measure_latency import core
+
+@pytest.fixture
+def sample_latency_results():
+    base_time = datetime.now()
+    results = []
+    for i in range(5):
+        upload_end = base_time + timedelta(milliseconds=i * 10)
+        api_response = upload_end + timedelta(milliseconds=5)
+        fs_mtime = upload_end + timedelta(milliseconds=10)
+        result = core.LatencyResult(
+            upload_end_time=upload_end,
+            api_response_time=api_response,
+            fs_mtime=fs_mtime,
+            offsets={}
+        )
+        result.offsets = result.compute_offsets()
+        results.append(result)
+    return results
+
+def test_latencyresult_compute_offsets():
+    t0 = datetime.now()
+    lr = core.LatencyResult(
+        upload_end_time=t0,
+        api_response_time=t0 + timedelta(milliseconds=50),
+        fs_mtime=t0 + timedelta(milliseconds=100),
+        offsets={}
+    )
+    offsets = lr.compute_offsets()
+    assert isinstance(offsets, dict)
+    assert pytest.approx(offsets["api_to_upload"], rel=1e-3) == 0.05
+    assert pytest.approx(offsets["fs_to_upload"], rel=1e-3) == 0.1
+    assert pytest.approx(offsets["fs_to_api"], rel=1e-3) == 0.05
+
+def test_analyze_results_statistics(sample_latency_results):
+    summary = core.analyze_results(sample_latency_results)
+    assert all(key in summary for key in ["p50", "p95", "max", "variance"])
+    assert isinstance(summary["p50"], float)
+    assert summary["max"] >= summary["p95"]
+
+def test_measure_latencies_runs(monkeypatch):
+    call_count = {"count": 0}
+
+    def fake_sleep(_):
+        call_count["count"] += 1
+
+    monkeypatch.setattr(core.time, "sleep", fake_sleep)
+    results = core.measure_latencies(3)
+    assert len(results) == 3
+    for r in results:
+        assert isinstance(r, core.LatencyResult)
+        assert "api_to_upload" in r.offsets
+    assert call_count["count"] == 3
+
+def test_invalid_input_empty_analysis():
+    with pytest.raises(AssertionError):
+        # assuming analyze_results asserts non-empty input under ci_ready
+        assert core.analyze_results([])  # should fail