diff --git a/artifact.timing_analysis/tests/test_core.py b/artifact.timing_analysis/tests/test_core.py new file mode 100644 index 0000000..fc61120 --- /dev/null +++ b/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'] == [] \ No newline at end of file