tests: backfill coverage for openrouter transport, auth sessions, cadence

Three new test files covering modules the audit flagged as having zero
direct coverage:

- test_openrouter_transport.py (18 tests): provider chain selection,
  endpoint resolution, _call_provider parse path (including the
  reasoning-field fallback and token-based cost estimation), and
  call_llm's cross-provider failover. Uses httpx.MockTransport so no
  network. Patches _call_provider for failover tests to bypass
  tenacity's retry delays.

- test_auth_session.py (7 tests): sign/verify round-trip, tampered
  cookie rejection, expired cookie rejection (via TTL monkeypatch),
  garbage input handling, salt isolation between session and pending
  serializers, and rejection of cookies signed with a different secret.

- test_cadence_policy.py (16 tests): is_active_window weekday/weekend
  + half-open interval boundaries, min_gap_hours across bands,
  should_run gating for first-run / active / off-hours / weekend
  / naive-datetime cases, and the NEWS_POLICY 20-minute / 3-hour
  variations.

Suite goes from 291 to 336 passing.
This commit is contained in:
Giorgio Gilestro 2026-05-28 13:58:28 +02:00
parent 83995e96c8
commit f9f4f25ef7
3 changed files with 500 additions and 0 deletions

View file

@ -0,0 +1,163 @@
"""Cadence policy — the gate that ai_log_job and indicator_summary_job
use to throttle OpenRouter spend outside active market hours.
Pure-function module, so tests just construct timestamps and assert on
the (should_run, reason) tuple. Uses the default policy (active window
07:00-21:00 UTC weekdays, no off-hours runs without 4+ hours since
last success, weekends 12+ hours).
"""
from __future__ import annotations
from datetime import datetime, timedelta, timezone
import pytest
from app.services.cadence import DEFAULT_POLICY, NEWS_POLICY, CadencePolicy
def _utc(year, month, day, hour, minute=0):
return datetime(year, month, day, hour, minute, tzinfo=timezone.utc)
# Pick reference timestamps used across tests. Wednesday 12:00 UTC is
# squarely inside the active window; Wednesday 03:00 is off-hours;
# Saturday 12:00 is weekend.
_WED_NOON = _utc(2026, 5, 27, 12) # Wednesday 12:00
_WED_PRE_DAWN = _utc(2026, 5, 27, 3) # Wednesday 03:00
_SAT_NOON = _utc(2026, 5, 30, 12) # Saturday 12:00
# ---------------------------------------------------------------------------
# is_active_window
# ---------------------------------------------------------------------------
def test_active_window_weekday_noon_is_active():
assert DEFAULT_POLICY.is_active_window(_WED_NOON) is True
def test_active_window_weekday_predawn_is_off_hours():
assert DEFAULT_POLICY.is_active_window(_WED_PRE_DAWN) is False
def test_active_window_weekend_always_off_hours():
"""Weekends bypass the hour check — even Saturday noon is throttled."""
assert DEFAULT_POLICY.is_active_window(_SAT_NOON) is False
def test_active_window_boundary_inclusive_start_exclusive_end():
"""07:00 UTC is the first active hour; 21:00 is the first off-hour.
Locks the half-open interval semantics in place."""
assert DEFAULT_POLICY.is_active_window(_utc(2026, 5, 27, 7)) is True
assert DEFAULT_POLICY.is_active_window(_utc(2026, 5, 27, 21)) is False
# ---------------------------------------------------------------------------
# min_gap_hours
# ---------------------------------------------------------------------------
def test_min_gap_uses_zero_during_active_window():
assert DEFAULT_POLICY.min_gap_hours(_WED_NOON) == 0.0
def test_min_gap_uses_off_hours_value_at_night():
assert DEFAULT_POLICY.min_gap_hours(_WED_PRE_DAWN) == 4.0
def test_min_gap_uses_weekend_value_on_saturday():
assert DEFAULT_POLICY.min_gap_hours(_SAT_NOON) == 12.0
# ---------------------------------------------------------------------------
# should_run — the function jobs call
# ---------------------------------------------------------------------------
def test_should_run_first_ever_call_always_proceeds():
ok, reason = DEFAULT_POLICY.should_run(None, now=_WED_NOON)
assert ok is True
assert "no prior" in reason.lower()
def test_should_run_during_active_window_always_proceeds():
"""Default policy has active_gap_h=0, so even a run from 1 minute ago
is allowed when we're in the active window."""
last = _WED_NOON - timedelta(minutes=1)
ok, reason = DEFAULT_POLICY.should_run(last, now=_WED_NOON)
assert ok is True
assert "active" in reason
def test_should_run_off_hours_too_soon_is_throttled():
"""Off-hours requires 4+ hours since last success. 1 hour ago → no."""
last = _WED_PRE_DAWN - timedelta(hours=1)
ok, reason = DEFAULT_POLICY.should_run(last, now=_WED_PRE_DAWN)
assert ok is False
assert "throttled" in reason
assert "off-hours" in reason
def test_should_run_off_hours_after_gap_proceeds():
last = _WED_PRE_DAWN - timedelta(hours=5)
ok, reason = DEFAULT_POLICY.should_run(last, now=_WED_PRE_DAWN)
assert ok is True
assert "off-hours" in reason
def test_should_run_weekend_requires_12h_gap():
"""Weekend gap is 12h. 6h is too soon; 13h is enough."""
ok6, _ = DEFAULT_POLICY.should_run(
_SAT_NOON - timedelta(hours=6), now=_SAT_NOON,
)
ok13, _ = DEFAULT_POLICY.should_run(
_SAT_NOON - timedelta(hours=13), now=_SAT_NOON,
)
assert ok6 is False
assert ok13 is True
def test_should_run_naive_datetime_treated_as_utc():
"""The DB column comes back as a naive datetime in some test paths;
the policy must coerce it to UTC rather than crash on tz subtraction."""
naive_last = _WED_PRE_DAWN.replace(tzinfo=None) - timedelta(hours=5)
ok, _ = DEFAULT_POLICY.should_run(naive_last, now=_WED_PRE_DAWN)
assert ok is True
# ---------------------------------------------------------------------------
# NEWS_POLICY — tighter gaps so 3 runs/hour during the active window.
# ---------------------------------------------------------------------------
def test_news_policy_active_gap_is_twenty_minutes():
# 20 minutes = 1/3 hour. Verify a 15-min-ago run is throttled but
# a 21-min-ago one is allowed.
last_15 = _WED_NOON - timedelta(minutes=15)
last_21 = _WED_NOON - timedelta(minutes=21)
assert NEWS_POLICY.should_run(last_15, now=_WED_NOON)[0] is False
assert NEWS_POLICY.should_run(last_21, now=_WED_NOON)[0] is True
def test_news_policy_off_hours_gap_is_three_hours():
last_2h = _WED_PRE_DAWN - timedelta(hours=2)
last_4h = _WED_PRE_DAWN - timedelta(hours=4)
assert NEWS_POLICY.should_run(last_2h, now=_WED_PRE_DAWN)[0] is False
assert NEWS_POLICY.should_run(last_4h, now=_WED_PRE_DAWN)[0] is True
# ---------------------------------------------------------------------------
# Bespoke policy — confirms the dataclass is reconfigurable for callers
# (the audit flagged this as risky to over-fit to defaults).
# ---------------------------------------------------------------------------
def test_custom_policy_with_active_gap_throttles_within_window():
"""active_gap_h=0.5 means even during the active window a run from
20 minutes ago is throttled verifies the gate isn't hardcoded to
'always run during active'."""
p = CadencePolicy(active_gap_h=0.5)
last = _WED_NOON - timedelta(minutes=20)
ok, reason = p.should_run(last, now=_WED_NOON)
assert ok is False
assert "throttled" in reason