deemanone/padelnomics
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: admin scenario PDF download + business plan export tests

Browse Source 
Add /scenarios/<id>/pdf admin route for direct PDF generation via WeasyPrint.
Fix plan.html Jinja template: .items → ['items'] to avoid dict method collision.
Add scenario fixture in conftest.py and comprehensive test suite for business
plan sections, PDF generation, worker handler, and export routes.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Deeman
2026-02-22 21:27:32 +01:00
parent cac3b3b324
commit 76695f3902
5 changed files with 378 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
web/src/padelnomics/admin/routes.py
View File

@@ -1406,6 +1406,48 @@ async def scenario_preview(scenario_id: int):
)
@bp.route("/scenarios/<int:scenario_id>/pdf")
@role_required("admin")
async def scenario_pdf(scenario_id: int):
"""Generate and immediately download a business plan PDF for a published scenario."""
from quart import request as req
from ..businessplan import get_plan_sections
from ..planner.calculator import validate_state
scenario = await fetch_one("SELECT * FROM published_scenarios WHERE id = ?", (scenario_id,))
if not scenario:
return jsonify({"error": "Scenario not found."}), 404
lang = req.args.get("lang", "en")
if lang not in ("en", "de"):
lang = "en"
state = validate_state(json.loads(scenario["state_json"]))
d = json.loads(scenario["calc_json"])
sections = get_plan_sections(state, d, lang)
sections["scenario_name"] = scenario["title"]
sections["location"] = scenario.get("location", "")
from pathlib import Path
template_dir = Path(__file__).parent.parent / "templates" / "businessplan"
html_template = (template_dir / "plan.html").read_text()
css = (template_dir / "plan.css").read_text()
from jinja2 import Template
rendered_html = Template(html_template).render(s=sections, css=css)
from weasyprint import HTML
pdf_bytes = HTML(string=rendered_html).write_pdf()
slug = scenario["slug"] or f"scenario-{scenario_id}"
filename = f"padel-business-plan-{slug}-{lang}.pdf"
return Response(
pdf_bytes,
mimetype="application/pdf",
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
)
# =============================================================================
# Article Management
# =============================================================================

2
web/src/padelnomics/admin/templates/admin/scenarios.html
View File

@@ -38,6 +38,8 @@
<td class="mono text-sm">{{ s.created_at[:10] }}</td>
<td class="text-right">
<a href="{{ url_for('admin.scenario_preview', scenario_id=s.id) }}" class="btn-outline btn-sm">Preview</a>
<a href="{{ url_for('admin.scenario_pdf', scenario_id=s.id, lang='en') }}" class="btn-outline btn-sm">PDF EN</a>
<a href="{{ url_for('admin.scenario_pdf', scenario_id=s.id, lang='de') }}" class="btn-outline btn-sm">PDF DE</a>
<a href="{{ url_for('admin.scenario_edit', scenario_id=s.id) }}" class="btn-outline btn-sm">Edit</a>
<form method="post" action="{{ url_for('admin.scenario_delete', scenario_id=s.id) }}" class="m-0" style="display: inline;">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">

4
web/src/padelnomics/templates/businessplan/plan.html
View File

@@ -53,7 +53,7 @@
<tr><th>{{ s.labels.item }}</th><th style="text-align:right">{{ s.labels.amount }}</th><th>{{ s.labels.notes }}</th></tr>
</thead>
<tbody>
{% for item in s.investment.items %}
{% for item in s.investment['items'] %}
<tr>
<td>{{ item.name }}</td>
<td style="text-align:right">{{ item.formatted_amount }}</td>
@@ -94,7 +94,7 @@
<tr><th>{{ s.labels.item }}</th><th style="text-align:right">{{ s.labels.monthly }}</th><th>{{ s.labels.notes }}</th></tr>
</thead>
<tbody>
{% for item in s.operations.items %}
{% for item in s.operations['items'] %}
<tr>
<td>{{ item.name }}</td>
<td style="text-align:right">{{ item.formatted_amount }}</td>

19
web/tests/conftest.py
View File

@@ -141,6 +141,25 @@ def create_subscription(db):
return _create
# ── Scenarios ────────────────────────────────────────────────
@pytest.fixture
async def scenario(db, test_user):
"""User scenario with valid planner state for PDF generation."""
import json
from padelnomics.planner.calculator import validate_state
state = validate_state({"dblCourts": 4, "sglCourts": 2})
now = datetime.utcnow().isoformat()
async with db.execute(
"""INSERT INTO scenarios (user_id, name, state_json, created_at)
VALUES (?, 'Test Scenario', ?, ?)""",
(test_user["id"], json.dumps(state), now),
) as cursor:
scenario_id = cursor.lastrowid
await db.commit()
return {"id": scenario_id, "state": state, "user_id": test_user["id"]}
# ── Config ───────────────────────────────────────────────────
@pytest.fixture(autouse=True)

313
web/tests/test_businessplan.py Normal file
View File

@@ -0,0 +1,313 @@
"""
Tests for the business plan PDF export pipeline.
Covers:
- get_plan_sections() — pure formatting function
- generate_business_plan() — end-to-end WeasyPrint rendering
- handle_generate_business_plan() — worker handler status transitions
- /planner/export and /planner/export/<id> routes
"""
import json
from datetime import datetime
from pathlib import Path
from unittest.mock import AsyncMock, patch
import pytest
from padelnomics.businessplan import generate_business_plan, get_plan_sections
from padelnomics.planner.calculator import calc, validate_state
# ── Helpers ──────────────────────────────────────────────────────────────────
EXPECTED_SECTION_KEYS = {
"lang",
"title",
"subtitle",
"courts",
"executive_summary",
"investment",
"operations",
"revenue",
"annuals",
"financing",
"metrics",
"cashflow_12m",
"labels",
}
def _make_sections(language: str = "en", country: str = "DE") -> dict:
state = validate_state({"dblCourts": 4, "sglCourts": 2, "country": country})
d = calc(state)
return get_plan_sections(state, d, language)
# ════════════════════════════════════════════════════════════
# get_plan_sections() — pure function, no DB, no WeasyPrint
# ════════════════════════════════════════════════════════════
class TestGetPlanSections:
def test_returns_all_sections_en(self):
sections = _make_sections("en")
assert EXPECTED_SECTION_KEYS.issubset(sections.keys())
def test_returns_all_sections_de(self):
sections = _make_sections("de")
assert EXPECTED_SECTION_KEYS.issubset(sections.keys())
def test_lang_field_matches_requested_language(self):
assert _make_sections("en")["lang"] == "en"
assert _make_sections("de")["lang"] == "de"
def test_currency_formatting_eu_style(self):
"""German/EU style: € prefix, dot as thousands separator."""
sections = _make_sections("de", country="DE")
capex = sections["investment"]["total"]
# EU-style: €1.234.567 (dot separator, no comma)
assert capex.startswith("€")
assert "," not in capex # no comma separator in EU style
def test_currency_formatting_us_style(self):
"""US style: $ prefix, comma as thousands separator."""
sections = _make_sections("en", country="US")
capex = sections["investment"]["total"]
assert capex.startswith("$")
assert "." not in capex.replace(".", "") # commas, not dots
def test_payback_formatting_en(self):
sections = _make_sections("en")
payback = sections["executive_summary"]["payback"]
# Should be "X months" or "X.X years" or "not reached"
assert isinstance(payback, str)
assert len(payback) > 0
def test_irr_formatting(self):
sections = _make_sections("en")
irr = sections["metrics"]["irr"]
assert "%" in irr or irr == "-"
def test_executive_summary_has_required_keys(self):
sections = _make_sections("en")
es = sections["executive_summary"]
for key in ("heading", "courts", "sqm", "total_capex", "equity", "loan",
"y1_revenue", "y3_ebitda", "irr", "payback"):
assert key in es, f"Missing key in executive_summary: {key}"
def test_labels_has_required_keys(self):
sections = _make_sections("en")
labels = sections["labels"]
for key in ("scenario", "generated_by", "disclaimer", "currency_sym",
"total_investment", "equity_required", "year3_ebitda"):
assert key in labels, f"Missing key in labels: {key}"
def test_annuals_has_5_years(self):
sections = _make_sections("en")
years = sections["annuals"]["years"]
assert len(years) == 5
def test_cashflow_12m_has_12_months(self):
sections = _make_sections("en")
months = sections["cashflow_12m"]["months"]
assert len(months) == 12
def test_investment_items_have_formatted_amounts(self):
sections = _make_sections("en")
for item in sections["investment"]["items"]:
assert "formatted_amount" in item
def test_de_labels_are_german(self):
de_sections = _make_sections("de")
en_sections = _make_sections("en")
# Title should differ between languages
assert de_sections["title"] != en_sections["title"]
# ════════════════════════════════════════════════════════════
# generate_business_plan() — end-to-end WeasyPrint
# ════════════════════════════════════════════════════════════
class TestGenerateBusinessPlan:
async def test_generates_valid_pdf_en(self, db, scenario):
pdf_bytes = await generate_business_plan(
scenario["id"], scenario["user_id"], "en"
)
assert isinstance(pdf_bytes, bytes)
assert pdf_bytes[:4] == b"%PDF"
assert len(pdf_bytes) > 10_000 # non-trivially sized
async def test_generates_valid_pdf_de(self, db, scenario):
pdf_bytes = await generate_business_plan(
scenario["id"], scenario["user_id"], "de"
)
assert isinstance(pdf_bytes, bytes)
assert pdf_bytes[:4] == b"%PDF"
assert len(pdf_bytes) > 10_000
async def test_raises_for_missing_scenario(self, db, test_user):
with pytest.raises(ValueError, match="not found"):
await generate_business_plan(99999, test_user["id"], "en")
async def test_raises_for_wrong_user(self, db, scenario):
with pytest.raises(ValueError, match="not found"):
await generate_business_plan(scenario["id"], scenario["user_id"] + 999, "en")
# ════════════════════════════════════════════════════════════
# 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 with db.execute(
"""INSERT INTO business_plan_exports
(user_id, scenario_id, language, status, created_at)
VALUES (?, ?, 'en', ?, ?)""",
(user_id, scenario_id, status, now),
) as cursor:
export_id = cursor.lastrowid
await db.commit()
return export_id
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"])
output_file = None
with patch("padelnomics.worker.send_email", new_callable=AsyncMock) as mock_email:
await handle_generate_business_plan({
"export_id": export_id,
"user_id": scenario["user_id"],
"scenario_id": scenario["id"],
"language": "en",
})
# Status should be 'ready'
from padelnomics.core import fetch_one
row = await fetch_one(
"SELECT * FROM business_plan_exports WHERE id = ?", (export_id,)
)
try:
assert row["status"] == "ready"
assert row["file_path"] is not None
assert row["completed_at"] is not None
# File should exist on disk and be a valid PDF
output_file = Path(row["file_path"])
assert output_file.exists()
assert output_file.read_bytes()[:4] == b"%PDF"
# Email should have been sent
mock_email.assert_called_once()
assert "to" in mock_email.call_args.kwargs
assert "subject" in mock_email.call_args.kwargs
finally:
if output_file and output_file.exists():
output_file.unlink()
async def test_marks_failed_on_bad_scenario(self, db, scenario):
"""Handler marks export failed when user_id doesn't match scenario owner."""
from padelnomics.worker import handle_generate_business_plan
# 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"])
with patch("padelnomics.worker.send_email", new_callable=AsyncMock):
with pytest.raises(ValueError):
await handle_generate_business_plan({
"export_id": export_id,
"user_id": wrong_user_id,
"scenario_id": scenario["id"],
"language": "en",
})
from padelnomics.core import fetch_one
row = await fetch_one(
"SELECT status FROM business_plan_exports WHERE id = ?", (export_id,)
)
assert row["status"] == "failed"
# ════════════════════════════════════════════════════════════
# Export routes — thin HTTP layer
# ════════════════════════════════════════════════════════════
class TestExportRoutes:
async def test_export_page_requires_login(self, client):
resp = await client.get("/en/planner/export")
assert resp.status_code == 302
async def test_export_page_shows_scenarios(self, auth_client, db, scenario):
resp = await auth_client.get("/en/planner/export")
assert resp.status_code == 200
html = (await resp.data).decode()
assert "Test Scenario" in html
async def test_download_nonexistent_returns_404(self, auth_client):
resp = await auth_client.get("/en/planner/export/99999")
assert resp.status_code == 404
async def test_download_pending_shows_generating_page(self, auth_client, db, scenario):
export_id = await _insert_export(
db, scenario["user_id"], scenario["id"], status="pending"
)
resp = await auth_client.get(f"/en/planner/export/{export_id}")
assert resp.status_code == 200
html = (await resp.data).decode()
# 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(
db, scenario["user_id"], scenario["id"], status="generating"
)
resp = await auth_client.get(f"/en/planner/export/{export_id}")
assert resp.status_code == 200
html = (await resp.data).decode()
assert "pdf" not in resp.content_type.lower()
async def test_download_ready_returns_pdf(self, auth_client, db, scenario, tmp_path):
# Write a minimal valid PDF to a temp file
dummy_pdf = b"%PDF-1.4 tiny test pdf"
pdf_file = tmp_path / "test_export.pdf"
pdf_file.write_bytes(dummy_pdf)
now = datetime.utcnow().isoformat()
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
await db.commit()
resp = await auth_client.get(f"/en/planner/export/{export_id}")
assert resp.status_code == 200
assert resp.content_type == "application/pdf"
data = await resp.data
assert data == dummy_pdf
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()
# Create a second user
async with db.execute(
"INSERT INTO users (email, name, created_at) VALUES (?, ?, ?)",
("other@example.com", "Other", now),
) as cursor:
other_id = cursor.lastrowid
await db.commit()
export_id = 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}")
assert resp.status_code == 404