#!/usr/bin/env python3
"""AgentForms Hardening Tests (Phase 4)
Tests for:
1. Admin session-based auth (login, logout, rate limiting, session validation)
2. SMTP circuit breaker (open after N failures, reset on success, backoff timing)
3. Email verification enforcement (blocks unverified users from key actions)
4. Error monitoring (500/502/503 handlers trigger alerts)
Uses shared flask_app from app.app (same pattern as test_auth.py).
"""
import json
import os
import random
import sqlite3
import sys
import time
import bcrypt
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", "hardening-test-secret")
os.environ.setdefault("ADMIN_USER", "admin")
os.environ.setdefault("ADMIN_PASS", "agentforms-dev-test-pass")
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
class TestAdminSessionAuth:
"""Test admin session-based authentication."""
@pytest.fixture
def client(self):
return flask_app.test_client()
def test_admin_login_success(self, client):
"""Test successful admin login with correct credentials."""
resp = client.post("/admin/login", data={
"email": os.environ.get("ADMIN_USER", "admin"),
"password": os.environ.get("ADMIN_PASS", "agentforms-dev-test-pass"),
}, follow_redirects=False)
assert resp.status_code in (200, 302), f"Login failed: {resp.status_code}"
# Check that we got a session
with client.session_transaction() as sess:
assert "admin_authenticated" in sess or resp.status_code == 200
def test_admin_login_wrong_password(self, client):
"""Test failed admin login with wrong password."""
resp = client.post("/admin/login", data={
"email": os.environ.get("ADMIN_USER", "admin"),
"password": "wrongpassword",
})
assert resp.status_code == 200, f"Expected 200 (login page with error), got {resp.status_code}"
assert b"Invalid credentials" in resp.data
def test_admin_login_invalid_username(self, client):
"""Test failed admin login with non-existent username."""
resp = client.post("/admin/login", data={
"email": "nonexistent_user",
"password": "anypass",
})
assert resp.status_code == 200, f"Expected 200 (login page with error), got {resp.status_code}"
def test_admin_panel_requires_auth(self, client):
"""Test that admin panel redirects unauthenticated users."""
resp = client.get("/admin/dashboard", follow_redirects=False)
assert resp.status_code in (302, 401), f"Expected redirect/401, got {resp.status_code}"
def test_admin_panel_session_auth(self, client):
"""Test that admin panel works with valid session."""
# First login
username = os.environ.get("ADMIN_USER", "admin")
password = os.environ.get("ADMIN_PASS", "agentforms-dev-test-pass")
resp = client.post("/admin/login", data={"email": username, "password": password})
assert resp.status_code in (200, 302)
# Now access admin dashboard
resp = client.get("/admin/dashboard")
assert resp.status_code == 200
def test_admin_logout(self, client):
"""Test admin logout clears session."""
# Login first
client.post("/admin/login", data={
"email": os.environ.get("ADMIN_USER", "admin"),
"password": os.environ.get("ADMIN_PASS", "agentforms-dev-test-pass"),
})
# Logout
resp = client.post("/admin/logout")
assert resp.status_code == 302
# Verify session is cleared
with client.session_transaction() as sess:
assert "admin_authenticated" not in sess
def test_admin_bcrypt_password(self):
"""Test that bcrypt password verification works."""
password = os.environ.get("ADMIN_PASS", "agentforms-dev-test-pass")
if password.startswith("$2"):
assert bcrypt.checkpw(password.encode(), bcrypt.hashpw(password.encode(), bcrypt.gensalt()))
def test_admin_api_sites(self, client):
"""Test admin API /admin/api/sites with session auth."""
# Login
client.post("/admin/login", data={
"email": os.environ.get("ADMIN_USER", "admin"),
"password": os.environ.get("ADMIN_PASS", "agentforms-dev-test-pass"),
})
# Access API
resp = client.get("/admin/api/sites")
assert resp.status_code == 200
data = resp.get_json()
assert isinstance(data, list)
class TestSMTPCircuitBreaker:
"""Test SMTP circuit breaker functionality."""
def test_circuit_breaker_initial_state(self):
"""Test circuit breaker starts in closed state."""
from app.routes.email import NotificationService
assert NotificationService._smtp_circuit_open is False
assert NotificationService._smtp_failure_count == 0
def test_circuit_breaker_opens_after_failures(self):
"""Test circuit breaker opens after max failures."""
from app.routes.email import NotificationService
# Reset state
NotificationService._smtp_failure_count = 0
NotificationService._smtp_circuit_open = False
# Simulate failures
for i in range(NotificationService._SMTP_MAX_FAILURES):
NotificationService._record_failure()
assert NotificationService._smtp_circuit_open is True
assert NotificationService._smtp_failure_count >= NotificationService._SMTP_MAX_FAILURES
def test_circuit_breaker_resets_on_success(self):
"""Test circuit breaker resets after a successful send."""
from app.routes.email import NotificationService
# Set up as failed
NotificationService._smtp_failure_count = 10
NotificationService._smtp_circuit_open = True
# Record success
NotificationService._record_success()
assert NotificationService._smtp_circuit_open is False
assert NotificationService._smtp_failure_count == 0
def test_circuit_breaker_status(self):
"""Test circuit breaker status endpoint."""
from app.routes.email import NotificationService
status = NotificationService.circuit_status()
assert "open" in status
assert "failure_count" in status
assert "last_failure_time" in status
def test_circuit_breaker_half_open(self):
"""Test circuit breaker allows test request after backoff."""
from app.routes.email import NotificationService
# Set up as failed
NotificationService._smtp_failure_count = NotificationService._SMTP_MAX_FAILURES
NotificationService._smtp_circuit_open = True
NotificationService._smtp_last_failure_time = time.time() - NotificationService._SMTP_MAX_BACKOFF - 10
# Should be half-open (allow one attempt)
assert NotificationService._check_circuit_breaker() is False
def test_send_respects_circuit_breaker(self):
"""Test that send() returns False when circuit is open."""
from app.routes.email import NotificationService
# Open the circuit
NotificationService._smtp_failure_count = NotificationService._SMTP_MAX_FAILURES
NotificationService._smtp_circuit_open = True
NotificationService._smtp_last_failure_time = time.time()
# Send should return False without trying
result = NotificationService.send("test@test.com", "Test", "Test body")
assert result is False
def test_status_includes_circuit_breaker(self):
"""Test that status() includes circuit breaker info."""
from app.routes.email import NotificationService
status = NotificationService.status()
assert "circuit_breaker" in status
assert isinstance(status["circuit_breaker"], dict)
class TestEmailVerificationEnforcement:
"""Test email verification enforcement decorator."""
@pytest.fixture
def client(self):
return flask_app.test_client()
def _create_user(self, client, email_verified=True):
"""Helper to create and login a test user."""
email = f"hardening-test{random.randint(1000, 9999)}@example.com"
password = "TestPass123"
name = "Test User"
resp = client.post("/auth/register", data={
"email": email,
"password": password,
"name": name,
}, follow_redirects=True)
assert resp.status_code == 200
# Update email_verified status in DB
import app.models
from app.models import hash_value
with flask_app.app_context():
conn = sqlite3.connect(app.models.DB_PATH)
email_hash = hash_value(email.lower())
conn.execute("UPDATE users SET email_verified = ? WHERE email_hash = ?",
(email_verified, email_hash))
conn.commit()
conn.close()
return email, password
def test_unverified_user_blocked_from_site_creation(self, client):
"""Test unverified user is blocked from creating sites."""
email, password = self._create_user(client, email_verified=False)
# Try to create a site
resp = client.post("/sites/new", data={
"name": "Test Site",
"owner_email": email,
}, follow_redirects=False)
# Should redirect to email verification
assert resp.status_code == 302, f"Expected redirect, got {resp.status_code}"
assert "email/verify" in resp.headers.get("Location", "")
def test_verified_user_can_create_site(self, client):
"""Test verified user can create sites normally."""
email, password = self._create_user(client, email_verified=True)
resp = client.post("/sites/new", data={
"name": "Test Site",
"owner_email": email,
}, follow_redirects=True)
assert resp.status_code == 200
def test_unverified_user_json_error(self, client):
"""Test unverified user gets proper error on protected routes."""
# Create unverified user and login
email, password = self._create_user(client, email_verified=False)
# /sites/new uses @verified_email_required — unverified users get redirected
resp = client.post("/sites/new", data={
"name": "Test Site",
"owner_email": email,
}, follow_redirects=False)
# Should redirect to email verification (302) or get 403
assert resp.status_code in (302, 403), f"Expected redirect/403, got {resp.status_code}"
class TestErrorMonitoring:
"""Test error monitoring and alerting."""
@pytest.fixture
def client(self):
return flask_app.test_client()
def test_500_handler(self, client):
"""Test 500 error handler returns proper response."""
resp = client.get("/this-route-does-not-exist-500-test")
# Should not crash the server
assert resp.status_code in (404, 500)
def test_alerting_service_configured(self):
"""Test that alerting service is properly configured."""
from app.services.alerting import AlertService
assert hasattr(AlertService, "error")
assert hasattr(AlertService, "warning")
if __name__ == "__main__":
pytest.main([__file__, "-v", "--tb=short"])