Spaces:
Running
Running
| from __future__ import annotations | |
| import time | |
| from pathlib import Path | |
| from fastapi.testclient import TestClient | |
| from racing_reports.api import routes_meta, routes_reports | |
| from racing_reports.reports.catalogs import ( | |
| BLOCK_ZSCORE_FIGURES, | |
| PRE_MATCH_FIGURES, | |
| PRE_MATCH_TABLES, | |
| ) | |
| from racing_reports.web import app as web_app | |
| def test_health_endpoint(): | |
| client = TestClient(web_app.app) | |
| resp = client.get("/api/health") | |
| assert resp.status_code == 200 | |
| data = resp.json() | |
| assert data["ok"] is True | |
| assert data["nn_enabled"] is False # MVP default | |
| def test_leagues_and_seasons(): | |
| client = TestClient(web_app.app) | |
| leagues = client.get("/api/leagues").json() | |
| assert "Spanish Segunda Division" in leagues | |
| seasons = client.get("/api/seasons", params={"league": "Spanish Segunda Division"}).json() | |
| assert "25-26" in seasons | |
| teams = client.get( | |
| "/api/teams", params={"league": "Spanish Segunda Division", "season": "25-26"} | |
| ).json() | |
| assert "Racing de Santander" in teams | |
| def test_catalog_pre_match(): | |
| client = TestClient(web_app.app) | |
| resp = client.get("/api/reports/pre_match/catalog") | |
| assert resp.status_code == 200 | |
| data = resp.json() | |
| assert data["report_type"] == "pre_match" | |
| assert [f["name"] for f in data["figures"]] == [s.name for s in PRE_MATCH_FIGURES] | |
| assert [t["name"] for t in data["tables"]] == [s.name for s in PRE_MATCH_TABLES] | |
| # las 2 figuras NN están marcadas | |
| nn_figs = [f for f in data["figures"] if f["requires_nn"]] | |
| assert len(nn_figs) == 2 | |
| def test_catalog_block_zscore(): | |
| client = TestClient(web_app.app) | |
| resp = client.get("/api/reports/block_zscore/catalog") | |
| assert resp.status_code == 200 | |
| data = resp.json() | |
| assert [f["name"] for f in data["figures"]] == [s.name for s in BLOCK_ZSCORE_FIGURES] | |
| assert data["tables"] == [] | |
| def test_catalog_unknown_returns_404(): | |
| client = TestClient(web_app.app) | |
| resp = client.get("/api/reports/unknown/catalog") | |
| assert resp.status_code == 404 | |
| def test_post_match_run_returns_503_in_mvp(): | |
| client = TestClient(web_app.app) | |
| resp = client.post( | |
| "/api/reports/post_match/run", | |
| json={"league": "X", "season": "Y", "team_a": "A", "team_b": "B"}, | |
| ) | |
| assert resp.status_code == 503 | |
| def test_run_requires_teams(): | |
| client = TestClient(web_app.app) | |
| resp = client.post( | |
| "/api/reports/pre_match/run", | |
| json={"league": "X", "season": "Y"}, | |
| ) | |
| assert resp.status_code == 400 | |
| def test_block_zscore_e2e(monkeypatch, sample_store, tmp_path: Path): | |
| """End-to-end: lanza job de bloques contra el sample_store, hace polling, | |
| descarga una figura PNG individual y el ZIP completo.""" | |
| # Reemplazamos el ReportRunner() que instancia routes_reports por uno | |
| # apuntado al sample_store (sin Azure). | |
| from racing_reports.runner import ReportRunner | |
| monkeypatch.setattr( | |
| routes_reports, | |
| "ReportRunner", | |
| lambda: ReportRunner(sample_store, settings=sample_store.settings), | |
| ) | |
| # Las settings del sample_store apuntan a un output_dir bajo tmp_path; la | |
| # whitelist de _safe_output_path usa DEFAULT_SETTINGS.output_dir, que es | |
| # distinto. Para el test parchamos el guard. | |
| from racing_reports.api import routes_jobs | |
| monkeypatch.setattr( | |
| routes_jobs, "_safe_output_path", lambda p: Path(p) | |
| ) | |
| client = TestClient(web_app.app) | |
| resp = client.post( | |
| "/api/reports/block_zscore/run", | |
| json={ | |
| "league": "Spanish Segunda Division", | |
| "season": "25-26", | |
| "team_a": "Racing de Santander", | |
| "team_b": "Huesca", | |
| }, | |
| ) | |
| assert resp.status_code == 202 | |
| job_id = resp.json()["job_id"] | |
| # Polling | |
| for _ in range(100): | |
| st = client.get(f"/api/jobs/{job_id}").json() | |
| if st["status"] in ("done", "error"): | |
| break | |
| time.sleep(0.1) | |
| assert st["status"] == "done", st.get("error") | |
| bundle = st["bundles"]["block_zscore"] | |
| assert len(bundle["figures"]) == len(BLOCK_ZSCORE_FIGURES) | |
| first_fig = bundle["figures"][0] | |
| assert first_fig["name"] == "block_team_a_attack" | |
| # Descargar PNG individual | |
| png_resp = client.get(first_fig["png_url"]) | |
| assert png_resp.status_code == 200 | |
| assert png_resp.headers["content-type"] == "image/png" | |
| assert len(png_resp.content) > 1000 # algo razonable | |
| # ZIP completo | |
| zip_resp = client.get(f"/api/jobs/{job_id}/zip", params={"report": "block_zscore"}) | |
| assert zip_resp.status_code == 200 | |
| assert zip_resp.headers["content-type"] == "application/zip" | |