diff --git a/web/src/padelnomics/admin/routes.py b/web/src/padelnomics/admin/routes.py index 5835db9..9734bec 100644 --- a/web/src/padelnomics/admin/routes.py +++ b/web/src/padelnomics/admin/routes.py @@ -1406,6 +1406,48 @@ async def scenario_preview(scenario_id: int): ) +@bp.route("/scenarios//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 # ============================================================================= diff --git a/web/src/padelnomics/admin/templates/admin/scenarios.html b/web/src/padelnomics/admin/templates/admin/scenarios.html index 4c51696..2ccb481 100644 --- a/web/src/padelnomics/admin/templates/admin/scenarios.html +++ b/web/src/padelnomics/admin/templates/admin/scenarios.html @@ -38,6 +38,8 @@ {{ s.created_at[:10] }} Preview + PDF EN + PDF DE Edit
diff --git a/web/src/padelnomics/templates/businessplan/plan.html b/web/src/padelnomics/templates/businessplan/plan.html index 9a2b9cd..e3de3a9 100644 --- a/web/src/padelnomics/templates/businessplan/plan.html +++ b/web/src/padelnomics/templates/businessplan/plan.html @@ -53,7 +53,7 @@ {{ s.labels.item }}{{ s.labels.amount }}{{ s.labels.notes }} - {% for item in s.investment.items %} + {% for item in s.investment['items'] %} {{ item.name }} {{ item.formatted_amount }} @@ -94,7 +94,7 @@ {{ s.labels.item }}{{ s.labels.monthly }}{{ s.labels.notes }} - {% for item in s.operations.items %} + {% for item in s.operations['items'] %} {{ item.name }} {{ item.formatted_amount }} diff --git a/web/tests/conftest.py b/web/tests/conftest.py index f803ebb..3fbdcc9 100644 --- a/web/tests/conftest.py +++ b/web/tests/conftest.py @@ -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) diff --git a/web/tests/test_businessplan.py b/web/tests/test_businessplan.py new file mode 100644 index 0000000..e45e9e7 --- /dev/null +++ b/web/tests/test_businessplan.py @@ -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/ 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