Add artifact.timing_analysis/tests/test_core.py

artifact.timing_analysis/tests/test_core.py

@@ -0,0 +1,65 @@
+import pytest
+import pandas as pd
+from datetime import datetime, timedelta
+from artifact_timing_analysis import core
+
+@pytest.fixture
+def sample_timing_data():
+    base_time = datetime.utcnow()
+    return [
+        {
+            'timestamp': (base_time + timedelta(seconds=i)).isoformat(),
+            't_gate_read': float(i),
+            't_index_visible': float(i) + (0.1 * i),
+            'offset': 0.1 * i
+        }
+        for i in range(1, 6)
+    ]
+
+@pytest.fixture
+def anomaly_data():
+    base_time = datetime.utcnow()
+    # introduce an anomaly with a large offset
+    records = [
+        {
+            'timestamp': (base_time + timedelta(seconds=i)).isoformat(),
+            't_gate_read': float(i),
+            't_index_visible': float(i) + (0.1 if i < 5 else 10.0),
+            'offset': 0.1 if i < 5 else 10.0
+        }
+        for i in range(1, 7)
+    ]
+    return records
+
+def test_analyze_timing_offsets_basic(sample_timing_data):
+    result = core.analyze_timing_offsets(sample_timing_data)
+    assert isinstance(result, dict)
+    assert 'mean_offset' in result
+    assert 'p95_offset' in result
+    assert 'anomalies' in result
+    # mean of 0.1*1..5 = 0.3
+    assert pytest.approx(result['mean_offset'], rel=1e-6) == 0.3
+    assert result['p95_offset'] >= result['mean_offset']
+
+def test_analyze_timing_offsets_detects_anomaly(anomaly_data):
+    result = core.analyze_timing_offsets(anomaly_data)
+    anomalies = result.get('anomalies', [])
+    assert any(abs(a['offset']) > 1.0 for a in anomalies)
+    assert result['p95_offset'] >= result['mean_offset']
+
+def test_report_timing_anomalies_output(monkeypatch, anomaly_data):
+    computed = {'anomalies': [{'timestamp': anomaly_data[-1]['timestamp'], 'offset': anomaly_data[-1]['offset']}], 'mean_offset': 0.0, 'p95_offset': 0.0}
+    # Monkeypatch analyze_timing_offsets to return this computed result
+    monkeypatch.setattr(core, 'analyze_timing_offsets', lambda _: computed)
+    report = core.report_timing_anomalies()
+    assert isinstance(report, str)
+    assert 'timestamp' in report
+    assert 'offset' in report
+    assert 'Anomalie' in report or 'Anomaly' in report
+
+def test_empty_input_edge_case():
+    result = core.analyze_timing_offsets([])
+    assert isinstance(result, dict)
+    assert result['mean_offset'] == 0.0
+    assert result['p95_offset'] == 0.0
+    assert result['anomalies'] == []