#!/usr/bin/env python3
"""Auth route tests (Phase 2 — 2a)
Tests for authentication routes:
- POST /auth/register — success, duplicate email, missing fields
- POST /auth/login — success, wrong password, already logged in
- POST /auth/logout — clears session
- Rate limiting on /auth/login (10 req/min)
- Magic login — request, consume, disabled
- Dashboard requires login
Uses shared flask_app from app.app (same pattern as test_phase2.py).
Test data is cleaned up after each test via _cleanup_test_data fixture.
"""
import json
import os
import random
import sqlite3
import sys
import pytest
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
# Use shared test DB (singleton pattern — only initializes once)
import tests.test_shared_db # noqa: F401
# Set required env vars BEFORE importing app modules
os.environ.setdefault("AGENTFORMS_SECRET_KEY", "auth-test-secret")
os.environ.setdefault("SMTP_HOST", "smtp.test.invalid")
os.environ.setdefault("SMTP_PORT", "587")
os.environ.setdefault("SMTP_USER", "test@test.invalid")
os.environ.setdefault("SMTP_PASS", "testpass")
os.environ.setdefault("ENCRYPTION_KEY", "5c321ae59453d7bae2e05a9e19e97347411a033d9858a0456d5cc63acfcb0369")
# Import shared Flask app — same instance used by all test files
from app.app import app as flask_app
# Ensure test mode
flask_app.config["TESTING"] = True
flask_app.config["SESSION_COOKIE_SECURE"] = False
# Disable rate limiting for tests — prevents 429 errors from auth endpoint limit (10 req/min)
for _key in list(flask_app.before_request_funcs.get(None, []) or []):
if getattr(_key, "__name__", None) == "rate_limit_check":
flask_app.before_request_funcs[None].remove(_key)
break
@pytest.fixture(autouse=True)
def _cleanup_test_data():
"""Auto-remove test users created during each test run.
Must delete in FK-order: submissions → sites → users
to avoid constraint violations when other test files also clean up.
"""
yield
# After each test, clean up test users and their data
try:
import app.models
conn = sqlite3.connect(app.models.DB_PATH)
test_patterns = [
"%auth-test%@example.com",
"%auth-login%@example.com",
"%auth-logout%@example.com",
"%auth-dashboard%@example.com",
"%auth-magic%@example.com",
"%auth-register%@example.com",
"%auth-user%@example.com",
]
where_clause = " OR ".join(["email LIKE ?"] * len(test_patterns))
# Get user IDs that match our test patterns
user_ids = [row[0] for row in conn.execute(
f"SELECT id FROM users WHERE {where_clause}", test_patterns
).fetchall()]
if user_ids:
# Build proper IN clause with individual ? placeholders
placeholders = ",".join(["?"] * len(user_ids))
ids = list(user_ids)
# FK-order: clean up dependent tables first, then users
# FK chain: submissions.site_id → sites.id; sites.user_id → users.id
# Delete webhook logs (reference submissions)
conn.execute(
f"DELETE FROM webhook_logs WHERE submission_id IN ("
f"SELECT id FROM submissions WHERE site_id IN ("
f"SELECT id FROM sites WHERE user_id IN ({placeholders})))", ids
)
# Delete submissions (reference sites)
conn.execute(
f"DELETE FROM submissions WHERE site_id IN ("
f"SELECT id FROM sites WHERE user_id IN ({placeholders}))", ids
)
# Delete form-related tables (reference sites)
for table in ["form_impressions", "form_sessions", "form_versions",
"form_variants", "form_actions", "ai_prompt_history",
"site_analytics", "webhook_destinations", "document_templates",
"documents", "invoice_schedules"]:
conn.execute(
f"DELETE FROM {table} WHERE site_id IN ("
f"SELECT id FROM sites WHERE user_id IN ({placeholders}))", ids
)
# Delete campaigns and related tables
conn.execute(
f"DELETE FROM campaign_recipients WHERE campaign_id IN ("
f"SELECT id FROM campaigns WHERE user_id IN ({placeholders}))", ids
)
conn.execute(
f"DELETE FROM campaign_ab_variants WHERE campaign_id IN ("
f"SELECT id FROM campaigns WHERE user_id IN ({placeholders}))", ids
)
conn.execute(
f"DELETE FROM campaign_reminders WHERE campaign_id IN ("
f"SELECT id FROM campaigns WHERE user_id IN ({placeholders}))", ids
)
conn.execute(
f"DELETE FROM campaigns WHERE user_id IN ({placeholders})", ids
)
# Delete user-related tables
conn.execute(f"DELETE FROM email_send_log WHERE user_id IN ({placeholders})", ids)
conn.execute(f"DELETE FROM custom_templates WHERE user_id IN ({placeholders})", ids)
conn.execute(f"DELETE FROM email_settings WHERE user_id IN ({placeholders})", ids)
conn.execute(f"DELETE FROM monthly_usage WHERE user_id IN ({placeholders})", ids)
conn.execute(f"DELETE FROM api_keys WHERE user_id IN ({placeholders})", ids)
# Delete sites (no more FK deps)
conn.execute(
f"DELETE FROM sites WHERE user_id IN ({placeholders})", ids
)
# Delete team membership (but not teams owned by others)
conn.execute(
f"DELETE FROM team_members WHERE user_id IN ({placeholders})", ids
)
# Finally delete users
conn.execute(f"DELETE FROM users WHERE {where_clause}", test_patterns)
conn.commit()
conn.close()
except Exception:
pass
class TestRegistration:
"""Test POST /auth/register."""
@pytest.fixture
def client(self):
return flask_app.test_client()
def _random_email(self):
return f"auth-test{random.randint(10000, 99999)}@example.com"
def test_register_success(self, client):
email = self._random_email()
resp = client.post("/auth/register", data={
"email": email,
"password": "SecurePass123",
"name": "Test User",
}, follow_redirects=True)
assert resp.status_code == 200
with client.session_transaction() as sess:
assert "user_id" in sess
def test_register_duplicate_email(self, client):
email = self._random_email()
client.post("/auth/register", data={
"email": email,
"password": "SecurePass123",
"name": "First User",
}, follow_redirects=True)
# Logout using JSON to bypass CSRF
client.post("/auth/logout", json={}, content_type="application/json")
resp = client.post("/auth/register", data={
"email": email,
"password": "AnotherPass123",
"name": "Duplicate User",
})
assert resp.status_code == 409
def test_register_missing_email(self, client):
resp = client.post("/auth/register", data={
"email": "",
"password": "SecurePass123",
"name": "No Email",
})
assert resp.status_code == 400
def test_register_short_password(self, client):
resp = client.post("/auth/register", data={
"email": self._random_email(),
"password": "short",
"name": "Short Pass",
})
assert resp.status_code == 400
def test_register_invalid_email_format(self, client):
resp = client.post("/auth/register", data={
"email": "not-an-email",
"password": "SecurePass123",
"name": "Bad Email",
})
assert resp.status_code == 400
def test_register_name_too_long(self, client):
resp = client.post("/auth/register", data={
"email": self._random_email(),
"password": "SecurePass123",
"name": "A" * 101,
})
assert resp.status_code == 400
def test_register_redirects_if_logged_in(self, client):
email = self._random_email()
client.post("/auth/register", data={
"email": email,
"password": "SecurePass123",
"name": "User",
})
# Already logged in — should redirect
resp = client.post("/auth/register", data={
"email": "other@example.com",
"password": "SecurePass123",
"name": "Other",
}, follow_redirects=False)
assert resp.status_code == 302
def test_register_get_renders_form(self, client):
resp = client.get("/auth/register")
assert resp.status_code == 200
class TestLogin:
"""Test POST /auth/login."""
@pytest.fixture
def client(self):
return flask_app.test_client()
def _random_email(self):
return f"auth-login{random.randint(10000, 99999)}@example.com"
def test_login_success(self, client):
email = self._random_email()
# Register then logout to get a clean session
client.post("/auth/register", data={
"email": email,
"password": "SecurePass123",
"name": "Login User",
}, follow_redirects=True)
client.post("/auth/logout", json={}, content_type="application/json")
resp = client.post("/auth/login", data={
"email": email,
"password": "SecurePass123",
}, follow_redirects=True)
assert resp.status_code == 200
with client.session_transaction() as sess:
assert "user_id" in sess
def test_login_wrong_password(self, client):
email = self._random_email()
client.post("/auth/register", data={
"email": email,
"password": "SecurePass123",
"name": "Wrong Pass User",
}, follow_redirects=True)
client.post("/auth/logout", json={}, content_type="application/json")
resp = client.post("/auth/login", data={
"email": email,
"password": "WrongPassword123",
})
assert resp.status_code == 401
def test_login_nonexistent_user(self, client):
resp = client.post("/auth/login", data={
"email": "nonexistent@example.com",
"password": "AnyPassword123",
})
assert resp.status_code == 401
def test_login_missing_fields(self, client):
resp = client.post("/auth/login", data={
"email": "",
"password": "",
})
assert resp.status_code == 400
def test_login_redirects_if_logged_in(self, client):
email = self._random_email()
client.post("/auth/register", data={
"email": email,
"password": "SecurePass123",
"name": "User",
})
resp = client.post("/auth/login", data={
"email": email,
"password": "SecurePass123",
}, follow_redirects=False)
assert resp.status_code == 302
def test_login_open_redirect_prevented(self, client):
email = self._random_email()
client.post("/auth/register", data={
"email": email,
"password": "SecurePass123",
"name": "User",
}, follow_redirects=True)
client.post("/auth/logout", json={}, content_type="application/json")
resp = client.post("/auth/login?next=https://evil.com", data={
"email": email,
"password": "SecurePass123",
}, follow_redirects=False)
# Should not redirect to external URL
location = resp.headers.get("Location", "")
assert "evil.com" not in location
def test_login_remember_me(self, client):
email = self._random_email()
client.post("/auth/register", data={
"email": email,
"password": "SecurePass123",
"name": "Remember User",
}, follow_redirects=True)
client.post("/auth/logout", json={}, content_type="application/json")
resp = client.post("/auth/login", data={
"email": email,
"password": "SecurePass123",
"remember": "on",
}, follow_redirects=True)
assert resp.status_code == 200
def test_login_get_renders_form(self, client):
resp = client.get("/auth/login")
assert resp.status_code == 200
class TestLogout:
"""Test POST /auth/logout."""
@pytest.fixture
def client(self):
return flask_app.test_client()
def _random_email(self):
return f"auth-logout{random.randint(10000, 99999)}@example.com"
def test_logout_clears_session(self, client):
email = self._random_email()
client.post("/auth/register", data={
"email": email,
"password": "SecurePass123",
"name": "Logout User",
}, follow_redirects=True)
# Use JSON to bypass CSRF (authenticated sessions skip CSRF for JSON)
resp = client.post("/auth/logout", json={}, content_type="application/json",
follow_redirects=False)
assert resp.status_code == 302
with client.session_transaction() as sess:
assert "user_id" not in sess
class TestDashboard:
"""Test /auth/dashboard access control."""
@pytest.fixture
def client(self):
return flask_app.test_client()
def test_dashboard_requires_login(self, client):
resp = client.get("/auth/dashboard", follow_redirects=False)
assert resp.status_code in (302, 403)
def test_dashboard_accessible_when_logged_in(self, client):
email = f"auth-dashboard{random.randint(10000, 99999)}@example.com"
client.post("/auth/register", data={
"email": email,
"password": "SecurePass123",
"name": "Dashboard User",
}, follow_redirects=True)
resp = client.get("/auth/dashboard")
assert resp.status_code == 200
class TestMagicLogin:
"""Test magic login flow."""
@pytest.fixture
def client(self):
return flask_app.test_client()
def _random_email(self):
return f"auth-magic{random.randint(10000, 99999)}@example.com"
def test_magic_login_request_json(self, client):
email = self._random_email()
client.post("/auth/register", data={
"email": email,
"password": "SecurePass123",
"name": "Magic User",
}, follow_redirects=True)
client.post("/auth/logout", json={}, content_type="application/json")
resp = client.post("/auth/magic", json={"email": email},
content_type="application/json")
data = resp.get_json()
# Non-admin emails get success message (always returns 200 to prevent enumeration)
# or 403 if magic login disabled for account
assert resp.status_code in (200, 403)
assert data is not None
def test_magic_login_request_invalid_email(self, client):
resp = client.post("/auth/magic", json={"email": "not-an-email"},
content_type="application/json")
assert resp.status_code == 400
def test_magic_login_request_no_email(self, client):
resp = client.post("/auth/magic", json={}, content_type="application/json")
assert resp.status_code == 400
def test_magic_login_disabled_account(self, client):
email = self._random_email()
client.post("/auth/register", data={
"email": email,
"password": "SecurePass123",
"name": "No Magic User",
}, follow_redirects=True)
# Disable magic login via JSON (authenticated session, JSON = no CSRF)
resp = client.post("/settings/security/magic-login",
json={"enabled": False},
content_type="application/json")
assert resp.status_code == 200
client.post("/auth/logout", json={}, content_type="application/json")
resp = client.post("/auth/magic", json={"email": email},
content_type="application/json")
# Disabled account should get 403
assert resp.status_code == 403
def test_magic_login_consume_valid_token(self, client):
email = self._random_email()
client.post("/auth/register", data={
"email": email,
"password": "SecurePass123",
"name": "Magic Consume User",
}, follow_redirects=True)
client.post("/auth/logout", json={}, content_type="application/json")
# Generate and consume token directly
from app.models import generate_magic_login_token, validate_magic_login_token
token = generate_magic_login_token(email)
assert token is not None
resp = client.get(f"/auth/magic/{token}", follow_redirects=True)
assert resp.status_code == 200
with client.session_transaction() as sess:
assert "user_id" in sess
def test_magic_login_invalid_token(self, client):
resp = client.get("/auth/magic/invalidtoken123", follow_redirects=True)
assert resp.status_code == 200 # Redirects to login with error message
if __name__ == "__main__":
pytest.main([__file__, "-v", "--tb=short"])