feat: production launch — email audit, team invites, security headers, router fixes
Some checks failed
CI / ruff (push) Successful in 9s
CI / pytest (push) Failing after 47m32s
CI / validate (push) Successful in 23s
CI / dependency-scanning (push) Successful in 29s
CI / docs (push) Has been skipped
CI / deploy (push) Has been skipped

- Fix loyalty & monitoring router bugs (_get_router → named routers)
- Implement team invitation email with send_template + seed templates (en/fr/de)
- Add SecurityHeadersMiddleware (nosniff, HSTS, referrer-policy, permissions-policy)
- Build email audit admin page: service, schemas, API, page route, menu, i18n, HTML, JS
- Clean stale TODO in platform-menu-config.js
- Add 67 tests (unit + integration) covering all new functionality

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-02-27 18:24:30 +01:00
parent 4ebd419987
commit ce822af883
25 changed files with 2485 additions and 19 deletions

View File

@@ -0,0 +1,371 @@
# app/modules/messaging/tests/integration/test_email_logs_api.py
"""
Integration tests for admin email logs (audit) API endpoints.
Tests the API routes at:
GET /api/v1/admin/email-logs
GET /api/v1/admin/email-logs/stats
GET /api/v1/admin/email-logs/{log_id}
Authentication: Uses super_admin_headers fixture (real JWT login).
"""
import uuid
from datetime import datetime
import pytest
from app.modules.messaging.models import EmailLog, EmailStatus
BASE = "/api/v1/admin/email-logs"
# =============================================================================
# FIXTURES
# =============================================================================
@pytest.fixture
def audit_logs(db):
"""Create email logs for integration testing."""
logs = []
# Sent logs
for i in range(3):
uid = uuid.uuid4().hex[:8]
log = EmailLog(
template_code="signup_welcome",
recipient_email=f"sent_{uid}@example.com",
recipient_name=f"Sent User {i}",
subject=f"Welcome {uid}",
body_html=f"<p>Welcome {uid}</p>",
body_text=f"Welcome {uid}",
from_email="noreply@orion.lu",
from_name="Orion",
status=EmailStatus.SENT.value,
provider="debug",
sent_at=datetime.utcnow(),
)
db.add(log)
logs.append(log)
# Failed logs
for i in range(2):
uid = uuid.uuid4().hex[:8]
log = EmailLog(
template_code="order_confirmation",
recipient_email=f"failed_{uid}@example.com",
recipient_name=f"Failed User {i}",
subject=f"Order {uid}",
body_html=f"<p>Order {uid}</p>",
from_email="noreply@orion.lu",
status=EmailStatus.FAILED.value,
provider="smtp",
error_message="Connection refused",
)
db.add(log)
logs.append(log)
# Pending log
uid = uuid.uuid4().hex[:8]
log = EmailLog(
template_code="team_invitation",
recipient_email=f"pending_{uid}@example.com",
subject=f"Invitation {uid}",
from_email="noreply@orion.lu",
status=EmailStatus.PENDING.value,
)
db.add(log)
logs.append(log)
db.commit()
for log in logs:
db.refresh(log)
return logs
# =============================================================================
# LIST EMAIL LOGS
# =============================================================================
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.email
class TestListEmailLogs:
"""Tests for GET /api/v1/admin/email-logs."""
def test_list_logs_success(self, client, super_admin_headers, audit_logs):
"""Admin can list email logs."""
response = client.get(BASE, headers=super_admin_headers)
assert response.status_code == 200
data = response.json()
assert "items" in data
assert "total" in data
assert "page" in data
assert "per_page" in data
assert "total_pages" in data
assert data["total"] == 6
def test_list_logs_pagination(self, client, super_admin_headers, audit_logs):
"""Pagination parameters work correctly."""
response = client.get(
f"{BASE}?page=1&per_page=2",
headers=super_admin_headers,
)
assert response.status_code == 200
data = response.json()
assert len(data["items"]) == 2
assert data["total"] == 6
assert data["page"] == 1
assert data["per_page"] == 2
assert data["total_pages"] == 3
def test_list_logs_page_two(self, client, super_admin_headers, audit_logs):
"""Second page returns remaining items."""
response = client.get(
f"{BASE}?page=2&per_page=4",
headers=super_admin_headers,
)
assert response.status_code == 200
data = response.json()
assert len(data["items"]) == 2
assert data["page"] == 2
def test_filter_by_status(self, client, super_admin_headers, audit_logs):
"""Filter by status returns matching logs."""
response = client.get(
f"{BASE}?status=sent",
headers=super_admin_headers,
)
assert response.status_code == 200
data = response.json()
assert data["total"] == 3
assert all(item["status"] == "sent" for item in data["items"])
def test_filter_by_template_code(self, client, super_admin_headers, audit_logs):
"""Filter by template_code returns matching logs."""
response = client.get(
f"{BASE}?template_code=order_confirmation",
headers=super_admin_headers,
)
assert response.status_code == 200
data = response.json()
assert data["total"] == 2
assert all(
item["template_code"] == "order_confirmation"
for item in data["items"]
)
def test_filter_by_search(self, client, super_admin_headers, audit_logs):
"""Search filter matches recipient email."""
email = audit_logs[0].recipient_email
response = client.get(
f"{BASE}?search={email}",
headers=super_admin_headers,
)
assert response.status_code == 200
data = response.json()
assert data["total"] >= 1
assert any(email in item["recipient_email"] for item in data["items"])
def test_combined_filters(self, client, super_admin_headers, audit_logs):
"""Multiple filters can be combined."""
response = client.get(
f"{BASE}?status=failed&template_code=order_confirmation",
headers=super_admin_headers,
)
assert response.status_code == 200
data = response.json()
assert data["total"] == 2
def test_empty_result(self, client, super_admin_headers, audit_logs):
"""Non-matching filter returns empty list."""
response = client.get(
f"{BASE}?search=nonexistent_xyz@nowhere.test",
headers=super_admin_headers,
)
assert response.status_code == 200
data = response.json()
assert data["total"] == 0
assert data["items"] == []
def test_requires_auth(self, client, audit_logs):
"""Unauthenticated request is rejected."""
response = client.get(BASE)
assert response.status_code == 401
def test_item_structure(self, client, super_admin_headers, audit_logs):
"""Response items have expected fields."""
response = client.get(
f"{BASE}?per_page=1",
headers=super_admin_headers,
)
assert response.status_code == 200
item = response.json()["items"][0]
assert "id" in item
assert "recipient_email" in item
assert "subject" in item
assert "status" in item
assert "template_code" in item
assert "created_at" in item
# List items should NOT include body
assert "body_html" not in item
assert "body_text" not in item
# =============================================================================
# EMAIL LOG STATS
# =============================================================================
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.email
class TestEmailLogStats:
"""Tests for GET /api/v1/admin/email-logs/stats."""
def test_stats_success(self, client, super_admin_headers, audit_logs):
"""Admin can retrieve email log stats."""
response = client.get(
f"{BASE}/stats",
headers=super_admin_headers,
)
assert response.status_code == 200
data = response.json()
assert "by_status" in data
assert "by_template" in data
assert "total" in data
def test_stats_status_counts(self, client, super_admin_headers, audit_logs):
"""Stats include correct counts by status."""
response = client.get(
f"{BASE}/stats",
headers=super_admin_headers,
)
data = response.json()
assert data["by_status"]["sent"] == 3
assert data["by_status"]["failed"] == 2
assert data["by_status"]["pending"] == 1
assert data["total"] == 6
def test_stats_template_counts(self, client, super_admin_headers, audit_logs):
"""Stats include correct counts by template."""
response = client.get(
f"{BASE}/stats",
headers=super_admin_headers,
)
data = response.json()
assert data["by_template"]["signup_welcome"] == 3
assert data["by_template"]["order_confirmation"] == 2
assert data["by_template"]["team_invitation"] == 1
def test_stats_requires_auth(self, client, audit_logs):
"""Unauthenticated request is rejected."""
response = client.get(f"{BASE}/stats")
assert response.status_code == 401
def test_stats_empty_database(self, client, super_admin_headers):
"""Stats return zeros when no logs exist."""
response = client.get(
f"{BASE}/stats",
headers=super_admin_headers,
)
assert response.status_code == 200
data = response.json()
assert data["total"] == 0
# =============================================================================
# EMAIL LOG DETAIL
# =============================================================================
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.email
class TestEmailLogDetail:
"""Tests for GET /api/v1/admin/email-logs/{log_id}."""
def test_detail_success(self, client, super_admin_headers, audit_logs):
"""Admin can retrieve full log detail."""
log_id = audit_logs[0].id
response = client.get(
f"{BASE}/{log_id}",
headers=super_admin_headers,
)
assert response.status_code == 200
data = response.json()
assert data["id"] == log_id
assert data["recipient_email"] == audit_logs[0].recipient_email
assert data["body_html"] is not None
assert data["from_email"] == "noreply@orion.lu"
def test_detail_includes_body(self, client, super_admin_headers, audit_logs):
"""Detail response includes body_html and body_text."""
log_id = audit_logs[0].id
response = client.get(
f"{BASE}/{log_id}",
headers=super_admin_headers,
)
data = response.json()
assert "body_html" in data
assert "body_text" in data
assert data["body_html"] is not None
def test_detail_includes_timestamps(self, client, super_admin_headers, audit_logs):
"""Detail response includes all timestamp fields."""
log_id = audit_logs[0].id
response = client.get(
f"{BASE}/{log_id}",
headers=super_admin_headers,
)
data = response.json()
assert "created_at" in data
assert "sent_at" in data
assert "delivered_at" in data
assert "opened_at" in data
assert "clicked_at" in data
def test_detail_failed_log_has_error(self, client, super_admin_headers, audit_logs):
"""Failed log detail includes error message."""
failed_log = next(log for log in audit_logs if log.status == EmailStatus.FAILED.value)
response = client.get(
f"{BASE}/{failed_log.id}",
headers=super_admin_headers,
)
data = response.json()
assert data["status"] == "failed"
assert data["error_message"] == "Connection refused"
def test_detail_not_found(self, client, super_admin_headers):
"""Non-existent log returns 404."""
response = client.get(
f"{BASE}/999999",
headers=super_admin_headers,
)
assert response.status_code == 404
def test_detail_requires_auth(self, client, audit_logs):
"""Unauthenticated request is rejected."""
log_id = audit_logs[0].id
response = client.get(f"{BASE}/{log_id}")
assert response.status_code == 401

View File

@@ -0,0 +1,53 @@
# app/modules/messaging/tests/integration/test_email_logs_page.py
"""
Integration tests for admin email logs page route (HTML rendering).
Tests the page route at:
GET /admin/email-logs
Authentication: Uses super_admin_headers fixture (real JWT login).
"""
import pytest
BASE = "/admin"
@pytest.mark.integration
@pytest.mark.email
class TestAdminEmailLogsPage:
"""Tests for GET /admin/email-logs."""
def test_page_renders(self, client, super_admin_headers):
"""Email logs page returns HTML."""
response = client.get(
f"{BASE}/email-logs",
headers=super_admin_headers,
)
assert response.status_code == 200
assert "text/html" in response.headers["content-type"]
def test_page_contains_alpine_component(self, client, super_admin_headers):
"""Page HTML references the emailLogsPage Alpine component."""
response = client.get(
f"{BASE}/email-logs",
headers=super_admin_headers,
)
assert b"emailLogsPage" in response.content
def test_page_includes_js(self, client, super_admin_headers):
"""Page HTML includes the email-logs.js script."""
response = client.get(
f"{BASE}/email-logs",
headers=super_admin_headers,
)
assert b"email-logs.js" in response.content
def test_page_requires_auth(self, client):
"""Unauthenticated request is redirected."""
response = client.get(
f"{BASE}/email-logs",
follow_redirects=False,
)
# Should redirect to login or return 401/403
assert response.status_code in (301, 302, 303, 307, 401, 403)

View File

@@ -0,0 +1,358 @@
# app/modules/messaging/tests/unit/test_email_log_service.py
"""Unit tests for EmailTemplateService email log (audit) methods."""
import uuid
from datetime import datetime, timedelta
import pytest
from app.modules.messaging.models import (
EmailCategory,
EmailLog,
EmailStatus,
EmailTemplate,
)
from app.modules.messaging.services.email_template_service import EmailTemplateService
# =============================================================================
# FIXTURES
# =============================================================================
@pytest.fixture
def email_template(db):
"""Create a test email template."""
uid = uuid.uuid4().hex[:8]
template = EmailTemplate(
code=f"test_template_{uid}",
language="en",
name="Test Template",
description="A test template",
category=EmailCategory.SYSTEM.value,
subject="Test subject {{ name }}",
body_html="<p>Hello {{ name }}</p>",
body_text="Hello {{ name }}",
is_active=True,
)
db.add(template)
db.commit()
db.refresh(template)
return template
@pytest.fixture
def email_logs(db, email_template):
"""Create a batch of email logs with various statuses."""
logs = []
statuses = [
(EmailStatus.SENT, 3),
(EmailStatus.FAILED, 2),
(EmailStatus.PENDING, 1),
(EmailStatus.DELIVERED, 2),
(EmailStatus.BOUNCED, 1),
]
for status, count in statuses:
for i in range(count):
uid = uuid.uuid4().hex[:8]
log = EmailLog(
template_code=email_template.code,
recipient_email=f"user_{uid}@example.com",
recipient_name=f"User {uid}",
subject=f"Test subject {uid}",
body_html=f"<p>Hello {uid}</p>",
body_text=f"Hello {uid}",
from_email="noreply@orion.lu",
from_name="Orion",
status=status.value,
provider="debug",
sent_at=datetime.utcnow() if status != EmailStatus.PENDING else None,
)
db.add(log)
logs.append(log)
db.commit()
for log in logs:
db.refresh(log)
return logs
@pytest.fixture
def multi_template_logs(db):
"""Create logs from multiple templates for stats testing."""
logs = []
templates = ["signup_welcome", "order_confirmation", "password_reset"]
for tpl_code in templates:
for i in range(2):
uid = uuid.uuid4().hex[:8]
log = EmailLog(
template_code=tpl_code,
recipient_email=f"{tpl_code}_{uid}@example.com",
subject=f"{tpl_code} email",
from_email="noreply@orion.lu",
status=EmailStatus.SENT.value,
provider="debug",
sent_at=datetime.utcnow(),
)
db.add(log)
logs.append(log)
# Add one failed log
log = EmailLog(
template_code="password_reset",
recipient_email="failed@example.com",
subject="Failed email",
from_email="noreply@orion.lu",
status=EmailStatus.FAILED.value,
provider="debug",
error_message="SMTP connection refused",
)
db.add(log)
logs.append(log)
db.commit()
for log in logs:
db.refresh(log)
return logs
# =============================================================================
# GET EMAIL LOGS
# =============================================================================
@pytest.mark.unit
@pytest.mark.email
class TestGetEmailLogs:
"""Tests for get_email_logs() service method."""
def test_returns_paginated_logs(self, db, email_logs):
"""Logs are returned with correct total count."""
service = EmailTemplateService(db)
items, total = service.get_email_logs(skip=0, limit=50)
assert total == 9 # 3 sent + 2 failed + 1 pending + 2 delivered + 1 bounced
assert len(items) == 9
def test_pagination_limit(self, db, email_logs):
"""Limit restricts number of returned items."""
service = EmailTemplateService(db)
items, total = service.get_email_logs(skip=0, limit=3)
assert total == 9
assert len(items) == 3
def test_pagination_skip(self, db, email_logs):
"""Skip offsets the results."""
service = EmailTemplateService(db)
items, total = service.get_email_logs(skip=5, limit=50)
assert total == 9
assert len(items) == 4
def test_filter_by_status(self, db, email_logs):
"""Filter by status returns matching logs only."""
service = EmailTemplateService(db)
items, total = service.get_email_logs(
filters={"status": "sent"}, skip=0, limit=50
)
assert total == 3
assert all(item["status"] == "sent" for item in items)
def test_filter_by_status_failed(self, db, email_logs):
"""Filter by failed status."""
service = EmailTemplateService(db)
items, total = service.get_email_logs(
filters={"status": "failed"}, skip=0, limit=50
)
assert total == 2
assert all(item["status"] == "failed" for item in items)
def test_filter_by_template_code(self, db, email_logs, email_template):
"""Filter by template_code returns matching logs."""
service = EmailTemplateService(db)
items, total = service.get_email_logs(
filters={"template_code": email_template.code}, skip=0, limit=50
)
assert total == 9
assert all(item["template_code"] == email_template.code for item in items)
def test_filter_by_recipient_search(self, db, email_logs):
"""Search filter matches recipient email substring."""
# Get the first log's email for a search term
service = EmailTemplateService(db)
first_email = email_logs[0].recipient_email
items, total = service.get_email_logs(
filters={"search": first_email}, skip=0, limit=50
)
assert total >= 1
assert any(first_email in item["recipient_email"] for item in items)
def test_filter_by_date_range(self, db, email_logs):
"""Date range filter works."""
service = EmailTemplateService(db)
yesterday = (datetime.utcnow() - timedelta(days=1)).isoformat()
tomorrow = (datetime.utcnow() + timedelta(days=1)).isoformat()
items, total = service.get_email_logs(
filters={"date_from": yesterday, "date_to": tomorrow},
skip=0, limit=50,
)
assert total == 9
def test_combined_filters(self, db, email_logs, email_template):
"""Multiple filters can be combined."""
service = EmailTemplateService(db)
items, total = service.get_email_logs(
filters={
"status": "sent",
"template_code": email_template.code,
},
skip=0, limit=50,
)
assert total == 3
assert all(item["status"] == "sent" for item in items)
def test_empty_result(self, db, email_logs):
"""Non-matching filter returns empty list."""
service = EmailTemplateService(db)
items, total = service.get_email_logs(
filters={"search": "nonexistent_address_xyz@nowhere.test"},
skip=0, limit=50,
)
assert total == 0
assert items == []
def test_log_item_structure(self, db, email_logs):
"""Returned items contain expected fields (no body content)."""
service = EmailTemplateService(db)
items, _ = service.get_email_logs(skip=0, limit=1)
item = items[0]
assert "id" in item
assert "recipient_email" in item
assert "subject" in item
assert "status" in item
assert "template_code" in item
assert "created_at" in item
# Body fields should NOT be in list items
assert "body_html" not in item
assert "body_text" not in item
def test_no_filters(self, db, email_logs):
"""Calling with None filters returns all logs."""
service = EmailTemplateService(db)
items, total = service.get_email_logs(filters=None, skip=0, limit=50)
assert total == 9
# =============================================================================
# GET EMAIL LOG DETAIL
# =============================================================================
@pytest.mark.unit
@pytest.mark.email
class TestGetEmailLogDetail:
"""Tests for get_email_log_detail() service method."""
def test_returns_full_detail(self, db, email_logs):
"""Detail includes body content and all metadata."""
service = EmailTemplateService(db)
log = email_logs[0]
detail = service.get_email_log_detail(log.id)
assert detail["id"] == log.id
assert detail["recipient_email"] == log.recipient_email
assert detail["subject"] == log.subject
assert detail["status"] == log.status
assert detail["body_html"] is not None
assert detail["body_text"] is not None
assert detail["from_email"] == "noreply@orion.lu"
def test_includes_timestamp_fields(self, db, email_logs):
"""Detail includes all timestamp fields."""
service = EmailTemplateService(db)
log = email_logs[0]
detail = service.get_email_log_detail(log.id)
assert "created_at" in detail
assert "sent_at" in detail
assert "delivered_at" in detail
assert "opened_at" in detail
assert "clicked_at" in detail
def test_not_found_raises(self, db):
"""Non-existent log ID raises ResourceNotFoundException."""
from app.exceptions.base import ResourceNotFoundException
service = EmailTemplateService(db)
with pytest.raises(ResourceNotFoundException):
service.get_email_log_detail(999999)
def test_includes_provider_info(self, db, email_logs):
"""Detail includes provider information."""
service = EmailTemplateService(db)
log = email_logs[0]
detail = service.get_email_log_detail(log.id)
assert detail["provider"] == "debug"
# =============================================================================
# GET EMAIL LOG STATS
# =============================================================================
@pytest.mark.unit
@pytest.mark.email
class TestGetEmailLogStats:
"""Tests for get_email_log_stats() service method."""
def test_returns_status_counts(self, db, email_logs):
"""Stats include counts grouped by status."""
service = EmailTemplateService(db)
stats = service.get_email_log_stats()
assert stats["by_status"]["sent"] == 3
assert stats["by_status"]["failed"] == 2
assert stats["by_status"]["pending"] == 1
assert stats["by_status"]["delivered"] == 2
assert stats["by_status"]["bounced"] == 1
def test_returns_template_counts(self, db, multi_template_logs):
"""Stats include counts grouped by template_code."""
service = EmailTemplateService(db)
stats = service.get_email_log_stats()
assert stats["by_template"]["signup_welcome"] == 2
assert stats["by_template"]["order_confirmation"] == 2
assert stats["by_template"]["password_reset"] == 3 # 2 sent + 1 failed
def test_returns_total(self, db, email_logs):
"""Stats total matches sum of all statuses."""
service = EmailTemplateService(db)
stats = service.get_email_log_stats()
assert stats["total"] == 9
assert stats["total"] == sum(stats["by_status"].values())
def test_empty_database(self, db):
"""Stats return zeros when no logs exist."""
service = EmailTemplateService(db)
stats = service.get_email_log_stats()
assert stats["total"] == 0
assert stats["by_status"] == {}
assert stats["by_template"] == {}