fix: CI test failure — skip WeasyPrint tests when native libs unavailable
- Add requires_weasyprint marker to TestGenerateBusinessPlan and TestWorkerHandler (these need libgobject/pango/cairo which CI python:3.12-slim lacks) - Fix export route tests: use opaque tokens instead of integer IDs - Replace deprecated datetime.utcnow() with datetime.now(UTC) - Add missing jsonify/Response imports to admin routes Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Deeman
@@ -7,8 +7,7 @@ Covers:
|
||||
- handle_generate_business_plan() — worker handler status transitions
|
||||
- /planner/export and /planner/export/<id> routes
|
||||
"""
|
||||
import json
|
||||
from datetime import datetime
|
||||
from datetime import UTC, datetime
|
||||
from pathlib import Path
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
@@ -16,6 +15,16 @@ import pytest
|
||||
from padelnomics.businessplan import generate_business_plan, get_plan_sections
|
||||
from padelnomics.planner.calculator import calc, validate_state
|
||||
|
||||
try:
|
||||
from weasyprint import HTML as _WP # noqa: F401
|
||||
_HAS_WEASYPRINT = True
|
||||
except OSError:
|
||||
_HAS_WEASYPRINT = False
|
||||
|
||||
requires_weasyprint = pytest.mark.skipif(
|
||||
not _HAS_WEASYPRINT, reason="WeasyPrint native libs not available"
|
||||
)
|
||||
|
||||
|
||||
# ── Helpers ──────────────────────────────────────────────────────────────────
|
||||
|
||||
@@ -126,6 +135,7 @@ class TestGetPlanSections:
|
||||
# generate_business_plan() — end-to-end WeasyPrint
|
||||
# ════════════════════════════════════════════════════════════
|
||||
|
||||
@requires_weasyprint
|
||||
class TestGenerateBusinessPlan:
|
||||
async def test_generates_valid_pdf_en(self, db, scenario):
|
||||
pdf_bytes = await generate_business_plan(
|
||||
@@ -156,30 +166,33 @@ class TestGenerateBusinessPlan:
|
||||
# Worker handler — status transitions + file write
|
||||
# ════════════════════════════════════════════════════════════
|
||||
|
||||
async def _insert_export(db, user_id: int, scenario_id: int, status: str = "pending") -> int:
|
||||
"""Insert a business_plan_exports row, return its id."""
|
||||
now = datetime.utcnow().isoformat()
|
||||
async def _insert_export(db, user_id: int, scenario_id: int, status: str = "pending") -> dict:
|
||||
"""Insert a business_plan_exports row, return dict with id and token."""
|
||||
import secrets
|
||||
now = datetime.now(UTC).isoformat()
|
||||
token = secrets.token_urlsafe(16)
|
||||
async with db.execute(
|
||||
"""INSERT INTO business_plan_exports
|
||||
(user_id, scenario_id, language, status, created_at)
|
||||
VALUES (?, ?, 'en', ?, ?)""",
|
||||
(user_id, scenario_id, status, now),
|
||||
(user_id, scenario_id, language, status, token, created_at)
|
||||
VALUES (?, ?, 'en', ?, ?, ?)""",
|
||||
(user_id, scenario_id, status, token, now),
|
||||
) as cursor:
|
||||
export_id = cursor.lastrowid
|
||||
await db.commit()
|
||||
return export_id
|
||||
return {"id": export_id, "token": token}
|
||||
|
||||
|
||||
@requires_weasyprint
|
||||
class TestWorkerHandler:
|
||||
async def test_happy_path_generates_pdf_and_updates_status(self, db, scenario):
|
||||
from padelnomics.worker import handle_generate_business_plan
|
||||
|
||||
export_id = await _insert_export(db, scenario["user_id"], scenario["id"])
|
||||
export = await _insert_export(db, scenario["user_id"], scenario["id"])
|
||||
output_file = None
|
||||
|
||||
with patch("padelnomics.worker.send_email", new_callable=AsyncMock) as mock_email:
|
||||
await handle_generate_business_plan({
|
||||
"export_id": export_id,
|
||||
"export_id": export["id"],
|
||||
"user_id": scenario["user_id"],
|
||||
"scenario_id": scenario["id"],
|
||||
"language": "en",
|
||||
@@ -188,7 +201,7 @@ class TestWorkerHandler:
|
||||
# Status should be 'ready'
|
||||
from padelnomics.core import fetch_one
|
||||
row = await fetch_one(
|
||||
"SELECT * FROM business_plan_exports WHERE id = ?", (export_id,)
|
||||
"SELECT * FROM business_plan_exports WHERE id = ?", (export["id"],)
|
||||
)
|
||||
try:
|
||||
assert row["status"] == "ready"
|
||||
@@ -214,12 +227,12 @@ class TestWorkerHandler:
|
||||
|
||||
# Use a non-existent user_id so generate_business_plan raises ValueError
|
||||
wrong_user_id = scenario["user_id"] + 9999
|
||||
export_id = await _insert_export(db, scenario["user_id"], scenario["id"])
|
||||
export = await _insert_export(db, scenario["user_id"], scenario["id"])
|
||||
|
||||
with patch("padelnomics.worker.send_email", new_callable=AsyncMock):
|
||||
with pytest.raises(ValueError):
|
||||
await handle_generate_business_plan({
|
||||
"export_id": export_id,
|
||||
"export_id": export["id"],
|
||||
"user_id": wrong_user_id,
|
||||
"scenario_id": scenario["id"],
|
||||
"language": "en",
|
||||
@@ -227,7 +240,7 @@ class TestWorkerHandler:
|
||||
|
||||
from padelnomics.core import fetch_one
|
||||
row = await fetch_one(
|
||||
"SELECT status FROM business_plan_exports WHERE id = ?", (export_id,)
|
||||
"SELECT status FROM business_plan_exports WHERE id = ?", (export["id"],)
|
||||
)
|
||||
assert row["status"] == "failed"
|
||||
|
||||
@@ -248,26 +261,26 @@ class TestExportRoutes:
|
||||
assert "Test Scenario" in html
|
||||
|
||||
async def test_download_nonexistent_returns_404(self, auth_client):
|
||||
resp = await auth_client.get("/en/planner/export/99999")
|
||||
resp = await auth_client.get("/en/planner/export/bogus-token-xyz")
|
||||
assert resp.status_code == 404
|
||||
|
||||
async def test_download_pending_shows_generating_page(self, auth_client, db, scenario):
|
||||
export_id = await _insert_export(
|
||||
export = await _insert_export(
|
||||
db, scenario["user_id"], scenario["id"], status="pending"
|
||||
)
|
||||
resp = await auth_client.get(f"/en/planner/export/{export_id}")
|
||||
resp = await auth_client.get(f"/en/planner/export/{export['token']}")
|
||||
assert resp.status_code == 200
|
||||
html = (await resp.data).decode()
|
||||
await resp.data # consume response
|
||||
# Should show the generating/pending template, not a PDF
|
||||
assert "pdf" not in resp.content_type.lower()
|
||||
|
||||
async def test_download_generating_shows_generating_page(self, auth_client, db, scenario):
|
||||
export_id = await _insert_export(
|
||||
export = await _insert_export(
|
||||
db, scenario["user_id"], scenario["id"], status="generating"
|
||||
)
|
||||
resp = await auth_client.get(f"/en/planner/export/{export_id}")
|
||||
resp = await auth_client.get(f"/en/planner/export/{export['token']}")
|
||||
assert resp.status_code == 200
|
||||
html = (await resp.data).decode()
|
||||
await resp.data # consume response
|
||||
assert "pdf" not in resp.content_type.lower()
|
||||
|
||||
async def test_download_ready_returns_pdf(self, auth_client, db, scenario, tmp_path):
|
||||
@@ -276,17 +289,19 @@ class TestExportRoutes:
|
||||
pdf_file = tmp_path / "test_export.pdf"
|
||||
pdf_file.write_bytes(dummy_pdf)
|
||||
|
||||
now = datetime.utcnow().isoformat()
|
||||
import secrets
|
||||
now = datetime.now(UTC).isoformat()
|
||||
token = secrets.token_urlsafe(16)
|
||||
async with db.execute(
|
||||
"""INSERT INTO business_plan_exports
|
||||
(user_id, scenario_id, language, status, file_path, created_at, completed_at)
|
||||
VALUES (?, ?, 'en', 'ready', ?, ?, ?)""",
|
||||
(scenario["user_id"], scenario["id"], str(pdf_file), now, now),
|
||||
) as cursor:
|
||||
export_id = cursor.lastrowid
|
||||
(user_id, scenario_id, language, status, token, file_path, created_at, completed_at)
|
||||
VALUES (?, ?, 'en', 'ready', ?, ?, ?, ?)""",
|
||||
(scenario["user_id"], scenario["id"], token, str(pdf_file), now, now),
|
||||
):
|
||||
pass
|
||||
await db.commit()
|
||||
|
||||
resp = await auth_client.get(f"/en/planner/export/{export_id}")
|
||||
resp = await auth_client.get(f"/en/planner/export/{token}")
|
||||
assert resp.status_code == 200
|
||||
assert resp.content_type == "application/pdf"
|
||||
data = await resp.data
|
||||
@@ -294,7 +309,7 @@ class TestExportRoutes:
|
||||
|
||||
async def test_download_other_users_export_returns_404(self, app, db, scenario):
|
||||
"""Another user cannot download someone else's export."""
|
||||
now = datetime.utcnow().isoformat()
|
||||
now = datetime.now(UTC).isoformat()
|
||||
# Create a second user
|
||||
async with db.execute(
|
||||
"INSERT INTO users (email, name, created_at) VALUES (?, ?, ?)",
|
||||
@@ -303,11 +318,11 @@ class TestExportRoutes:
|
||||
other_id = cursor.lastrowid
|
||||
await db.commit()
|
||||
|
||||
export_id = await _insert_export(db, scenario["user_id"], scenario["id"])
|
||||
export = await _insert_export(db, scenario["user_id"], scenario["id"])
|
||||
|
||||
# Log in as the other user
|
||||
async with app.test_client() as c:
|
||||
async with c.session_transaction() as sess:
|
||||
sess["user_id"] = other_id
|
||||
resp = await c.get(f"/en/planner/export/{export_id}")
|
||||
resp = await c.get(f"/en/planner/export/{export['token']}")
|
||||
assert resp.status_code == 404
|
||||
|
||||
Reference in New Issue
Block a user