32 lines
1.3 KiB
Python
32 lines
1.3 KiB
Python
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from app.community_scheduler import MAX_BACKOFF_SECONDS, configured_sources, _static_registry, oldest_site_source, retry_delay
|
|
from app.config import Settings
|
|
|
|
|
|
def test_all_authorized_sources_are_scheduled() -> None:
|
|
assert set(_static_registry()) == {"rf4db", "rf4stat-fishing", "rf4stat-post", "rf4map", "rf4posts-spot"}
|
|
|
|
|
|
def test_community_interval_cannot_be_less_than_30_minutes() -> None:
|
|
with pytest.raises(ValidationError):
|
|
Settings(community_import_interval_seconds=1799)
|
|
|
|
|
|
def test_failed_runs_back_off_but_success_resets_delay() -> None:
|
|
assert retry_delay(["failed"]) == 1800
|
|
assert retry_delay(["failed", "failed", "failed"]) == 7200
|
|
assert retry_delay(["failed"] * 20) == MAX_BACKOFF_SECONDS
|
|
assert retry_delay(["success", "failed"]) == 1800
|
|
|
|
|
|
def test_same_site_endpoints_rotate_by_oldest_attempt() -> None:
|
|
now = datetime.now(timezone.utc)
|
|
all_keys = {"rf4db", "rf4stat-fishing", "rf4stat-post", "rf4map", "rf4posts-spot"}
|
|
assert oldest_site_source("rf4stat-fishing", {}, all_keys) == "rf4stat-fishing"
|
|
latest = {"rf4stat-fishing": now, "rf4stat-post": now - timedelta(hours=1)}
|
|
assert oldest_site_source("rf4stat-fishing", latest, all_keys) == "rf4stat-post"
|