Spaces:
Running
Running
File size: 2,857 Bytes
c20d7c0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 | """Tests for source runner resilience."""
import time
from unittest.mock import MagicMock, patch
import pytest
from app.services.source_runner import (
AllSourcesFailed,
SourceDataInsufficient,
SourceResult,
SourceUnavailable,
_backoff_seconds,
_count_rows,
_is_transient_error,
run_sources,
)
class TestCountRows:
def test_list(self):
assert _count_rows([1, 2, 3]) == 3
def test_dict_with_daily_key(self):
assert _count_rows({"daily": [1, 2], "other": "value"}) == 2
def test_dict_no_list_keys(self):
assert _count_rows({"key": "value"}) is None
def test_none(self):
assert _count_rows(None) is None
class TestIsTransientError:
def test_timeout_is_transient(self):
assert _is_transient_error(TimeoutError("timeout")) is True
def test_source_unavailable_is_transient(self):
assert _is_transient_error(SourceUnavailable("unavailable")) is True
def test_source_data_insufficient_not_transient(self):
assert _is_transient_error(SourceDataInsufficient("insufficient")) is False
def test_generic_exception_not_transient(self):
assert _is_transient_error(ValueError("bad")) is False
class TestBackoffSeconds:
def test_exponential_growth(self):
b1 = _backoff_seconds(1.0, 0.0, 10.0, 1)
b2 = _backoff_seconds(1.0, 0.0, 10.0, 2)
b3 = _backoff_seconds(1.0, 0.0, 10.0, 3)
assert b1 < b2 < b3
def test_capped_at_max(self):
delay = _backoff_seconds(1.0, 0.0, 5.0, 10)
assert delay <= 5.0
class TestRunSources:
def test_success_returns_data(self):
sources = [("test", lambda: {"daily": [1, 2, 3]})]
result, attempts = run_sources("test", sources, timeout_seconds=5, retry_attempts=1)
assert result.source == "test"
assert len(attempts) == 1
assert attempts[0]["ok"] is True
def test_min_rows_rejects_insufficient(self):
sources = [("test", lambda: {"daily": [1]})]
with pytest.raises(AllSourcesFailed):
run_sources("test", sources, timeout_seconds=5, retry_attempts=1, min_rows=5)
def test_fallback_on_failure(self):
def good():
return {"daily": [1, 2, 3]}
def bad():
raise ValueError("fail")
sources = [("bad", bad), ("good", good)]
result, attempts = run_sources("test", sources, timeout_seconds=5, retry_attempts=1)
assert result.source == "good"
assert len(attempts) == 2
assert attempts[0]["ok"] is False
assert attempts[1]["ok"] is True
def test_all_fail_raises(self):
def fail():
raise ValueError("fail")
sources = [("fail", fail)]
with pytest.raises(AllSourcesFailed):
run_sources("test", sources, timeout_seconds=5, retry_attempts=1)
|