67 lines
No EOL
2.3 KiB
Python
67 lines
No EOL
2.3 KiB
Python
import json
|
|
import pytest
|
|
from pathlib import Path
|
|
import pandas as pd
|
|
|
|
import src.artifact_1.main as main
|
|
|
|
|
|
def load_test_data():
|
|
test_data_path = Path(__file__).parent / 'data' / 'test_timing_data.json'
|
|
if not test_data_path.exists():
|
|
# fallback synthetic data
|
|
return [
|
|
{"expires_at_dist_hours": 48.0, "delta_t": 1.2, "pinned": True, "unpinned": False},
|
|
{"expires_at_dist_hours": 12.0, "delta_t": -0.5, "pinned": False, "unpinned": True},
|
|
{"expires_at_dist_hours": 5.0, "delta_t": 0.3, "pinned": False, "unpinned": True}
|
|
]
|
|
with open(test_data_path, encoding='utf-8') as f:
|
|
return json.load(f)
|
|
|
|
|
|
@pytest.fixture(scope='module')
|
|
def sample_data():
|
|
return load_test_data()
|
|
|
|
|
|
def test_analyze_timing_data_basic(sample_data):
|
|
result = main.analyze_timing_data(sample_data)
|
|
assert isinstance(result, dict)
|
|
assert 'negative_dt_count' in result
|
|
assert 'total_count' in result
|
|
assert result['total_count'] == len(sample_data)
|
|
assert result['negative_dt_count'] <= result['total_count']
|
|
|
|
|
|
def test_analyze_timing_data_near_expiry_threshold(sample_data):
|
|
result = main.analyze_timing_data(sample_data)
|
|
threshold = result.get('recommended_near_expiry_threshold')
|
|
assert isinstance(threshold, (int, float))
|
|
assert 0 <= threshold <= 72
|
|
|
|
|
|
def test_analyze_timing_data_invalid_input():
|
|
# wrong type
|
|
with pytest.raises((ValueError, TypeError)):
|
|
main.analyze_timing_data('not a list')
|
|
# missing keys
|
|
bad_data = [{"delta_t": 1.0}]
|
|
with pytest.raises((KeyError, ValueError)):
|
|
main.analyze_timing_data(bad_data)
|
|
|
|
|
|
def test_analyze_timing_data_pandas_consistency(sample_data):
|
|
# Ensure that conversion to pandas does not break the structure assumptions
|
|
df = pd.DataFrame(sample_data)
|
|
assert all(col in df.columns for col in ['expires_at_dist_hours', 'delta_t'])
|
|
result = main.analyze_timing_data(sample_data)
|
|
assert isinstance(result.get('mean_delta_t'), (float, int))
|
|
|
|
|
|
def test_analyze_timing_data_negative_dt_flag(sample_data):
|
|
has_negative = any(d['delta_t'] < 0 for d in sample_data)
|
|
result = main.analyze_timing_data(sample_data)
|
|
if has_negative:
|
|
assert result['negative_dt_count'] > 0
|
|
else:
|
|
assert result['negative_dt_count'] == 0 |