989 lines
41 KiB
Python
989 lines
41 KiB
Python
"""
|
|
Tests for waitlist mode (lean startup smoke test).
|
|
|
|
Covers configuration, migration, worker tasks, auth routes, supplier routes,
|
|
edge cases (duplicates, invalid emails), and full integration flows.
|
|
"""
|
|
|
|
import sqlite3
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
from padelnomics.worker import handle_send_waitlist_confirmation
|
|
|
|
from padelnomics import core
|
|
|
|
# ── Fixtures ──────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def mock_csrf_validation():
|
|
"""Mock CSRF validation to always pass in tests."""
|
|
with patch("padelnomics.core.validate_csrf_token", return_value=True):
|
|
yield
|
|
|
|
|
|
# ── Helpers ───────────────────────────────────────────────────────
|
|
|
|
|
|
async def _get_csrf_token(client):
|
|
"""Get a valid CSRF token by making a GET request first."""
|
|
await client.get("/")
|
|
async with client.session_transaction() as sess:
|
|
return sess.get("csrf_token", "test_fallback_token")
|
|
|
|
|
|
def _table_names(conn):
|
|
"""Return sorted list of user-visible table names (synchronous)."""
|
|
rows = conn.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table'"
|
|
" AND name NOT LIKE 'sqlite_%' ORDER BY name"
|
|
).fetchall()
|
|
return [r[0] for r in rows]
|
|
|
|
|
|
def _column_names(conn, table):
|
|
"""Return list of column names for a table (synchronous)."""
|
|
return [r[1] for r in conn.execute(f"PRAGMA table_info({table})").fetchall()]
|
|
|
|
|
|
async def _count_rows(db, table):
|
|
"""Count rows in a table (async)."""
|
|
async with db.execute(f"SELECT COUNT(*) as cnt FROM {table}") as cursor:
|
|
row = await cursor.fetchone()
|
|
return row[0] if row else 0
|
|
|
|
|
|
async def _get_waitlist_entry(db, email, intent="signup"):
|
|
"""Fetch a waitlist entry by email and intent (async)."""
|
|
async with db.execute(
|
|
"SELECT * FROM waitlist WHERE email = ? AND intent = ?",
|
|
(email, intent)
|
|
) as cursor:
|
|
row = await cursor.fetchone()
|
|
return dict(row) if row else None
|
|
|
|
|
|
# ── TestConfiguration ─────────────────────────────────────────────
|
|
|
|
|
|
class TestConfiguration:
|
|
"""Test WAITLIST_MODE config flag."""
|
|
|
|
def test_waitlist_mode_defaults_to_false(self):
|
|
"""WAITLIST_MODE should default to false when not set."""
|
|
# Config is loaded from env, but we can check the default behavior
|
|
# The _env helper treats empty string same as unset
|
|
assert hasattr(core.config, "WAITLIST_MODE")
|
|
assert isinstance(core.config.WAITLIST_MODE, bool)
|
|
|
|
def test_waitlist_mode_can_be_enabled(self):
|
|
"""WAITLIST_MODE can be set to True via config."""
|
|
with patch.object(core.config, "WAITLIST_MODE", True):
|
|
assert core.config.WAITLIST_MODE is True
|
|
|
|
def test_waitlist_mode_can_be_disabled(self):
|
|
"""WAITLIST_MODE can be set to False via config."""
|
|
with patch.object(core.config, "WAITLIST_MODE", False):
|
|
assert core.config.WAITLIST_MODE is False
|
|
|
|
|
|
# ── TestMigration ─────────────────────────────────────────────────
|
|
|
|
|
|
class TestMigration:
|
|
"""Test 0014_add_waitlist migration."""
|
|
|
|
def test_creates_waitlist_table(self, tmp_path):
|
|
"""Migration should create waitlist table."""
|
|
from padelnomics.migrations.migrate import migrate
|
|
db_path = str(tmp_path / "test.db")
|
|
migrate(db_path)
|
|
conn = sqlite3.connect(db_path)
|
|
tables = _table_names(conn)
|
|
conn.close()
|
|
assert "waitlist" in tables
|
|
|
|
def test_waitlist_table_has_correct_columns(self, tmp_path):
|
|
"""waitlist table should have all required columns."""
|
|
from padelnomics.migrations.migrate import migrate
|
|
db_path = str(tmp_path / "test.db")
|
|
migrate(db_path)
|
|
conn = sqlite3.connect(db_path)
|
|
cols = _column_names(conn, "waitlist")
|
|
conn.close()
|
|
assert "id" in cols
|
|
assert "email" in cols
|
|
assert "intent" in cols
|
|
assert "source" in cols
|
|
assert "plan" in cols
|
|
assert "ip_address" in cols
|
|
assert "created_at" in cols
|
|
|
|
def test_waitlist_has_unique_constraint(self, tmp_path):
|
|
"""waitlist should enforce UNIQUE(email, intent)."""
|
|
from padelnomics.migrations.migrate import migrate
|
|
db_path = str(tmp_path / "test.db")
|
|
migrate(db_path)
|
|
conn = sqlite3.connect(db_path)
|
|
|
|
# Insert first row
|
|
conn.execute(
|
|
"INSERT INTO waitlist (email, intent) VALUES (?, ?)",
|
|
("test@example.com", "signup")
|
|
)
|
|
conn.commit()
|
|
|
|
# Attempt duplicate - should fail without OR IGNORE
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
conn.execute(
|
|
"INSERT INTO waitlist (email, intent) VALUES (?, ?)",
|
|
("test@example.com", "signup")
|
|
)
|
|
conn.commit()
|
|
|
|
conn.close()
|
|
|
|
def test_waitlist_allows_same_email_different_intent(self, tmp_path):
|
|
"""waitlist should allow same email with different intent."""
|
|
from padelnomics.migrations.migrate import migrate
|
|
db_path = str(tmp_path / "test.db")
|
|
migrate(db_path)
|
|
conn = sqlite3.connect(db_path)
|
|
|
|
conn.execute(
|
|
"INSERT INTO waitlist (email, intent) VALUES (?, ?)",
|
|
("test@example.com", "signup")
|
|
)
|
|
conn.execute(
|
|
"INSERT INTO waitlist (email, intent) VALUES (?, ?)",
|
|
("test@example.com", "supplier_growth")
|
|
)
|
|
conn.commit()
|
|
|
|
count = conn.execute("SELECT COUNT(*) FROM waitlist").fetchone()[0]
|
|
conn.close()
|
|
assert count == 2
|
|
|
|
def test_waitlist_has_email_index(self, tmp_path):
|
|
"""Migration should create index on email column."""
|
|
from padelnomics.migrations.migrate import migrate
|
|
db_path = str(tmp_path / "test.db")
|
|
migrate(db_path)
|
|
conn = sqlite3.connect(db_path)
|
|
|
|
indexes = conn.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='index' AND tbl_name='waitlist'"
|
|
).fetchall()
|
|
index_names = [idx[0] for idx in indexes]
|
|
|
|
conn.close()
|
|
assert any("email" in idx for idx in index_names)
|
|
|
|
|
|
# ── TestWorkerTask ────────────────────────────────────────────────
|
|
|
|
|
|
class TestWorkerTask:
|
|
"""Test send_waitlist_confirmation worker task."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sends_entrepreneur_confirmation(self):
|
|
"""Task sends confirmation email for entrepreneur signup."""
|
|
with patch("padelnomics.worker.send_email", new_callable=AsyncMock) as mock_send:
|
|
await handle_send_waitlist_confirmation({
|
|
"email": "entrepreneur@example.com",
|
|
"intent": "signup",
|
|
})
|
|
|
|
mock_send.assert_called_once()
|
|
call_args = mock_send.call_args
|
|
assert call_args.kwargs["to"] == "entrepreneur@example.com"
|
|
assert "launching soon" in call_args.kwargs["subject"].lower()
|
|
assert "waitlist" in call_args.kwargs["html"].lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sends_supplier_confirmation(self):
|
|
"""Task sends confirmation email for supplier signup."""
|
|
with patch("padelnomics.worker.send_email", new_callable=AsyncMock) as mock_send:
|
|
await handle_send_waitlist_confirmation({
|
|
"email": "supplier@example.com",
|
|
"intent": "supplier_growth",
|
|
})
|
|
|
|
mock_send.assert_called_once()
|
|
call_args = mock_send.call_args
|
|
assert call_args.kwargs["to"] == "supplier@example.com"
|
|
assert "growth" in call_args.kwargs["subject"].lower()
|
|
assert "supplier" in call_args.kwargs["html"].lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_supplier_email_includes_plan_name(self):
|
|
"""Supplier confirmation should mention the specific plan."""
|
|
with patch("padelnomics.worker.send_email", new_callable=AsyncMock) as mock_send:
|
|
await handle_send_waitlist_confirmation({
|
|
"email": "supplier@example.com",
|
|
"intent": "supplier_pro",
|
|
})
|
|
|
|
call_args = mock_send.call_args
|
|
html = call_args.kwargs["html"]
|
|
assert "pro" in html.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_uses_transactional_email_address(self):
|
|
"""Task should use transactional sender address."""
|
|
with patch("padelnomics.worker.send_email", new_callable=AsyncMock) as mock_send:
|
|
await handle_send_waitlist_confirmation({
|
|
"email": "test@example.com",
|
|
"intent": "signup",
|
|
})
|
|
|
|
call_args = mock_send.call_args
|
|
assert call_args.kwargs["from_addr"] == core.EMAIL_ADDRESSES["transactional"]
|
|
|
|
|
|
# ── TestAuthRoutes ────────────────────────────────────────────────
|
|
|
|
|
|
class TestAuthRoutes:
|
|
"""Test /auth/signup route with WAITLIST_MODE."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_normal_signup_when_waitlist_disabled(self, client, db):
|
|
"""Normal signup flow when WAITLIST_MODE is false."""
|
|
with patch.object(core.config, "WAITLIST_MODE", False):
|
|
response = await client.get("/auth/signup")
|
|
assert response.status_code == 200
|
|
html = await response.get_data(as_text=True)
|
|
# Should see normal signup form, not waitlist form
|
|
assert "Create Free Account" in html or "Sign Up" in html
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shows_waitlist_form_when_enabled(self, client, db):
|
|
"""GET /auth/signup shows waitlist form when WAITLIST_MODE is true."""
|
|
with patch.object(core.config, "WAITLIST_MODE", True):
|
|
response = await client.get("/auth/signup")
|
|
assert response.status_code == 200
|
|
html = await response.get_data(as_text=True)
|
|
assert "waitlist" in html.lower()
|
|
assert "join" in html.lower() or "early access" in html.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_captures_email_to_waitlist_table(self, client, db):
|
|
"""POST /auth/signup inserts email into waitlist table."""
|
|
with patch.object(core.config, "WAITLIST_MODE", True), \
|
|
patch("padelnomics.worker.enqueue", new_callable=AsyncMock):
|
|
|
|
await client.post("/auth/signup", form={
|
|
"csrf_token": "test_token",
|
|
"email": "new@example.com",
|
|
"plan": "free",
|
|
})
|
|
|
|
# Check database
|
|
entry = await _get_waitlist_entry(db, "new@example.com", "free")
|
|
assert entry is not None
|
|
assert entry["email"] == "new@example.com"
|
|
assert entry["intent"] == "free"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_enqueues_confirmation_email(self, client, db):
|
|
"""POST /auth/signup enqueues waitlist confirmation email."""
|
|
with patch.object(core.config, "WAITLIST_MODE", True), \
|
|
patch("padelnomics.worker.enqueue", new_callable=AsyncMock) as mock_enqueue:
|
|
|
|
await client.post("/auth/signup", form={
|
|
"csrf_token": "test_token",
|
|
"email": "new@example.com",
|
|
"plan": "signup",
|
|
})
|
|
|
|
mock_enqueue.assert_called_once_with(
|
|
"send_waitlist_confirmation",
|
|
{"email": "new@example.com", "intent": "signup"}
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shows_confirmation_page(self, client, db):
|
|
"""POST /auth/signup shows waitlist_confirmed.html."""
|
|
with patch.object(core.config, "WAITLIST_MODE", True), \
|
|
patch("padelnomics.worker.enqueue", new_callable=AsyncMock):
|
|
|
|
response = await client.post("/auth/signup", form={
|
|
"csrf_token": "test_token",
|
|
"email": "new@example.com",
|
|
})
|
|
|
|
assert response.status_code == 200
|
|
html = await response.get_data(as_text=True)
|
|
assert "new@example.com" in html
|
|
assert "waitlist" in html.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handles_duplicate_email_gracefully(self, client, db):
|
|
"""Duplicate email submission shows same success page."""
|
|
with patch.object(core.config, "WAITLIST_MODE", True), \
|
|
patch("padelnomics.worker.enqueue", new_callable=AsyncMock):
|
|
|
|
# Insert first entry directly
|
|
await db.execute(
|
|
"INSERT INTO waitlist (email, intent) VALUES (?, ?)",
|
|
("existing@example.com", "signup")
|
|
)
|
|
await db.commit()
|
|
|
|
# Submit same email via form
|
|
response = await client.post("/auth/signup", form={
|
|
"csrf_token": "test_token",
|
|
"email": "existing@example.com",
|
|
"plan": "signup",
|
|
})
|
|
|
|
# Should show success page (not error)
|
|
assert response.status_code == 200
|
|
html = await response.get_data(as_text=True)
|
|
assert "existing@example.com" in html
|
|
|
|
# Should only have one row
|
|
count = await _count_rows(db, "waitlist")
|
|
assert count == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rejects_invalid_email(self, client, db):
|
|
"""POST with invalid email shows error."""
|
|
with patch.object(core.config, "WAITLIST_MODE", True):
|
|
response = await client.post("/auth/signup", form={
|
|
"csrf_token": "test_token",
|
|
"email": "not-an-email",
|
|
}, follow_redirects=True)
|
|
|
|
html = await response.get_data(as_text=True)
|
|
assert "valid email" in html.lower() or "error" in html.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_captures_ip_address(self, client, db):
|
|
"""POST captures request IP address in waitlist table."""
|
|
with patch.object(core.config, "WAITLIST_MODE", True), \
|
|
patch("padelnomics.worker.enqueue", new_callable=AsyncMock):
|
|
|
|
await client.post("/auth/signup", form={
|
|
"csrf_token": "test_token",
|
|
"email": "test@example.com",
|
|
})
|
|
|
|
entry = await _get_waitlist_entry(db, "test@example.com")
|
|
assert entry is not None
|
|
assert entry["ip_address"] is not None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_adds_to_resend_audience_when_configured(self, client, db):
|
|
"""POST auto-creates Resend audience per blueprint and adds the contact."""
|
|
with patch.object(core.config, "WAITLIST_MODE", True), \
|
|
patch.object(core.config, "RESEND_API_KEY", "re_test_key"), \
|
|
patch("padelnomics.worker.enqueue", new_callable=AsyncMock), \
|
|
patch("resend.Audiences.create", return_value={"id": "aud_test"}) as mock_create_aud, \
|
|
patch("resend.Contacts.create") as mock_create_contact:
|
|
|
|
await client.post("/auth/signup", form={
|
|
"csrf_token": "test_token",
|
|
"email": "test@example.com",
|
|
})
|
|
|
|
mock_create_aud.assert_called_once_with({"name": "waitlist-auth"})
|
|
mock_create_contact.assert_called_once()
|
|
call_args = mock_create_contact.call_args[0][0]
|
|
assert call_args["email"] == "test@example.com"
|
|
assert call_args["audience_id"] == "aud_test"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resend_audience_cached_on_second_signup(self, client, db):
|
|
"""Second signup for same blueprint reuses cached audience — no extra API call."""
|
|
with patch.object(core.config, "WAITLIST_MODE", True), \
|
|
patch.object(core.config, "RESEND_API_KEY", "re_test_key"), \
|
|
patch("padelnomics.worker.enqueue", new_callable=AsyncMock), \
|
|
patch("resend.Audiences.create", return_value={"id": "aud_cached"}) as mock_create_aud, \
|
|
patch("resend.Contacts.create"):
|
|
|
|
await client.post("/auth/signup", form={
|
|
"csrf_token": "test_token",
|
|
"email": "first@example.com",
|
|
})
|
|
await client.post("/auth/signup", form={
|
|
"csrf_token": "test_token",
|
|
"email": "second@example.com",
|
|
})
|
|
|
|
# Audience created only once; second call reads from DB cache
|
|
mock_create_aud.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_supplier_and_auth_signups_go_to_different_audiences(self, client, db):
|
|
"""Auth and supplier signups are added to separate Resend audiences."""
|
|
with patch.object(core.config, "WAITLIST_MODE", True), \
|
|
patch.object(core.config, "RESEND_API_KEY", "re_test_key"), \
|
|
patch("padelnomics.worker.enqueue", new_callable=AsyncMock), \
|
|
patch("resend.Audiences.create", side_effect=[
|
|
{"id": "aud_auth"},
|
|
{"id": "aud_suppliers"},
|
|
]) as mock_create_aud, \
|
|
patch("resend.Contacts.create") as mock_create_contact:
|
|
|
|
await client.post("/auth/signup", form={
|
|
"csrf_token": "test_token",
|
|
"email": "entrepreneur@example.com",
|
|
})
|
|
await client.post("/en/suppliers/signup/waitlist", form={
|
|
"csrf_token": "test_token",
|
|
"email": "supplier@example.com",
|
|
"plan": "supplier_growth",
|
|
})
|
|
|
|
audience_names = [
|
|
call[0][0]["name"] for call in mock_create_aud.call_args_list
|
|
]
|
|
assert "waitlist-auth" in audience_names
|
|
assert "waitlist-suppliers" in audience_names
|
|
|
|
contact_audience_ids = [
|
|
call[0][0]["audience_id"] for call in mock_create_contact.call_args_list
|
|
]
|
|
assert "aud_auth" in contact_audience_ids
|
|
assert "aud_suppliers" in contact_audience_ids
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resend_audience_create_failure_still_succeeds(self, client, db):
|
|
"""Audiences.create failure is silent — signup still returns success page."""
|
|
with patch.object(core.config, "WAITLIST_MODE", True), \
|
|
patch.object(core.config, "RESEND_API_KEY", "re_test_key"), \
|
|
patch("padelnomics.worker.enqueue", new_callable=AsyncMock), \
|
|
patch("resend.Audiences.create", side_effect=Exception("Resend down")):
|
|
|
|
response = await client.post("/auth/signup", form={
|
|
"csrf_token": "test_token",
|
|
"email": "test@example.com",
|
|
})
|
|
|
|
assert response.status_code == 200
|
|
html = await response.get_data(as_text=True)
|
|
assert "test@example.com" in html
|
|
|
|
|
|
# ── TestSupplierRoutes ────────────────────────────────────────────
|
|
|
|
|
|
class TestSupplierRoutes:
|
|
"""Test /suppliers/signup route with WAITLIST_MODE."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shows_waitlist_form_when_enabled(self, client, db):
|
|
"""GET /suppliers/signup shows waitlist form when WAITLIST_MODE is true."""
|
|
with patch.object(core.config, "WAITLIST_MODE", True):
|
|
response = await client.get("/en/suppliers/signup?plan=supplier_growth")
|
|
assert response.status_code == 200
|
|
html = await response.get_data(as_text=True)
|
|
assert "waitlist" in html.lower()
|
|
assert "supplier" in html.lower()
|
|
assert "growth" in html.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shows_normal_wizard_when_disabled(self, client, db):
|
|
"""GET /suppliers/signup shows normal wizard when WAITLIST_MODE is false."""
|
|
with patch.object(core.config, "WAITLIST_MODE", False):
|
|
response = await client.get("/en/suppliers/signup")
|
|
assert response.status_code == 200
|
|
html = await response.get_data(as_text=True)
|
|
# Should see wizard step 1, not waitlist form
|
|
assert "step" in html.lower() or "plan" in html.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_captures_supplier_email_to_waitlist(self, client, db):
|
|
"""POST /suppliers/signup/waitlist inserts email into waitlist table."""
|
|
with patch.object(core.config, "WAITLIST_MODE", True), \
|
|
patch("padelnomics.worker.enqueue", new_callable=AsyncMock):
|
|
|
|
await client.post("/en/suppliers/signup/waitlist", form={
|
|
"csrf_token": "test_token",
|
|
"email": "supplier@example.com",
|
|
"plan": "supplier_growth",
|
|
})
|
|
|
|
# Check database
|
|
entry = await _get_waitlist_entry(db, "supplier@example.com", "supplier")
|
|
assert entry is not None
|
|
assert entry["email"] == "supplier@example.com"
|
|
assert entry["intent"] == "supplier"
|
|
assert entry["plan"] == "supplier_growth"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_enqueues_supplier_confirmation_email(self, client, db):
|
|
"""POST /suppliers/signup/waitlist enqueues confirmation with plan intent."""
|
|
with patch.object(core.config, "WAITLIST_MODE", True), \
|
|
patch("padelnomics.worker.enqueue", new_callable=AsyncMock) as mock_enqueue:
|
|
|
|
await client.post("/en/suppliers/signup/waitlist", form={
|
|
"csrf_token": "test_token",
|
|
"email": "supplier@example.com",
|
|
"plan": "supplier_pro",
|
|
})
|
|
|
|
mock_enqueue.assert_called_once_with(
|
|
"send_waitlist_confirmation",
|
|
{"email": "supplier@example.com", "intent": "supplier_pro"}
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shows_supplier_confirmation_page(self, client, db):
|
|
"""POST /suppliers/signup/waitlist shows supplier-specific confirmation."""
|
|
with patch.object(core.config, "WAITLIST_MODE", True), \
|
|
patch("padelnomics.worker.enqueue", new_callable=AsyncMock):
|
|
|
|
response = await client.post("/en/suppliers/signup/waitlist", form={
|
|
"csrf_token": "test_token",
|
|
"email": "supplier@example.com",
|
|
"plan": "supplier_growth",
|
|
})
|
|
|
|
assert response.status_code == 200
|
|
html = await response.get_data(as_text=True)
|
|
assert "supplier@example.com" in html
|
|
assert "supplier" in html.lower()
|
|
assert "waitlist" in html.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handles_duplicate_supplier_email(self, client, db):
|
|
"""Duplicate supplier email shows same success page."""
|
|
with patch.object(core.config, "WAITLIST_MODE", True), \
|
|
patch("padelnomics.worker.enqueue", new_callable=AsyncMock):
|
|
|
|
# Insert first entry
|
|
await db.execute(
|
|
"INSERT INTO waitlist (email, intent, plan) VALUES (?, ?, ?)",
|
|
("existing@supplier.com", "supplier", "supplier_growth")
|
|
)
|
|
await db.commit()
|
|
|
|
# Submit duplicate
|
|
response = await client.post("/en/suppliers/signup/waitlist", form={
|
|
"csrf_token": "test_token",
|
|
"email": "existing@supplier.com",
|
|
"plan": "supplier_growth",
|
|
})
|
|
|
|
# Should show success page
|
|
assert response.status_code == 200
|
|
|
|
# Should still have only one row
|
|
count = await _count_rows(db, "waitlist")
|
|
assert count == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rejects_invalid_supplier_email(self, client, db):
|
|
"""POST with invalid email redirects with error."""
|
|
with patch.object(core.config, "WAITLIST_MODE", True):
|
|
response = await client.post("/en/suppliers/signup/waitlist", form={
|
|
"csrf_token": "test_token",
|
|
"email": "invalid",
|
|
"plan": "supplier_growth",
|
|
}, follow_redirects=True)
|
|
|
|
html = await response.get_data(as_text=True)
|
|
assert "valid email" in html.lower() or "error" in html.lower()
|
|
|
|
|
|
# ── TestEdgeCases ─────────────────────────────────────────────────
|
|
|
|
|
|
class TestEdgeCases:
|
|
"""Test edge cases and error handling."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_empty_email_rejected(self, client, db):
|
|
"""Empty email field shows error."""
|
|
with patch.object(core.config, "WAITLIST_MODE", True):
|
|
response = await client.post("/auth/signup", form={
|
|
"csrf_token": "test_token",
|
|
"email": "",
|
|
}, follow_redirects=True)
|
|
|
|
html = await response.get_data(as_text=True)
|
|
assert "valid email" in html.lower() or "error" in html.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_whitespace_only_email_rejected(self, client, db):
|
|
"""Whitespace-only email shows error."""
|
|
with patch.object(core.config, "WAITLIST_MODE", True):
|
|
response = await client.post("/auth/signup", form={
|
|
"csrf_token": "test_token",
|
|
"email": " ",
|
|
}, follow_redirects=True)
|
|
|
|
html = await response.get_data(as_text=True)
|
|
assert "valid email" in html.lower() or "error" in html.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_email_normalized_to_lowercase(self, client, db):
|
|
"""Email addresses are normalized to lowercase."""
|
|
with patch.object(core.config, "WAITLIST_MODE", True), \
|
|
patch("padelnomics.worker.enqueue", new_callable=AsyncMock):
|
|
|
|
await client.post("/auth/signup", form={
|
|
"csrf_token": "test_token",
|
|
"email": "Test@EXAMPLE.COM",
|
|
})
|
|
|
|
entry = await _get_waitlist_entry(db, "test@example.com")
|
|
assert entry is not None
|
|
assert entry["email"] == "test@example.com"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_database_error_shows_success_page(self, client, db):
|
|
"""Database errors are silently caught and success page still shown."""
|
|
with patch.object(core.config, "WAITLIST_MODE", True), \
|
|
patch("padelnomics.worker.enqueue", new_callable=AsyncMock), \
|
|
patch("padelnomics.core.execute", side_effect=Exception("DB error")):
|
|
|
|
response = await client.post("/auth/signup", form={
|
|
"csrf_token": "test_token",
|
|
"email": "test@example.com",
|
|
})
|
|
|
|
# Should still show success page (fail silently)
|
|
assert response.status_code == 200
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resend_api_error_handled_gracefully(self, client, db):
|
|
"""Resend API errors don't break the flow."""
|
|
with patch.object(core.config, "WAITLIST_MODE", True), \
|
|
patch.object(core.config, "RESEND_API_KEY", "re_test_key"), \
|
|
patch("padelnomics.worker.enqueue", new_callable=AsyncMock), \
|
|
patch("resend.Audiences.create", return_value={"id": "aud_test"}), \
|
|
patch("resend.Contacts.create", side_effect=Exception("API error")):
|
|
|
|
response = await client.post("/auth/signup", form={
|
|
"csrf_token": "test_token",
|
|
"email": "test@example.com",
|
|
})
|
|
|
|
# Should still show success page
|
|
assert response.status_code == 200
|
|
html = await response.get_data(as_text=True)
|
|
assert "test@example.com" in html
|
|
|
|
|
|
# ── TestIntegration ───────────────────────────────────────────────
|
|
|
|
|
|
class TestPlannerExport:
|
|
"""Test planner export waitlist gating."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_export_shows_waitlist_when_enabled(self, auth_client, db):
|
|
"""GET /planner/export shows waitlist page when WAITLIST_MODE is true."""
|
|
with patch.object(core.config, "WAITLIST_MODE", True):
|
|
response = await auth_client.get("/en/planner/export")
|
|
assert response.status_code == 200
|
|
html = await response.get_data(as_text=True)
|
|
assert "coming soon" in html.lower()
|
|
assert "business plan" in html.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_export_shows_normal_page_when_disabled(self, auth_client, db):
|
|
"""GET /planner/export shows normal export page when WAITLIST_MODE is false."""
|
|
with patch.object(core.config, "WAITLIST_MODE", False):
|
|
response = await auth_client.get("/en/planner/export")
|
|
assert response.status_code == 200
|
|
html = await response.get_data(as_text=True)
|
|
# Should see normal export page, not waitlist
|
|
assert "scenario" in html.lower() or "language" in html.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_export_requires_login(self, client, db):
|
|
"""Export page requires authentication."""
|
|
with patch.object(core.config, "WAITLIST_MODE", True):
|
|
response = await client.get("/en/planner/export", follow_redirects=False)
|
|
# Should redirect to login
|
|
assert response.status_code == 302 or response.status_code == 401
|
|
|
|
|
|
# ── TestWaitlistGateDecorator ─────────────────────────────────────
|
|
|
|
|
|
class TestWaitlistGateDecorator:
|
|
"""Test waitlist_gate decorator via integration tests."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_passes_through_when_waitlist_disabled(self, client):
|
|
"""Decorator passes through to normal flow when WAITLIST_MODE=false."""
|
|
with patch.object(core.config, "WAITLIST_MODE", False):
|
|
response = await client.get("/auth/signup")
|
|
html = await response.get_data(as_text=True)
|
|
# Should see normal signup, not waitlist
|
|
assert "waitlist" not in html.lower() or "create" in html.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_intercepts_get_when_waitlist_enabled(self, client):
|
|
"""Decorator intercepts GET requests when WAITLIST_MODE=true."""
|
|
with patch.object(core.config, "WAITLIST_MODE", True):
|
|
response = await client.get("/auth/signup")
|
|
html = await response.get_data(as_text=True)
|
|
# Should see waitlist page
|
|
assert "waitlist" in html.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ignores_post_requests(self, client, db):
|
|
"""Decorator lets POST requests through even when WAITLIST_MODE=true."""
|
|
with patch.object(core.config, "WAITLIST_MODE", True), \
|
|
patch("padelnomics.worker.enqueue", new_callable=AsyncMock):
|
|
# POST should still be handled by waitlist logic, not bypassed
|
|
response = await client.post("/auth/signup", form={
|
|
"csrf_token": "test_token",
|
|
"email": "test@example.com",
|
|
})
|
|
assert response.status_code == 200
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_evaluates_callable_context(self, client):
|
|
"""Decorator evaluates callable context values at request time."""
|
|
with patch.object(core.config, "WAITLIST_MODE", True):
|
|
# Test that plan query param is passed to template
|
|
response = await client.get("/auth/signup?plan=starter")
|
|
assert response.status_code == 200
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handles_complex_context_variables(self, client):
|
|
"""Decorator handles multiple context variables for suppliers."""
|
|
with patch.object(core.config, "WAITLIST_MODE", True):
|
|
response = await client.get("/en/suppliers/signup?plan=supplier_pro")
|
|
html = await response.get_data(as_text=True)
|
|
assert response.status_code == 200
|
|
assert "supplier" in html.lower()
|
|
|
|
|
|
# ── TestCaptureWaitlistEmail ──────────────────────────────────────
|
|
|
|
|
|
class TestCaptureWaitlistEmail:
|
|
"""Test capture_waitlist_email helper via integration tests."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_new_insert_returns_true(self, client, db):
|
|
"""Submitting new email creates database entry."""
|
|
with patch.object(core.config, "WAITLIST_MODE", True), \
|
|
patch("padelnomics.worker.enqueue", new_callable=AsyncMock):
|
|
|
|
await client.post("/auth/signup", form={
|
|
"csrf_token": "test_token",
|
|
"email": "new@example.com",
|
|
})
|
|
|
|
# Verify DB entry created
|
|
entry = await _get_waitlist_entry(db, "new@example.com")
|
|
assert entry is not None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_duplicate_returns_false(self, client, db):
|
|
"""Submitting duplicate email doesn't create second entry."""
|
|
# Insert first entry
|
|
await db.execute(
|
|
"INSERT INTO waitlist (email, intent) VALUES (?, ?)",
|
|
("existing@example.com", "signup")
|
|
)
|
|
await db.commit()
|
|
|
|
with patch.object(core.config, "WAITLIST_MODE", True), \
|
|
patch("padelnomics.worker.enqueue", new_callable=AsyncMock):
|
|
|
|
await client.post("/auth/signup", form={
|
|
"csrf_token": "test_token",
|
|
"email": "existing@example.com",
|
|
})
|
|
|
|
# Should still have only one row
|
|
count = await _count_rows(db, "waitlist")
|
|
assert count == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_enqueues_only_on_new(self, client, db):
|
|
"""Email confirmation only enqueued for new signups."""
|
|
# Already tested by test_enqueues_confirmation_email in TestAuthRoutes
|
|
pass
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_adds_to_resend_audience(self, client, db):
|
|
"""Submitting email adds to Resend audience when configured."""
|
|
# Already tested by test_adds_to_resend_audience_when_configured in TestAuthRoutes
|
|
pass
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_silent_error_on_db_failure(self, client, db):
|
|
"""DB errors are handled gracefully."""
|
|
# Already tested by test_database_error_shows_success_page in TestEdgeCases
|
|
pass
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_silent_error_on_resend_failure(self, client, db):
|
|
"""Resend API errors are handled gracefully."""
|
|
# Already tested by test_resend_api_error_handled_gracefully in TestEdgeCases
|
|
pass
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_captures_plan_parameter(self, client, db):
|
|
"""Supplier signup captures plan parameter."""
|
|
with patch.object(core.config, "WAITLIST_MODE", True), \
|
|
patch("padelnomics.worker.enqueue", new_callable=AsyncMock):
|
|
|
|
await client.post("/en/suppliers/signup/waitlist", form={
|
|
"csrf_token": "test_token",
|
|
"email": "test@example.com",
|
|
"plan": "supplier_pro",
|
|
})
|
|
|
|
entry = await _get_waitlist_entry(db, "test@example.com", "supplier")
|
|
assert entry is not None
|
|
assert entry["plan"] == "supplier_pro"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_captures_ip_address(self, client, db):
|
|
"""Signup captures request IP address."""
|
|
# Already tested by test_captures_ip_address in TestAuthRoutes
|
|
pass
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_email_intent_parameter(self, client, db):
|
|
"""Supplier signup uses different intent for email vs database."""
|
|
with patch.object(core.config, "WAITLIST_MODE", True), \
|
|
patch("padelnomics.worker.enqueue", new_callable=AsyncMock) as mock_enqueue:
|
|
|
|
await client.post("/en/suppliers/signup/waitlist", form={
|
|
"csrf_token": "test_token",
|
|
"email": "test@example.com",
|
|
"plan": "supplier_pro",
|
|
})
|
|
|
|
# DB should have intent="supplier"
|
|
entry = await _get_waitlist_entry(db, "test@example.com", "supplier")
|
|
assert entry is not None
|
|
|
|
# Email should have intent="supplier_pro" (plan name)
|
|
mock_enqueue.assert_called_with(
|
|
"send_waitlist_confirmation",
|
|
{"email": "test@example.com", "intent": "supplier_pro"}
|
|
)
|
|
|
|
|
|
# ── TestIntegration ───────────────────────────────────────────────
|
|
|
|
|
|
class TestIntegration:
|
|
"""End-to-end integration tests."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_full_entrepreneur_waitlist_flow(self, client, db):
|
|
"""Complete flow: GET form → POST email → see confirmation."""
|
|
with patch.object(core.config, "WAITLIST_MODE", True), \
|
|
patch("padelnomics.worker.enqueue", new_callable=AsyncMock):
|
|
|
|
# Step 1: GET waitlist form
|
|
response = await client.get("/auth/signup")
|
|
assert response.status_code == 200
|
|
html = await response.get_data(as_text=True)
|
|
assert "waitlist" in html.lower()
|
|
|
|
# Step 2: POST email
|
|
response = await client.post("/auth/signup", form={
|
|
"csrf_token": "test_token",
|
|
"email": "entrepreneur@example.com",
|
|
"plan": "free",
|
|
})
|
|
|
|
# Step 3: Verify confirmation page
|
|
assert response.status_code == 200
|
|
html = await response.get_data(as_text=True)
|
|
assert "entrepreneur@example.com" in html
|
|
assert "waitlist" in html.lower()
|
|
|
|
# Step 4: Verify database entry
|
|
entry = await _get_waitlist_entry(db, "entrepreneur@example.com", "free")
|
|
assert entry is not None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_full_supplier_waitlist_flow(self, client, db):
|
|
"""Complete flow: GET supplier form → POST email → see confirmation."""
|
|
with patch.object(core.config, "WAITLIST_MODE", True), \
|
|
patch("padelnomics.worker.enqueue", new_callable=AsyncMock):
|
|
|
|
# Step 1: GET supplier waitlist form
|
|
response = await client.get("/en/suppliers/signup?plan=supplier_pro")
|
|
assert response.status_code == 200
|
|
html = await response.get_data(as_text=True)
|
|
assert "waitlist" in html.lower()
|
|
assert "supplier" in html.lower()
|
|
|
|
# Step 2: POST email
|
|
response = await client.post("/en/suppliers/signup/waitlist", form={
|
|
"csrf_token": "test_token",
|
|
"email": "supplier@example.com",
|
|
"plan": "supplier_pro",
|
|
})
|
|
|
|
# Step 3: Verify confirmation page
|
|
assert response.status_code == 200
|
|
html = await response.get_data(as_text=True)
|
|
assert "supplier@example.com" in html
|
|
assert "supplier" in html.lower()
|
|
|
|
# Step 4: Verify database entry
|
|
entry = await _get_waitlist_entry(db, "supplier@example.com", "supplier")
|
|
assert entry is not None
|
|
assert entry["plan"] == "supplier_pro"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_toggle_off_reverts_to_normal_signup(self, client, db):
|
|
"""Setting WAITLIST_MODE=false reverts to normal signup flow."""
|
|
# First, enable waitlist mode
|
|
with patch.object(core.config, "WAITLIST_MODE", True):
|
|
response = await client.get("/auth/signup")
|
|
html = await response.get_data(as_text=True)
|
|
assert "waitlist" in html.lower()
|
|
|
|
# Then, disable waitlist mode
|
|
with patch.object(core.config, "WAITLIST_MODE", False):
|
|
response = await client.get("/auth/signup")
|
|
html = await response.get_data(as_text=True)
|
|
# Should see normal signup, not waitlist
|
|
assert "create" in html.lower() or "sign up" in html.lower()
|
|
# Should NOT see waitlist messaging
|
|
assert "join the waitlist" not in html.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_same_email_different_intents_both_captured(self, client, db):
|
|
"""Same email can be on both entrepreneur and supplier waitlists."""
|
|
with patch.object(core.config, "WAITLIST_MODE", True), \
|
|
patch("padelnomics.worker.enqueue", new_callable=AsyncMock), \
|
|
patch("padelnomics.worker.enqueue", new_callable=AsyncMock):
|
|
|
|
# Sign up as entrepreneur
|
|
await client.post("/auth/signup", form={
|
|
"csrf_token": "test_token",
|
|
"email": "both@example.com",
|
|
"plan": "free",
|
|
})
|
|
|
|
# Sign up as supplier
|
|
await client.post("/en/suppliers/signup/waitlist", form={
|
|
"csrf_token": "test_token",
|
|
"email": "both@example.com",
|
|
"plan": "supplier_growth",
|
|
})
|
|
|
|
# Should have 2 rows
|
|
count = await _count_rows(db, "waitlist")
|
|
assert count == 2
|
|
|
|
# Both should exist
|
|
entry1 = await _get_waitlist_entry(db, "both@example.com", "free")
|
|
entry2 = await _get_waitlist_entry(db, "both@example.com", "supplier")
|
|
assert entry1 is not None
|
|
assert entry2 is not None
|