import asyncio from datetime import datetime, timedelta, timezone import math import re from types import SimpleNamespace import httpx import pytest from app.config import MODEL_VERSION, Settings from app.core.calibration import calibrate_probability from app.core.history import performance_metrics, settle_history from app.core.market import market_consensus from app.core.names import resolve_identity from app.core.radar import build_radar from app.models import FinishedMatch, TeamIdentity from app.pipeline import DailyPipeline from app.providers.football_data import FootballDataProvider from app.providers.http_client import ProviderError, ResilientHTTP from app.providers.odds_api import OddsAPIProvider, _odds_api_timestamp from app.storage import StateStore def test_state_store_rejects_non_finite_json(tmp_path): store = StateStore(tmp_path) with pytest.raises(ValueError): store.save_state({"probability": math.nan}) assert not store.state_path.exists() assert list(tmp_path.iterdir()) == [] def test_state_store_recovers_from_non_object_json(tmp_path): store = StateStore(tmp_path) store.state_path.write_text("[]", encoding="utf-8") state = store.load_state() assert state["status"] == "error" assert state["picks"] == [] def test_performance_ignores_invalid_probabilities(): history = [ {"result": "win", "probability": 0.70, "profit_units": 0.5}, {"result": "loss", "probability": float("nan"), "profit_units": -1.0}, {"result": "win", "probability": 1.5, "profit_units": 0.5}, ] metrics = performance_metrics(history) assert metrics["settled"] == 1 assert metrics["wins"] == 1 def test_performance_treats_invalid_profit_as_zero(): history = [ {"result": "win", "probability": 0.70, "profit_units": "oops"}, {"result": "loss", "probability": 0.65, "profit_units": float("inf")}, ] metrics = performance_metrics(history) assert metrics["settled"] == 2 assert metrics["profit_units"] == 0.0 assert math.isfinite(metrics["max_drawdown_units"]) def test_calibration_ignores_non_finite_probability_history(): history = [{ "result": "win", "probability": float("nan"), "model_version": MODEL_VERSION, "competition_code": "PL", }] * 30 probability, metadata = calibrate_probability( 0.70, history, model_version=MODEL_VERSION, competition_code="PL", ) assert probability == 0.70 assert metadata["effective_samples"] == 0.0 def test_provider_error_detail_redacts_credentials(): secret = "private-api-key-value" response = httpx.Response(401, text=f"invalid api key: {secret}") detail = ResilientHTTP._safe_error_detail( response, {"apiKey": secret, "regions": "eu"}, {"X-Auth-Token": "another-private-token"}, ) assert secret not in detail assert "[redacted]" in detail def test_market_consensus_ignores_malformed_nested_rows(): event = { "home_team": "Alpha", "away_team": "Beta", "bookmakers": [None, "bad", {"markets": [None, {"key": "h2h", "outcomes": [None]}]}], } market = market_consensus(event) assert market.bookmakers == 0 @pytest.mark.parametrize( ("expanded", "short"), [ ("PSV Eindhoven", "PSV"), ("AZ Alkmaar", "AZ"), ("NEC Nijmegen", "NEC"), ], ) def test_known_dutch_acronyms_resolve_exactly_without_short_fuzzy(expanded, short): identity = TeamIdentity("id:1", short, (short,)) resolved, score, _margin = resolve_identity(expanded, [identity], minimum=82) assert resolved == identity assert score == 100.0 unknown, unknown_score, _ = resolve_identity("ABC United", [identity], minimum=82) assert unknown is None assert unknown_score == 0.0 def test_radar_ranks_readings_but_excludes_approved_or_invalid_events(): rows = [ { "approved": False, "event_id": "low", "selection": "Alpha", "probability": 0.55, "conservative_probability": 0.52, "odd": 1.8, "safe_score": 61, "model_ev": -0.01, "reason": "probabilidade abaixo do filtro; retorno esperado negativo", }, { "approved": False, "event_id": "high", "selection": "Beta", "probability": 0.63, "conservative_probability": 0.60, "odd": 1.6, "safe_score": 70, "model_ev": 0.01, "blockers": ["SafeScore abaixo do mínimo"], }, { "approved": False, "event_id": "approved-event", "selection": "Gamma", "probability": 0.80, "odd": 1.3, }, {"approved": False, "event_id": "no-model", "selection": "", "probability": 0.9}, ] radar = build_radar(rows, approved_event_ids={"approved-event"}, limit=5) assert [row["event_id"] for row in radar] == ["high", "low"] assert radar[0]["approved"] is False assert radar[0]["label"] == "EM OBSERVAÇÃO" assert radar[1]["blockers"] == [ "probabilidade abaixo do filtro", "retorno esperado negativo", ] def test_odds_provider_does_not_report_total_failure_as_empty_success(): class FailingHTTP: async def get_json(self, *args, **kwargs): raise ProviderError("provider unavailable", 503) provider = OddsAPIProvider("test-key", FailingHTTP()) with pytest.raises(ProviderError, match="Todas as consultas"): asyncio.run(provider.fetch_events(("soccer_epl",), 24)) assert provider.queried_keys == ["soccer_epl"] assert len(provider.errors) == 1 def test_odds_provider_uses_second_precision_utc_timestamps(): class CapturingHTTP: odds_params = None async def get_json(self, url, **kwargs): if url.endswith("/sports/"): return ([{"key": "soccer_epl", "active": True}], httpx.Headers()) self.odds_params = kwargs["params"] return ([], httpx.Headers()) http = CapturingHTTP() provider = OddsAPIProvider("test-key", http) asyncio.run(provider.fetch_events(("soccer_epl",), 24)) exact_utc = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$") time_from = http.odds_params["commenceTimeFrom"] time_to = http.odds_params["commenceTimeTo"] assert exact_utc.fullmatch(time_from) assert exact_utc.fullmatch(time_to) parsed_from = datetime.strptime(time_from, "%Y-%m-%dT%H:%M:%SZ") parsed_to = datetime.strptime(time_to, "%Y-%m-%dT%H:%M:%SZ") assert parsed_to - parsed_from == timedelta(hours=24) def test_odds_timestamp_normalizes_timezone_and_rollover(): local = datetime( 2026, 12, 31, 23, 59, 59, 999999, tzinfo=timezone(timedelta(hours=3)), ) assert _odds_api_timestamp(local) == "2026-12-31T20:59:59Z" assert _odds_api_timestamp(local + timedelta(hours=6)) == "2027-01-01T02:59:59Z" def test_football_provider_treats_404_with_usable_previous_season_as_fallback(monkeypatch): provider = FootballDataProvider("test-token", http=None) calls = [] previous_matches = [ FinishedMatch( match_id=f"previous-{index}", competition="PL", utc_date=datetime.now(timezone.utc) - timedelta(days=index + 1), home="Alpha FC", away="Beta FC", home_goals=2, away_goals=0, ) for index in range(40) ] async def fake_fetch(competition_code, season): calls.append((competition_code, season)) if len(calls) == 1: return [], f"{competition_code}/{season}: HTTP 404", 404 return previous_matches, None, None monkeypatch.setattr(provider, "_fetch_season", fake_fetch) matches, meta = asyncio.run( provider.fetch_finished(240, ("soccer_epl",), cached_matches=[]) ) assert matches == list(reversed(previous_matches)) assert meta["errors"] == [] assert len(meta["fallbacks"]) == 1 assert meta["fallbacks"][0]["requested_season"] == calls[0][1] assert meta["fallbacks"][0]["fallback_season"] == calls[1][1] assert meta["fallbacks"][0]["matches"] == 40 assert meta["competitions"]["PL"]["previous_loaded"] is True @pytest.mark.parametrize( ("current_status", "previous_count"), [(404, 39), (500, 40)], ) def test_football_provider_does_not_mask_failed_or_insufficient_fallback( monkeypatch, current_status, previous_count, ): provider = FootballDataProvider("test-token", http=None) calls = 0 async def fake_fetch(competition_code, season): nonlocal calls calls += 1 if calls == 1: return [], f"{competition_code}/{season}: HTTP {current_status}", current_status matches = [ FinishedMatch( match_id=f"fallback-{index}", competition="PL", utc_date=datetime.now(timezone.utc) - timedelta(days=index + 1), home="Alpha FC", away="Beta FC", home_goals=1, away_goals=0, ) for index in range(previous_count) ] return matches, None, None monkeypatch.setattr(provider, "_fetch_season", fake_fetch) with pytest.raises(ProviderError, match="Todas as consultas utilizáveis"): asyncio.run(provider.fetch_finished(240, ("soccer_epl",), cached_matches=[])) @pytest.mark.parametrize( ("status_code", "message"), [ (401, "unauthorized"), (429, "rate limit exceeded"), (400, "Your API token is invalid"), ], ) def test_football_provider_fails_fast_for_auth_or_rate_limit( monkeypatch, status_code, message, ): provider = FootballDataProvider("test-token", http=None) async def failing_get(_url, _params): raise ProviderError(message, status_code) monkeypatch.setattr(provider, "_get", failing_get) with pytest.raises(ProviderError, match=message): asyncio.run(provider._fetch_season("PL", 2026)) def test_pipeline_running_includes_direct_lock_holder(): pipeline = DailyPipeline(settings=None, store=None) async def check(): await pipeline._lock.acquire() try: assert pipeline.running is True assert pipeline.trigger_background() is False finally: pipeline._lock.release() asyncio.run(check()) def test_settings_fail_closed_on_invalid_regions_or_leagues(monkeypatch): monkeypatch.setenv("ODDS_REGIONS", "eu,invalid") monkeypatch.setenv("ODDS_SPORT_KEYS", "soccer_epl,soccer_typo") configured = Settings() assert configured.odds_regions == "" assert configured.sport_keys == () assert configured.required_ready is False def test_unicode_cron_secret_is_compared_without_server_error(monkeypatch): import app.main as main_module monkeypatch.setattr(main_module, "settings", SimpleNamespace(cron_secret="segredo-ç")) assert main_module._authorized("segredo-ç") is True assert main_module._authorized("segredo-c") is False def test_lifespan_triggers_background_scan_when_state_is_stale(monkeypatch): import app.main as main_module calls = {"triggered": 0, "shutdown": 0} class FakeStore: def restore_from_hub_if_needed(self): return None def load_state(self): return { "status": "ok", "generated_at": "2026-08-12T10:00:00+00:00", } def save_state(self, _state): return None class FakePipeline: running = False def recent_success(self, _minutes): return False def trigger_background(self): calls["triggered"] += 1 return True async def shutdown(self): calls["shutdown"] += 1 monkeypatch.setattr(main_module, "settings", SimpleNamespace(required_ready=True, min_scan_interval_minutes=180)) monkeypatch.setattr(main_module, "store", FakeStore()) monkeypatch.setattr(main_module, "pipeline", FakePipeline()) async def scenario(): async with main_module.lifespan(main_module.app): pass asyncio.run(scenario()) assert calls["triggered"] == 1 assert calls["shutdown"] == 1 def test_health_reports_state_age_and_staleness(monkeypatch): import app.main as main_module monkeypatch.setattr(main_module, "settings", SimpleNamespace(min_scan_interval_minutes=180, football_data_token="", odds_api_key="", cron_secret="", hf_token="", hf_dataset_repo="")) monkeypatch.setattr(main_module.store, "load_state", lambda: { "status": "ok", "generated_at": "2026-08-14T10:00:00+00:00", }) monkeypatch.setattr(main_module.pipeline, "running", False) health = asyncio.run(main_module.health()) assert health["state_stale"] is True assert health["state_age_minutes"] >= 180 def test_state_endpoint_hides_expired_picks(monkeypatch): import app.main as main_module raw_state = { "status": "ok", "generated_at": "2026-08-15T12:00:00+00:00", "summary": { "events": 2, "historical_matches": 50, "approved": 2, "rejected": 0, "radar": 0, }, "picks": [ { "event_id": "expired", "kickoff": "2026-08-15T09:00:00+00:00", "odd": 1.7, "probability": 0.7, "conservative_probability": 0.6, "safe_score": 81, "competition_code": "PL", "home": "A", "away": "B", "selection": "A", }, { "event_id": "live", "kickoff": "2026-08-15T18:00:00+00:00", "odd": 1.8, "probability": 0.72, "conservative_probability": 0.62, "safe_score": 83, "competition_code": "PL", "home": "C", "away": "D", "selection": "C", }, ], "tickets": {"safe": None, "balanced": None, "freebet": None}, "warnings": [], } monkeypatch.setattr(main_module.store, "load_state", lambda: raw_state) state = asyncio.run(main_module.state()) assert [pick["event_id"] for pick in state["picks"]] == ["live"] assert state["summary"]["approved"] == 1 def test_lifespan_prunes_expired_picks_and_refreshes_on_startup(monkeypatch): import app.main as main_module saved_states = [] triggered = {"count": 0} class FakeStore: def restore_from_hub_if_needed(self): return None def load_state(self): return { "status": "ok", "generated_at": "2026-08-15T12:00:00+00:00", "summary": { "events": 2, "historical_matches": 50, "approved": 2, "rejected": 0, "radar": 0, }, "picks": [ { "event_id": "expired", "kickoff": "2026-08-15T09:00:00+00:00", "odd": 1.7, "probability": 0.7, "conservative_probability": 0.6, "safe_score": 81, "competition_code": "PL", "home": "A", "away": "B", "selection": "A", }, ], "warnings": [], } def save_state(self, state): saved_states.append(state) class FakePipeline: running = False def recent_success(self, _minutes): return True def trigger_background(self): triggered["count"] += 1 return True async def shutdown(self): return None monkeypatch.setattr(main_module, "settings", SimpleNamespace(required_ready=True, min_scan_interval_minutes=180)) monkeypatch.setattr(main_module, "store", FakeStore()) monkeypatch.setattr(main_module, "pipeline", FakePipeline()) async def scenario(): async with main_module.lifespan(main_module.app): pass asyncio.run(scenario()) assert triggered["count"] == 1 assert saved_states assert saved_states[0]["picks"] == [] assert saved_states[0]["summary"]["approved"] == 0 def test_settlement_prefers_resolved_ids_and_records_audit_fields(): kickoff = datetime.now(timezone.utc) - timedelta(hours=4) history = [{ "event_id": "event-1", "kickoff": kickoff.isoformat(), "competition_code": "PL", "home": "Nome divergente", "away": "Outro nome", "resolved_home_key": "id:10", "resolved_away_key": "id:20", "side": "home", "odd": 1.8, "probability": 0.7, "result": None, }] matches = [FinishedMatch( match_id="match-99", competition="PL", utc_date=kickoff, home="Canonical Home", away="Canonical Away", home_goals=2, away_goals=1, home_id="10", away_id="20", )] settle_history(history, matches) assert history[0]["result"] == "win" assert history[0]["settled_match_id"] == "match-99" def test_settlement_never_runs_before_match_can_finish(): kickoff = datetime.now(timezone.utc) - timedelta(minutes=15) history = [{ "event_id": "event-live", "kickoff": kickoff.isoformat(), "competition_code": "PL", "home": "Alpha", "away": "Beta", "side": "home", "odd": 1.5, "probability": 0.7, "result": None, }] matches = [FinishedMatch("m", "PL", kickoff, "Alpha", "Beta", 1, 0)] settle_history(history, matches) assert history[0]["result"] is None def test_api_security_headers_allow_hugging_face_embedding(): from fastapi.testclient import TestClient from app.main import app with TestClient(app) as client: page = client.get("/") api = client.get("/api/health") csp = page.headers["content-security-policy"] assert "frame-ancestors https://huggingface.co https://*.huggingface.co" in csp assert "x-frame-options" not in page.headers assert api.headers["cache-control"] == "no-store" assert api.json()["version"] == MODEL_VERSION def test_admin_scan_blocks_recent_duplicate_by_default(monkeypatch): from fastapi.testclient import TestClient import app.main as main_module monkeypatch.setattr(main_module, "_authorized", lambda _secret: True) monkeypatch.setattr(main_module.pipeline, "recent_success", lambda _minutes: True) with TestClient(main_module.app) as client: response = client.post( "/api/admin/scan?wait=1", headers={"X-Cron-Secret": "test"}, ) assert response.status_code == 200 assert response.json()["accepted"] is False assert "scan recente" in response.json()["message"]