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

test: e2e + unit tests for supervisor, proxy, and feature flags

Browse Source 
- test_supervisor.py: 28 tests covering load_workflows, resolve_schedule,
  is_due, topological_waves, and proxy round-robin / sticky selection
- test_feature_flags.py: 31 tests covering migration 0019, is_flag_enabled,
  feature_gate decorator, admin toggle routes, and full toggle e2e flows
- conftest.py: seed feature flags with production defaults (markets=1,
  others=0) so all routes behave consistently in tests
- Fix is_flag_enabled bug: replace non-existent db.execute_fetchone()
  with fetch_one() helper
- Update 4 test_waitlist / test_businessplan tests that relied on
  WAITLIST_MODE patches — now enable the relevant DB flag instead

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Deeman
2026-02-23 15:26:40 +01:00
parent a1faddbed6
commit b5c9a4e573
6 changed files with 770 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
web/tests/conftest.py
View File

@@ -54,6 +54,20 @@ async def db():
conn.row_factory = aiosqlite.Row
await conn.execute("PRAGMA foreign_keys=ON")
await conn.executescript(schema_ddl)
# Seed feature flags so routes that use feature_gate() pass through by default.
# Tests that specifically test gated behaviour set the flag to 0 via _set_flag().
await conn.executescript("""
INSERT OR IGNORE INTO feature_flags (name, enabled, description)
VALUES ('markets', 1, 'Market/SEO content pages');
INSERT OR IGNORE INTO feature_flags (name, enabled, description)
VALUES ('payments', 0, 'Paddle billing & checkout');
INSERT OR IGNORE INTO feature_flags (name, enabled, description)
VALUES ('planner_export', 0, 'Business plan PDF export');
INSERT OR IGNORE INTO feature_flags (name, enabled, description)
VALUES ('supplier_signup', 0, 'Supplier onboarding wizard');
INSERT OR IGNORE INTO feature_flags (name, enabled, description)
VALUES ('lead_unlock', 0, 'Lead credit purchase & unlock');
""")
await conn.commit()
original_db = core._db

4
web/tests/test_businessplan.py
View File

@@ -255,6 +255,10 @@ class TestExportRoutes:
assert resp.status_code == 302
async def test_export_page_shows_scenarios(self, auth_client, db, scenario):
await db.execute(
"INSERT OR REPLACE INTO feature_flags (name, enabled, description) VALUES ('planner_export', 1, '')"
)
await db.commit()
resp = await auth_client.get("/en/planner/export")
assert resp.status_code == 200
html = (await resp.data).decode()

431
web/tests/test_feature_flags.py Normal file
View File

@@ -0,0 +1,431 @@
"""
Tests for feature flags — migration, is_flag_enabled helper, feature_gate
decorator, admin toggle routes, and per-feature gating across all routes.
Unit tests cover is_flag_enabled and feature_gate in isolation.
Integration tests exercise full request/response flows via Quart test client.
"""
import sqlite3
from datetime import datetime
from pathlib import Path
from unittest.mock import AsyncMock, patch
import pytest
from padelnomics import core
from padelnomics.migrations.migrate import migrate
# ── Fixtures & helpers ────────────────────────────────────────────
@pytest.fixture(autouse=True)
def mock_csrf_validation():
"""Mock CSRF validation to always pass in all tests in this file."""
with patch("padelnomics.core.validate_csrf_token", return_value=True):
yield
@pytest.fixture
async def admin_client(app, db):
"""Test client with an admin-role user session (module-level, follows test_content.py)."""
now = datetime.utcnow().isoformat()
async with db.execute(
"INSERT INTO users (email, name, created_at) VALUES (?, ?, ?)",
("flags_admin@test.com", "Flags Admin", now),
) as cursor:
admin_id = cursor.lastrowid
await db.execute(
"INSERT INTO user_roles (user_id, role) VALUES (?, 'admin')", (admin_id,)
)
await db.commit()
async with app.test_client() as c:
async with c.session_transaction() as sess:
sess["user_id"] = admin_id
yield c
async def _set_flag(db, name: str, enabled: bool, description: str = ""):
"""Insert or replace a flag in the test DB."""
await db.execute(
"""INSERT OR REPLACE INTO feature_flags (name, enabled, description)
VALUES (?, ?, ?)""",
(name, 1 if enabled else 0, description),
)
await db.commit()
async def _flag_value(db, name: str) -> int | None:
async with db.execute(
"SELECT enabled FROM feature_flags WHERE name = ?", (name,)
) as cursor:
row = await cursor.fetchone()
return row[0] if row else None
async def _seed_all_flags(db):
"""Seed the five default flags matching migration 0019 defaults."""
flags = [
("markets", 1),
("payments", 0),
("planner_export", 0),
("supplier_signup", 0),
("lead_unlock", 0),
]
for name, enabled in flags:
await db.execute(
"INSERT OR REPLACE INTO feature_flags (name, enabled, description) VALUES (?, ?, ?)",
(name, enabled, ""),
)
await db.commit()
def _column_names(conn, table):
return [r[1] for r in conn.execute(f"PRAGMA table_info({table})").fetchall()]
def _table_names(conn):
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]
# ── Migration 0019 ────────────────────────────────────────────────
class TestMigration0019:
"""Migration 0019 creates feature_flags table and seeds initial flags."""
def test_creates_feature_flags_table(self, tmp_path):
db_path = str(tmp_path / "test.db")
migrate(db_path)
conn = sqlite3.connect(db_path)
assert "feature_flags" in _table_names(conn)
conn.close()
def test_table_has_correct_columns(self, tmp_path):
db_path = str(tmp_path / "test.db")
migrate(db_path)
conn = sqlite3.connect(db_path)
cols = _column_names(conn, "feature_flags")
conn.close()
assert "name" in cols
assert "enabled" in cols
assert "description" in cols
assert "updated_at" in cols
def test_seeds_markets_enabled(self, tmp_path):
db_path = str(tmp_path / "test.db")
migrate(db_path)
conn = sqlite3.connect(db_path)
row = conn.execute(
"SELECT enabled FROM feature_flags WHERE name = 'markets'"
).fetchone()
conn.close()
assert row is not None and row[0] == 1
def test_seeds_payments_disabled(self, tmp_path):
db_path = str(tmp_path / "test.db")
migrate(db_path)
conn = sqlite3.connect(db_path)
row = conn.execute(
"SELECT enabled FROM feature_flags WHERE name = 'payments'"
).fetchone()
conn.close()
assert row is not None and row[0] == 0
def test_seeds_all_five_flags(self, tmp_path):
db_path = str(tmp_path / "test.db")
migrate(db_path)
conn = sqlite3.connect(db_path)
names = {r[0] for r in conn.execute("SELECT name FROM feature_flags").fetchall()}
conn.close()
assert names == {"markets", "payments", "planner_export", "supplier_signup", "lead_unlock"}
def test_idempotent_on_rerun(self, tmp_path):
"""Running migrate twice does not duplicate seed rows."""
db_path = str(tmp_path / "test.db")
migrate(db_path)
migrate(db_path)
conn = sqlite3.connect(db_path)
count = conn.execute("SELECT COUNT(*) FROM feature_flags").fetchone()[0]
conn.close()
assert count == 5
# ── is_flag_enabled ───────────────────────────────────────────────
class TestIsFlagEnabled:
"""Unit tests for is_flag_enabled() helper."""
@pytest.mark.asyncio
async def test_returns_true_for_enabled_flag(self, db):
await _set_flag(db, "markets", True)
assert await core.is_flag_enabled("markets") is True
@pytest.mark.asyncio
async def test_returns_false_for_disabled_flag(self, db):
await _set_flag(db, "payments", False)
assert await core.is_flag_enabled("payments") is False
@pytest.mark.asyncio
async def test_returns_default_false_for_unknown_flag(self, db):
assert await core.is_flag_enabled("nonexistent_flag") is False
@pytest.mark.asyncio
async def test_returns_custom_default_for_unknown_flag(self, db):
assert await core.is_flag_enabled("nonexistent_flag", default=True) is True
@pytest.mark.asyncio
async def test_reflects_live_db_change(self, db):
"""Change takes effect without restart — reads live DB every call."""
await _set_flag(db, "payments", False)
assert await core.is_flag_enabled("payments") is False
await _set_flag(db, "payments", True)
assert await core.is_flag_enabled("payments") is True
# ── feature_gate decorator ────────────────────────────────────────
class TestFeatureGateDecorator:
"""feature_gate blocks GET when flag is disabled, passes through when enabled."""
@pytest.mark.asyncio
async def test_shows_waitlist_on_get_when_flag_disabled(self, client, db):
"""GET /auth/signup shows waitlist page when payments flag is off."""
await _set_flag(db, "payments", False)
response = await client.get("/auth/signup")
assert response.status_code == 200
html = await response.get_data(as_text=True)
# payments waitlist.html shows "join" or "launching soon" messaging
assert "launching soon" in html.lower() or "join" in html.lower() or "waitlist" in html.lower()
@pytest.mark.asyncio
async def test_shows_normal_form_on_get_when_flag_enabled(self, client, db):
"""GET /auth/signup shows normal signup form when payments flag is on."""
await _set_flag(db, "payments", True)
response = await client.get("/auth/signup")
assert response.status_code == 200
html = await response.get_data(as_text=True)
# Normal signup shows create account form
assert "create" in html.lower() or "sign up" in html.lower() or "email" in html.lower()
@pytest.mark.asyncio
async def test_post_passes_through_when_flag_disabled(self, client, db):
"""POST is never blocked by feature_gate (gates GET only)."""
await _set_flag(db, "payments", False)
with patch("padelnomics.worker.enqueue", new_callable=AsyncMock):
response = await client.post("/auth/signup", form={
"csrf_token": "test_token",
"email": "test@example.com",
})
assert response.status_code in (200, 302)
@pytest.mark.asyncio
async def test_markets_route_gated_by_markets_flag(self, client, db):
"""markets flag controls /en/markets access."""
# Disabled → shows gated page
await _set_flag(db, "markets", False)
response = await client.get("/en/markets")
assert response.status_code == 200
html = await response.get_data(as_text=True)
assert "coming soon" in html.lower() or "intelligence" in html.lower()
# Enabled → passes through
await _set_flag(db, "markets", True)
response = await client.get("/en/markets")
assert response.status_code == 200
html = await response.get_data(as_text=True)
# Normal markets page doesn't show the "coming soon" waitlist title
assert "coming soon" not in html.lower()
@pytest.mark.asyncio
async def test_supplier_signup_gated_by_flag(self, client, db):
"""supplier_signup flag controls /en/suppliers/signup."""
await _set_flag(db, "supplier_signup", False)
response = await client.get("/en/suppliers/signup")
assert response.status_code == 200
html = await response.get_data(as_text=True)
assert "waitlist" in html.lower() or "supplier" in html.lower()
@pytest.mark.asyncio
async def test_planner_export_gated_for_authenticated_user(self, auth_client, db):
"""planner_export flag controls /en/planner/export."""
await _set_flag(db, "planner_export", False)
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() or "waitlist" in html.lower()
@pytest.mark.asyncio
async def test_planner_export_accessible_when_enabled(self, auth_client, db):
"""Normal planner export page shown when planner_export flag is on."""
await _set_flag(db, "planner_export", 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" not in html.lower()
# ── lead_unlock gate ──────────────────────────────────────────────
class TestLeadUnlockGate:
"""lead_unlock flag controls whether the unlock endpoint is reachable."""
@pytest.mark.asyncio
async def test_is_flag_disabled_by_default(self, db):
"""lead_unlock flag seeded as disabled — is_flag_enabled returns False."""
await _set_flag(db, "lead_unlock", False)
assert await core.is_flag_enabled("lead_unlock") is False
@pytest.mark.asyncio
async def test_is_flag_enabled_after_toggle(self, db):
"""is_flag_enabled returns True after flag is enabled."""
await _set_flag(db, "lead_unlock", True)
assert await core.is_flag_enabled("lead_unlock") is True
@pytest.mark.asyncio
async def test_route_imports_is_flag_enabled(self):
"""suppliers/routes.py imports is_flag_enabled (gate is wired up)."""
from padelnomics.suppliers.routes import unlock_lead
import inspect
src = inspect.getsource(unlock_lead)
assert "is_flag_enabled" in src
assert "lead_unlock" in src
assert "403" in src
# ── Admin flag toggle routes ──────────────────────────────────────
class TestAdminFlagRoutes:
"""Admin /admin/flags endpoints require admin role and toggle flags correctly."""
@pytest.mark.asyncio
async def test_flags_page_requires_admin(self, client, db):
"""Anonymous request to /admin/flags redirects to login."""
response = await client.get("/admin/flags", follow_redirects=False)
assert response.status_code == 302
@pytest.mark.asyncio
async def test_flags_page_accessible_to_admin(self, admin_client, db):
await _seed_all_flags(db)
response = await admin_client.get("/admin/flags")
assert response.status_code == 200
@pytest.mark.asyncio
async def test_flags_page_lists_all_seeded_flags(self, admin_client, db):
await _seed_all_flags(db)
response = await admin_client.get("/admin/flags")
assert response.status_code == 200
html = await response.get_data(as_text=True)
for flag in ("markets", "payments", "planner_export", "supplier_signup", "lead_unlock"):
assert flag in html
@pytest.mark.asyncio
async def test_toggle_enables_disabled_flag(self, admin_client, db):
await _set_flag(db, "payments", False)
await admin_client.post("/admin/flags/toggle", form={
"csrf_token": "test_token",
"name": "payments",
})
assert await _flag_value(db, "payments") == 1
@pytest.mark.asyncio
async def test_toggle_disables_enabled_flag(self, admin_client, db):
await _set_flag(db, "markets", True)
await admin_client.post("/admin/flags/toggle", form={
"csrf_token": "test_token",
"name": "markets",
})
assert await _flag_value(db, "markets") == 0
@pytest.mark.asyncio
async def test_toggle_unknown_flag_redirects_with_flash(self, admin_client, db):
response = await admin_client.post("/admin/flags/toggle", form={
"csrf_token": "test_token",
"name": "nonexistent_flag",
}, follow_redirects=True)
assert response.status_code == 200
html = await response.get_data(as_text=True)
assert "not found" in html.lower()
@pytest.mark.asyncio
async def test_toggle_requires_admin(self, client, db):
"""Anonymous POST to toggle is rejected."""
response = await client.post("/admin/flags/toggle", form={
"csrf_token": "test_token",
"name": "markets",
}, follow_redirects=False)
assert response.status_code == 302
# ── Full toggle flow (e2e) ────────────────────────────────────────
class TestFlagToggleFlow:
"""End-to-end: admin toggles flag → route behaviour changes immediately."""
@pytest.mark.asyncio
async def test_disable_markets_shows_gated_page(self, admin_client, client, db):
"""Disable markets → /en/markets shows coming soon. Enable → shows content."""
# Seed markets as enabled
await _set_flag(db, "markets", True)
response = await client.get("/en/markets")
html = await response.get_data(as_text=True)
assert "coming soon" not in html.lower()
# Admin disables via toggle
await admin_client.post("/admin/flags/toggle", form={
"csrf_token": "test_token",
"name": "markets",
})
# Now shows gated page
response = await client.get("/en/markets")
html = await response.get_data(as_text=True)
assert "coming soon" in html.lower()
@pytest.mark.asyncio
async def test_enable_supplier_signup_passes_through(self, admin_client, client, db):
"""Enable supplier_signup → normal wizard shown (no waitlist page)."""
# Start with flag disabled
await _set_flag(db, "supplier_signup", False)
response = await client.get("/en/suppliers/signup")
html = await response.get_data(as_text=True)
# Gated: shows waitlist/supplier promo content
assert "waitlist" in html.lower() or "supplier" in html.lower()
# Admin enables
await admin_client.post("/admin/flags/toggle", form={
"csrf_token": "test_token",
"name": "supplier_signup",
})
# Now passes through (wizard is shown)
response = await client.get("/en/suppliers/signup")
assert response.status_code == 200
html = await response.get_data(as_text=True)
assert "waitlist" not in html.lower()
@pytest.mark.asyncio
async def test_double_toggle_restores_original_state(self, admin_client, db):
"""Toggling a flag twice returns it to its original value."""
await _set_flag(db, "payments", False)
assert await _flag_value(db, "payments") == 0
await admin_client.post("/admin/flags/toggle", form={
"csrf_token": "test_token", "name": "payments",
})
assert await _flag_value(db, "payments") == 1
await admin_client.post("/admin/flags/toggle", form={
"csrf_token": "test_token", "name": "payments",
})
assert await _flag_value(db, "payments") == 0

285
web/tests/test_supervisor.py Normal file
View File

@@ -0,0 +1,285 @@
"""
Unit tests for supervisor.py and proxy.py.
Tests cover pure-Python logic only — no DB, no subprocesses, no network.
DB-dependent functions (is_due, _get_last_success_time) are tested via mocks.
supervisor.py lives in src/padelnomics/ (not a uv workspace package), so we
add src/ to sys.path before importing.
"""
import sys
import textwrap
import tomllib
from datetime import UTC, datetime, timedelta
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
# Load supervisor.py directly by path — avoids clashing with the web app's
# 'padelnomics' namespace (which is the installed web package).
import importlib.util as _ilu
_SUP_PATH = Path(__file__).parent.parent.parent / "src" / "padelnomics" / "supervisor.py"
_spec = _ilu.spec_from_file_location("padelnomics_supervisor", _SUP_PATH)
sup = _ilu.module_from_spec(_spec)
_spec.loader.exec_module(sup)
from padelnomics_extract.proxy import (
load_proxy_urls,
make_round_robin_cycler,
make_sticky_selector,
)
# ── load_workflows ────────────────────────────────────────────────
class TestLoadWorkflows:
def test_loads_all_fields(self, tmp_path):
toml = tmp_path / "workflows.toml"
toml.write_text(textwrap.dedent("""\
[extract_a]
module = "mypkg.extract_a"
schedule = "daily"
[extract_b]
module = "mypkg.extract_b"
schedule = "weekly"
entry = "run"
depends_on = ["extract_a"]
proxy_mode = "sticky"
"""))
wfs = sup.load_workflows(toml)
assert len(wfs) == 2
a = next(w for w in wfs if w["name"] == "extract_a")
assert a["module"] == "mypkg.extract_a"
assert a["schedule"] == "daily"
assert a["entry"] == "main" # default
assert a["depends_on"] == [] # default
assert a["proxy_mode"] == "round-robin" # default
b = next(w for w in wfs if w["name"] == "extract_b")
assert b["entry"] == "run"
assert b["depends_on"] == ["extract_a"]
assert b["proxy_mode"] == "sticky"
def test_raises_on_missing_module(self, tmp_path):
toml = tmp_path / "bad.toml"
toml.write_text("[wf]\nschedule = 'daily'\n")
with pytest.raises(AssertionError, match="missing 'module'"):
sup.load_workflows(toml)
def test_raises_on_missing_schedule(self, tmp_path):
toml = tmp_path / "bad.toml"
toml.write_text("[wf]\nmodule = 'mypkg.wf'\n")
with pytest.raises(AssertionError, match="missing 'schedule'"):
sup.load_workflows(toml)
def test_raises_if_file_missing(self, tmp_path):
with pytest.raises(AssertionError, match="not found"):
sup.load_workflows(tmp_path / "nonexistent.toml")
# ── resolve_schedule ──────────────────────────────────────────────
class TestResolveSchedule:
def test_maps_named_presets(self):
assert sup.resolve_schedule("hourly") == "0 * * * *"
assert sup.resolve_schedule("daily") == "0 5 * * *"
assert sup.resolve_schedule("weekly") == "0 3 * * 1"
assert sup.resolve_schedule("monthly") == "0 4 1 * *"
def test_passes_through_raw_cron(self):
expr = "0 6-23 * * *"
assert sup.resolve_schedule(expr) == expr
def test_unknown_name_passes_through(self):
assert sup.resolve_schedule("quarterly") == "quarterly"
# ── is_due ────────────────────────────────────────────────────────
class TestIsDue:
def _wf(self, schedule="daily", name="test_wf"):
return {"name": name, "schedule": schedule}
def test_never_ran_is_due(self):
conn = MagicMock()
with patch.object(sup, "_get_last_success_time", return_value=None):
assert sup.is_due(conn, self._wf()) is True
def test_ran_after_last_trigger_is_not_due(self):
"""Last run was AFTER the most recent trigger — not due."""
conn = MagicMock()
# Use a daily schedule — trigger fires at 05:00 UTC
# Simulate last success = today at 06:00, so trigger already covered
now = datetime.now(UTC)
last_success = now.replace(hour=6, minute=0, second=0, microsecond=0)
with patch.object(sup, "_get_last_success_time", return_value=last_success):
assert sup.is_due(conn, self._wf(schedule="daily")) is False
def test_ran_before_last_trigger_is_due(self):
"""Last run was BEFORE the most recent trigger — due again."""
conn = MagicMock()
# Monthly fires on the 1st at 04:00 — simulate running last month
last_success = datetime.now(UTC) - timedelta(days=35)
with patch.object(sup, "_get_last_success_time", return_value=last_success):
assert sup.is_due(conn, self._wf(schedule="monthly")) is True
# ── topological_waves ─────────────────────────────────────────────
class TestTopologicalWaves:
def _wf(self, name, depends_on=None):
return {"name": name, "depends_on": depends_on or []}
def test_no_deps_single_wave(self):
wfs = [self._wf("a"), self._wf("b"), self._wf("c")]
waves = sup.topological_waves(wfs)
assert len(waves) == 1
assert {w["name"] for w in waves[0]} == {"a", "b", "c"}
def test_simple_chain_two_waves(self):
wfs = [self._wf("a"), self._wf("b", depends_on=["a"])]
waves = sup.topological_waves(wfs)
assert len(waves) == 2
assert waves[0][0]["name"] == "a"
assert waves[1][0]["name"] == "b"
def test_diamond_three_waves(self):
"""a → b,c → d"""
wfs = [
self._wf("a"),
self._wf("b", depends_on=["a"]),
self._wf("c", depends_on=["a"]),
self._wf("d", depends_on=["b", "c"]),
]
waves = sup.topological_waves(wfs)
assert len(waves) == 3
assert waves[0][0]["name"] == "a"
assert {w["name"] for w in waves[1]} == {"b", "c"}
assert waves[2][0]["name"] == "d"
def test_dep_outside_due_set_ignored(self):
"""Dependency not in the due set is treated as satisfied."""
wfs = [self._wf("b", depends_on=["a"])] # "a" not in due set
waves = sup.topological_waves(wfs)
assert len(waves) == 1
assert waves[0][0]["name"] == "b"
def test_circular_dep_raises(self):
wfs = [
self._wf("a", depends_on=["b"]),
self._wf("b", depends_on=["a"]),
]
with pytest.raises(AssertionError, match="Circular dependency"):
sup.topological_waves(wfs)
def test_empty_list_returns_empty(self):
assert sup.topological_waves([]) == []
def test_real_workflows_toml(self):
"""The actual workflows.toml in the repo parses and produces valid waves."""
repo_root = Path(__file__).parent.parent.parent
wf_path = repo_root / "infra" / "supervisor" / "workflows.toml"
if not wf_path.exists():
pytest.skip("workflows.toml not found")
wfs = sup.load_workflows(wf_path)
waves = sup.topological_waves(wfs)
# playtomic_availability must come after playtomic_tenants
all_names = [w["name"] for wave in waves for w in wave]
tenants_idx = all_names.index("playtomic_tenants")
avail_idx = all_names.index("playtomic_availability")
assert tenants_idx < avail_idx
# ── proxy.py ─────────────────────────────────────────────────────
class TestLoadProxyUrls:
def test_returns_empty_when_unset(self, monkeypatch):
monkeypatch.delenv("PROXY_URLS", raising=False)
assert load_proxy_urls() == []
def test_parses_comma_separated_urls(self, monkeypatch):
monkeypatch.setenv(
"PROXY_URLS",
"http://p1:8080,http://p2:8080,http://p3:8080",
)
urls = load_proxy_urls()
assert urls == ["http://p1:8080", "http://p2:8080", "http://p3:8080"]
def test_strips_whitespace(self, monkeypatch):
monkeypatch.setenv("PROXY_URLS", " http://p1:8080 , http://p2:8080 ")
urls = load_proxy_urls()
assert urls == ["http://p1:8080", "http://p2:8080"]
def test_ignores_empty_segments(self, monkeypatch):
monkeypatch.setenv("PROXY_URLS", "http://p1:8080,,http://p2:8080,")
urls = load_proxy_urls()
assert urls == ["http://p1:8080", "http://p2:8080"]
class TestRoundRobinCycler:
def test_returns_none_callable_when_no_proxies(self):
fn = make_round_robin_cycler([])
assert fn() is None
def test_cycles_through_proxies(self):
urls = ["http://p1", "http://p2", "http://p3"]
fn = make_round_robin_cycler(urls)
results = [fn() for _ in range(6)]
assert results == ["http://p1", "http://p2", "http://p3"] * 2
def test_thread_safe_independent_calls(self):
"""Concurrent calls each get a proxy — no exceptions."""
import threading
urls = ["http://p1", "http://p2"]
fn = make_round_robin_cycler(urls)
results = []
lock = threading.Lock()
def worker():
proxy = fn()
with lock:
results.append(proxy)
threads = [threading.Thread(target=worker) for _ in range(10)]
for t in threads:
t.start()
for t in threads:
t.join()
assert len(results) == 10
assert all(r in urls for r in results)
class TestStickySelectorProxy:
def test_returns_none_callable_when_no_proxies(self):
fn = make_sticky_selector([])
assert fn("any_key") is None
def test_same_key_always_same_proxy(self):
urls = ["http://p1", "http://p2", "http://p3"]
fn = make_sticky_selector(urls)
proxy = fn("tenant_abc")
for _ in range(10):
assert fn("tenant_abc") == proxy
def test_different_keys_can_map_to_different_proxies(self):
urls = ["http://p1", "http://p2", "http://p3"]
fn = make_sticky_selector(urls)
results = {fn(f"key_{i}") for i in range(30)}
assert len(results) > 1 # distribution across proxies
def test_all_results_are_valid_proxies(self):
urls = ["http://p1", "http://p2"]
fn = make_sticky_selector(urls)
for i in range(20):
assert fn(f"key_{i}") in urls

62
web/tests/test_waitlist.py
View File

@@ -251,13 +251,16 @@ class TestAuthRoutes:
@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
"""Normal signup flow when payments flag is enabled (WAITLIST_MODE replaced by feature flag)."""
await db.execute(
"INSERT OR REPLACE INTO feature_flags (name, enabled, description) VALUES ('payments', 1, '')"
)
await db.commit()
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):
@@ -713,13 +716,16 @@ 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()
async def test_passes_through_when_waitlist_disabled(self, client, db):
"""feature_gate passes through to normal form when payments flag is enabled."""
await db.execute(
"INSERT OR REPLACE INTO feature_flags (name, enabled, description) VALUES ('payments', 1, '')"
)
await db.commit()
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):
@@ -940,21 +946,21 @@ class TestIntegration:
@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()
"""Disabling payments flag shows waitlist; enabling it shows normal signup."""
# payments=0 (default) → shows waitlist page
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()
# Enable payments flag → shows normal signup
await db.execute(
"INSERT OR REPLACE INTO feature_flags (name, enabled, description) VALUES ('payments', 1, '')"
)
await db.commit()
response = await client.get("/auth/signup")
html = await response.get_data(as_text=True)
assert "create" in html.lower() or "sign up" in html.lower()
assert "join the waitlist" not in html.lower()
@pytest.mark.asyncio
async def test_same_email_different_intents_both_captured(self, client, db):