deemanone/padelnomics
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
f1181342ad0f55f26758569c7c987389dc6f7ba8
padelnomics/web/tests/test_phase0.py
Deeman 815fb7d98a feat: replace sequential IDs with opaque tokens in public URLs 
Sequential IDs in /planner/export/<id> and /leads/<id>/unlock leaked
business volume (e.g. export_id=47 reveals ~47 PDFs sold). Replace with
22-char URL-safe tokens that carry no countable information.

- Migration 0017: adds `token TEXT` to business_plan_exports and
  lead_requests, backfills existing rows with secrets.token_urlsafe(16),
  creates unique indexes for fast lookups
- billing/routes.py: INSERT into business_plan_exports includes token
- leads/routes.py: INSERT into lead_requests includes token; enqueue
  payload includes lead_token; verify_quote() looks up by token
- planner/routes.py: /export/<token> route (was /export/<int:export_id>)
- suppliers/routes.py: /leads/<token>/unlock (was /leads/<int:lead_id>)
- worker.py: email links use token for both export and verify URLs
- Templates: url_for() calls use token= param
- test_phase0.py: _submit_guest_quote() returns (lead_id, auth_token,
  lead_token); verify URL tests use opaque lead token

Integer PKs unchanged; admin routes unchanged.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-02-22 17:03:13 +01:00

701 lines
29 KiB
Python
Raw Blame History

"""
Phase 0 tests: guest mode, new calculator variables, heat score, quote flow, migration.
"""
import math
import pytest
from padelnomics.leads.routes import calculate_heat_score
from padelnomics.planner.calculator import DEFAULTS, calc, validate_state
ALL_COMBOS = [
("indoor", "rent"),
("indoor", "buy"),
("outdoor", "rent"),
("outdoor", "buy"),
]
def default_state(**overrides):
s = {**DEFAULTS, **overrides}
return validate_state(s)
def _assert_finite(obj, path=""):
if isinstance(obj, float):
assert math.isfinite(obj), f"Non-finite at {path}: {obj}"
elif isinstance(obj, dict):
for k, v in obj.items():
_assert_finite(v, f"{path}.{k}")
elif isinstance(obj, list):
for i, v in enumerate(obj):
_assert_finite(v, f"{path}[{i}]")
# ════════════════════════════════════════════════════════════
# Guest mode
# ════════════════════════════════════════════════════════════
class TestGuestMode:
async def test_planner_accessible_without_login(self, client):
"""GET /planner/ returns 200 for unauthenticated user."""
resp = await client.get("/en/planner/")
assert resp.status_code == 200
async def test_calculate_endpoint_works_without_login(self, client):
"""POST /planner/calculate returns HTML partial for guest."""
resp = await client.post(
"/en/planner/calculate",
data={"dblCourts": "4", "activeTab": "capex"},
)
assert resp.status_code == 200
html = (await resp.data).decode()
# HTMX endpoint returns an HTML partial containing CAPEX data
assert "capex" in html.lower() or "metric-card" in html
async def test_scenario_routes_require_login(self, client):
"""Save/load/delete/list scenarios still require auth."""
resp = await client.post(
"/en/planner/scenarios/save",
json={"name": "test", "state_json": "{}"},
)
assert resp.status_code in (302, 401)
async def test_planner_hides_save_for_guest(self, client):
"""Planner HTML does not render scenario controls for guests."""
resp = await client.get("/en/planner/")
html = (await resp.data).decode()
assert "saveScenarioBtn" not in html
async def test_planner_shows_save_for_auth(self, auth_client):
"""Planner HTML renders scenario controls for logged-in users."""
resp = await auth_client.get("/en/planner/")
html = (await resp.data).decode()
assert "saveScenarioBtn" in html
# ════════════════════════════════════════════════════════════
# New calculator variables — defaults
# ════════════════════════════════════════════════════════════
class TestNewCalculatorVariables:
def test_budget_target_in_defaults(self):
assert "budgetTarget" in DEFAULTS
assert DEFAULTS["budgetTarget"] == 0
def test_glass_type_in_defaults(self):
assert "glassType" in DEFAULTS
assert DEFAULTS["glassType"] == "standard"
def test_lighting_type_in_defaults(self):
assert "lightingType" in DEFAULTS
assert DEFAULTS["lightingType"] == "led_standard"
# ════════════════════════════════════════════════════════════
# Glass type
# ════════════════════════════════════════════════════════════
class TestGlassType:
def test_panoramic_glass_increases_capex(self):
d_std = calc(default_state(glassType="standard"))
d_pan = calc(default_state(glassType="panoramic"))
assert d_pan["capex"] > d_std["capex"]
@pytest.mark.parametrize("glass", ["standard", "panoramic"])
@pytest.mark.parametrize("venue,own", ALL_COMBOS)
def test_glass_type_all_combos(self, venue, own, glass):
d = calc(default_state(venue=venue, own=own, glassType=glass))
assert d["capex"] > 0
_assert_finite(d)
def test_panoramic_applies_1_4x_multiplier(self):
"""Panoramic courts cost 1.4x standard courts."""
d_std = calc(default_state(glassType="standard"))
d_pan = calc(default_state(glassType="panoramic"))
std_courts = next(i for i in d_std["capexItems"] if i["name"] == "Padel Courts")
pan_courts = next(i for i in d_pan["capexItems"] if i["name"] == "Padel Courts")
assert pan_courts["amount"] == pytest.approx(std_courts["amount"] * 1.4, abs=1)
# ════════════════════════════════════════════════════════════
# Lighting type
# ════════════════════════════════════════════════════════════
class TestLightingType:
def test_led_competition_increases_capex(self):
d_std = calc(default_state(lightingType="led_standard"))
d_comp = calc(default_state(lightingType="led_competition"))
assert d_comp["capex"] > d_std["capex"]
def test_natural_light_outdoor(self):
d_led = calc(default_state(venue="outdoor", lightingType="led_standard"))
d_nat = calc(default_state(venue="outdoor", lightingType="natural"))
assert d_nat["capex"] < d_led["capex"]
def test_natural_light_zeroes_outdoor_lighting(self):
d = calc(default_state(venue="outdoor", lightingType="natural"))
lighting_item = next(
(i for i in d["capexItems"] if i["name"] == "Lighting"), None
)
assert lighting_item is not None
assert lighting_item["amount"] == 0
@pytest.mark.parametrize("light", ["led_standard", "led_competition"])
@pytest.mark.parametrize("venue,own", ALL_COMBOS)
def test_lighting_type_all_combos(self, venue, own, light):
d = calc(default_state(venue=venue, own=own, lightingType=light))
assert d["capex"] > 0
_assert_finite(d)
# ════════════════════════════════════════════════════════════
# Budget target
# ════════════════════════════════════════════════════════════
class TestBudgetTarget:
def test_budget_variance_when_set(self):
d = calc(default_state(budgetTarget=300000))
assert "budgetVariance" in d
assert d["budgetVariance"] == d["capex"] - 300000
def test_budget_variance_zero_when_no_budget(self):
d = calc(default_state(budgetTarget=0))
assert d["budgetVariance"] == 0
assert d["budgetPct"] == 0
def test_budget_pct_calculated(self):
d = calc(default_state(budgetTarget=200000))
assert d["budgetPct"] == pytest.approx(d["capex"] / 200000 * 100)
def test_budget_target_passthrough(self):
d = calc(default_state(budgetTarget=500000))
assert d["budgetTarget"] == 500000
# ════════════════════════════════════════════════════════════
# Heat score
# ════════════════════════════════════════════════════════════
class TestHeatScore:
def test_hot_lead(self):
"""High-readiness signals = hot."""
form = {
"timeline": "asap",
"location_status": "lease_signed",
"financing_status": "self_funded",
"decision_process": "solo",
"previous_supplier_contact": "received_quotes",
"budget_estimate": "500000",
}
assert calculate_heat_score(form) == "hot"
def test_cool_lead(self):
"""Low-readiness signals = cool."""
form = {
"timeline": "12+mo",
"location_status": "still_searching",
"financing_status": "not_started",
"decision_process": "committee",
"previous_supplier_contact": "first_time",
"budget_estimate": "0",
}
assert calculate_heat_score(form) == "cool"
def test_warm_lead(self):
"""Mid-readiness signals = warm."""
form = {
"timeline": "3-6mo",
"location_status": "location_found",
"financing_status": "seeking",
"decision_process": "partners",
"budget_estimate": "150000",
}
assert calculate_heat_score(form) == "warm"
def test_empty_form_is_cool(self):
assert calculate_heat_score({}) == "cool"
def test_timeline_6_12mo_scores_1(self):
assert calculate_heat_score({"timeline": "6-12mo"}) == "cool"
def test_high_budget_alone_not_hot(self):
"""Budget alone shouldn't make a lead hot."""
assert calculate_heat_score({"budget_estimate": "1000000"}) == "cool"
def test_permit_granted_scores_4(self):
assert calculate_heat_score({"location_status": "permit_granted"}) == "cool"
# Combined with timeline makes it warm
form = {"location_status": "permit_granted", "timeline": "asap"}
assert calculate_heat_score(form) == "warm"
def test_permit_pending_scores_3(self):
form = {
"location_status": "permit_pending",
"timeline": "3-6mo",
"financing_status": "self_funded",
}
assert calculate_heat_score(form) == "warm"
def test_converting_existing_scores_2(self):
assert calculate_heat_score({"location_status": "converting_existing"}) == "cool"
def test_permit_not_filed_scores_2(self):
assert calculate_heat_score({"location_status": "permit_not_filed"}) == "cool"
def test_location_found_scores_1(self):
assert calculate_heat_score({"location_status": "location_found"}) == "cool"
# ════════════════════════════════════════════════════════════
# Quote request route
# ════════════════════════════════════════════════════════════
class TestQuoteRequest:
async def test_quote_form_loads(self, client):
"""GET /leads/quote returns 200 with wizard shell."""
resp = await client.get("/en/leads/quote")
assert resp.status_code == 200
async def test_quote_prefill_from_params(self, client):
"""Query params pre-fill the form and start on step 2."""
resp = await client.get("/en/leads/quote?venue=outdoor&courts=6")
assert resp.status_code == 200
async def test_quote_step_endpoint(self, client):
"""GET /leads/quote/step/1 returns 200 partial."""
resp = await client.get("/en/leads/quote/step/1")
assert resp.status_code == 200
async def test_quote_step_invalid(self, client):
"""GET /leads/quote/step/0 returns 400."""
resp = await client.get("/en/leads/quote/step/0")
assert resp.status_code == 400
async def test_quote_step_post_advances(self, client):
"""POST to step 1 with valid data returns step 2."""
await client.get("/en/leads/quote")
async with client.session_transaction() as sess:
csrf = sess.get("csrf_token", "")
resp = await client.post(
"/en/leads/quote/step/1",
form={
"_accumulated": "{}",
"facility_type": "indoor",
"court_count": "4",
"csrf_token": csrf,
},
)
assert resp.status_code == 200
html = (await resp.data).decode()
assert "Location" in html
async def test_quote_submit_creates_lead(self, client, db):
"""POST /leads/quote creates a lead_requests row as pending_verification."""
# Get CSRF token first
await client.get("/en/leads/quote")
async with client.session_transaction() as sess:
csrf = sess.get("csrf_token", "")
resp = await client.post(
"/en/leads/quote",
form={
"facility_type": "indoor",
"court_count": "4",
"glass_type": "panoramic",
"lighting_type": "led_standard",
"build_context": "new_standalone",
"country": "DE",
"timeline": "3-6mo",
"location_status": "location_found",
"financing_status": "self_funded",
"decision_process": "solo",
"stakeholder_type": "entrepreneur",
"contact_name": "Test User",
"contact_email": "test@example.com",
"contact_phone": "+491234567890",
"csrf_token": csrf,
},
)
assert resp.status_code == 200
async with db.execute("SELECT * FROM lead_requests WHERE lead_type = 'quote'") as cur:
rows = await cur.fetchall()
assert len(rows) == 1
row = dict(rows[0])
assert row["status"] == "pending_verification"
assert row["heat_score"] in ("hot", "warm", "cool")
assert row["contact_email"] == "test@example.com"
assert row["facility_type"] == "indoor"
assert row["stakeholder_type"] == "entrepreneur"
async def test_quote_submit_without_login(self, client, db):
"""Guests get a user created and linked; lead is pending_verification."""
await client.get("/en/leads/quote")
async with client.session_transaction() as sess:
csrf = sess.get("csrf_token", "")
resp = await client.post(
"/en/leads/quote",
form={
"facility_type": "indoor",
"court_count": "2",
"country": "DE",
"timeline": "3-6mo",
"stakeholder_type": "entrepreneur",
"contact_name": "Guest",
"contact_email": "guest@example.com",
"contact_phone": "+491234567890",
"csrf_token": csrf,
},
)
assert resp.status_code == 200
async with db.execute(
"SELECT user_id, status FROM lead_requests WHERE contact_email = 'guest@example.com'"
) as cur:
row = await cur.fetchone()
assert row is not None
assert row[0] is not None # user_id linked via get-or-create
assert row[1] == "pending_verification"
async def test_quote_submit_with_login(self, auth_client, db, test_user):
"""Logged-in user with matching email skips verification (status='new')."""
await auth_client.get("/en/leads/quote")
async with auth_client.session_transaction() as sess:
csrf = sess.get("csrf_token", "")
resp = await auth_client.post(
"/en/leads/quote",
form={
"facility_type": "outdoor",
"court_count": "6",
"country": "DE",
"timeline": "asap",
"stakeholder_type": "entrepreneur",
"contact_name": "Auth User",
"contact_email": "test@example.com", # matches test_user email
"contact_phone": "+491234567890",
"csrf_token": csrf,
},
)
assert resp.status_code == 200
async with db.execute(
"SELECT user_id, status FROM lead_requests WHERE contact_email = 'test@example.com'"
) as cur:
row = await cur.fetchone()
assert row is not None
assert row[0] == test_user["id"]
assert row[1] == "new"
async def test_venue_search_build_context(self, client, db):
"""Build context 'venue_search' is stored correctly."""
await client.get("/en/leads/quote")
async with client.session_transaction() as sess:
csrf = sess.get("csrf_token", "")
resp = await client.post(
"/en/leads/quote",
form={
"facility_type": "indoor",
"court_count": "4",
"build_context": "venue_search",
"country": "DE",
"timeline": "6-12mo",
"stakeholder_type": "developer",
"contact_name": "Venue Search",
"contact_email": "venue@example.com",
"contact_phone": "+491234567890",
"csrf_token": csrf,
},
)
assert resp.status_code == 200
async with db.execute(
"SELECT build_context FROM lead_requests WHERE contact_email = 'venue@example.com'"
) as cur:
row = await cur.fetchone()
assert row is not None
assert row[0] == "venue_search"
async def test_stakeholder_type_stored(self, client, db):
"""stakeholder_type field is stored correctly."""
await client.get("/en/leads/quote")
async with client.session_transaction() as sess:
csrf = sess.get("csrf_token", "")
resp = await client.post(
"/en/leads/quote",
form={
"facility_type": "indoor",
"court_count": "6",
"country": "DE",
"timeline": "asap",
"stakeholder_type": "tennis_club",
"contact_name": "Club Owner",
"contact_email": "club@example.com",
"contact_phone": "+491234567890",
"csrf_token": csrf,
},
)
assert resp.status_code == 200
async with db.execute(
"SELECT stakeholder_type FROM lead_requests WHERE contact_email = 'club@example.com'"
) as cur:
row = await cur.fetchone()
assert row is not None
assert row[0] == "tennis_club"
async def test_submitted_page_has_context(self, client):
"""Guest quote submission shows 'check your email' verify page."""
await client.get("/en/leads/quote")
async with client.session_transaction() as sess:
csrf = sess.get("csrf_token", "")
resp = await client.post(
"/en/leads/quote",
form={
"facility_type": "indoor",
"court_count": "6",
"country": "DE",
"timeline": "3-6mo",
"stakeholder_type": "entrepreneur",
"contact_name": "Context Test",
"contact_email": "ctx@example.com",
"contact_phone": "+491234567890",
"csrf_token": csrf,
},
)
assert resp.status_code == 200
html = (await resp.data).decode()
assert "check your email" in html.lower()
assert "ctx@example.com" in html
async def test_quote_validation_rejects_missing_fields(self, client):
"""POST /leads/quote returns 422 JSON when mandatory fields missing."""
await client.get("/en/leads/quote")
async with client.session_transaction() as sess:
csrf = sess.get("csrf_token", "")
resp = await client.post(
"/en/leads/quote",
json={
"facility_type": "indoor",
"court_count": "4",
"contact_name": "",
"contact_email": "",
},
headers={"X-CSRF-Token": csrf},
)
assert resp.status_code == 422
data = await resp.get_json()
assert data["ok"] is False
assert len(data["errors"]) >= 3 # country, timeline, stakeholder_type + name + email + phone
# ════════════════════════════════════════════════════════════
# Quote verification (double opt-in)
# ════════════════════════════════════════════════════════════
class TestQuoteVerification:
"""Double opt-in email verification for quote requests."""
QUOTE_FORM = {
"facility_type": "indoor",
"court_count": "4",
"country": "DE",
"timeline": "3-6mo",
"stakeholder_type": "entrepreneur",
"contact_name": "Verify Test",
"contact_email": "verify@example.com",
"contact_phone": "+491234567890",
}
async def _submit_guest_quote(self, client, db, email="verify@example.com"):
"""Helper: submit a quote as a guest, return (lead_id, auth_token, lead_token)."""
await client.get("/en/leads/quote")
async with client.session_transaction() as sess:
csrf = sess.get("csrf_token", "")
form = {**self.QUOTE_FORM, "contact_email": email, "csrf_token": csrf}
await client.post("/en/leads/quote", form=form)
async with db.execute(
"SELECT id, token FROM lead_requests WHERE contact_email = ?", (email,)
) as cur:
row = await cur.fetchone()
lead_id, lead_token = row[0], row[1]
async with db.execute(
"SELECT token FROM auth_tokens ORDER BY id DESC LIMIT 1"
) as cur:
auth_token = (await cur.fetchone())[0]
return lead_id, auth_token, lead_token
async def test_guest_quote_creates_pending_lead(self, client, db):
"""Guest quote creates lead with status='pending_verification'."""
lead_id, _, _lt = await self._submit_guest_quote(client, db)
async with db.execute(
"SELECT status FROM lead_requests WHERE id = ?", (lead_id,)
) as cur:
row = await cur.fetchone()
assert row[0] == "pending_verification"
async def test_logged_in_same_email_skips_verification(self, auth_client, db, test_user):
"""Logged-in user with matching email gets status='new' and 'matched' page."""
await auth_client.get("/en/leads/quote")
async with auth_client.session_transaction() as sess:
csrf = sess.get("csrf_token", "")
resp = await auth_client.post(
"/en/leads/quote",
form={
**self.QUOTE_FORM,
"contact_email": "test@example.com", # matches test_user
"csrf_token": csrf,
},
)
assert resp.status_code == 200
html = (await resp.data).decode()
assert "matched" in html.lower()
async with db.execute(
"SELECT status FROM lead_requests WHERE contact_email = 'test@example.com'"
) as cur:
row = await cur.fetchone()
assert row[0] == "new"
async def test_logged_in_different_email_needs_verification(self, auth_client, db, test_user):
"""Logged-in user with different email still needs verification."""
await auth_client.get("/en/leads/quote")
async with auth_client.session_transaction() as sess:
csrf = sess.get("csrf_token", "")
resp = await auth_client.post(
"/en/leads/quote",
form={
**self.QUOTE_FORM,
"contact_email": "other@example.com", # different from test_user
"csrf_token": csrf,
},
)
assert resp.status_code == 200
html = (await resp.data).decode()
assert "check your email" in html.lower()
async with db.execute(
"SELECT status FROM lead_requests WHERE contact_email = 'other@example.com'"
) as cur:
row = await cur.fetchone()
assert row[0] == "pending_verification"
async def test_verify_link_activates_lead(self, client, db):
"""GET /leads/verify with valid token sets status='new' and verified_at."""
lead_id, auth_token, lead_token = await self._submit_guest_quote(client, db)
resp = await client.get(f"/en/leads/verify?token={auth_token}&lead={lead_token}")
assert resp.status_code == 200
async with db.execute(
"SELECT status, verified_at FROM lead_requests WHERE id = ?", (lead_id,)
) as cur:
row = await cur.fetchone()
assert row[0] == "new"
assert row[1] is not None # verified_at timestamp set
async def test_verify_sets_session(self, client, db):
"""Verification link logs the user in (sets session user_id)."""
_lead_id, auth_token, lead_token = await self._submit_guest_quote(client, db)
await client.get(f"/en/leads/verify?token={auth_token}&lead={lead_token}")
async with client.session_transaction() as sess:
assert "user_id" in sess
async def test_verify_expired_token(self, client, db):
"""Expired/used token redirects with error."""
_lead_id, auth_token, lead_token = await self._submit_guest_quote(client, db)
# Expire the token
await db.execute("UPDATE auth_tokens SET expires_at = '2000-01-01T00:00:00'")
await db.commit()
resp = await client.get(
f"/en/leads/verify?token={auth_token}&lead={lead_token}",
follow_redirects=False,
)
assert resp.status_code == 302
async def test_verify_already_verified_lead(self, client, db):
"""Attempting to verify an already-activated lead shows error."""
lead_id, auth_token, lead_token = await self._submit_guest_quote(client, db)
# Manually activate the lead
await db.execute(
"UPDATE lead_requests SET status = 'new' WHERE id = ?", (lead_id,)
)
await db.commit()
resp = await client.get(
f"/en/leads/verify?token={auth_token}&lead={lead_token}",
follow_redirects=False,
)
assert resp.status_code == 302
async def test_verify_missing_params(self, client, db):
"""Missing token or lead params redirects."""
resp = await client.get("/en/leads/verify", follow_redirects=False)
assert resp.status_code == 302
resp = await client.get("/en/leads/verify?token=abc", follow_redirects=False)
assert resp.status_code == 302
async def test_guest_quote_creates_user(self, client, db):
"""Guest quote submission creates a user row for the contact email."""
await self._submit_guest_quote(client, db, email="newuser@example.com")
async with db.execute(
"SELECT id FROM users WHERE email = 'newuser@example.com'"
) as cur:
row = await cur.fetchone()
assert row is not None
# ════════════════════════════════════════════════════════════
# Migration / schema
# ════════════════════════════════════════════════════════════
class TestSchema:
async def test_schema_has_new_columns(self, db):
"""Fresh DB from schema.sql has all expanded lead_requests columns."""
async with db.execute("PRAGMA table_info(lead_requests)") as cur:
cols = {r[1] for r in await cur.fetchall()}
for expected in (
"facility_type", "glass_type", "lighting_type",
"build_context", "country", "timeline",
"location_status", "financing_status",
"heat_score", "contact_name", "contact_email",
"contact_phone", "contact_company",
"wants_financing_help", "decision_process",
"previous_supplier_contact", "services_needed",
"additional_info", "stakeholder_type", "verified_at",
):
assert expected in cols, f"Missing column: {expected}"
async def test_user_id_nullable(self, db):
"""lead_requests.user_id should accept NULL for guest leads."""
await db.execute(
"INSERT INTO lead_requests (lead_type, contact_email, created_at) VALUES (?, ?, datetime('now'))",
("quote", "guest@example.com"),
)
await db.commit()
async with db.execute(
"SELECT user_id FROM lead_requests WHERE contact_email = 'guest@example.com'"
) as cur:
row = await cur.fetchone()
assert row[0] is None
Reference in New Issue View Git Blame Copy Permalink