"""
Email Templates (INDL-32) tests.
Covers PRD Test Scenarios T-01..T-14 (API-observable subset) and Security
Acceptance Criteria SEC-01, SEC-02, SEC-06, SEC-07, SEC-08, SEC-10 for the
``/api/v1/settings/email-templates`` endpoints.
NOTE: These tests require migrations 0047 ("align email_templates schema with
INDL-32 PRD") and 0048 ("unique index on (tenant_id, name)") to be
applied to the test database. At the time this file was written,
``alembic upgrade head`` was BROKEN for the whole project (see
src/database/migrations/versions/0042_contact_inquiries_topic_fields.py:
revision="0042a" but down_revision="0042", and no migration file has
revision="0042" — Alembic cannot build its revision map at all). Both
the ``indelis`` and ``indelis_test`` databases were found still on the
OLD ad-hoc email_templates shape (trigger/is_enabled/variables columns)
despite alembic_version reporting 0051/0052. The 0047/0048 DDL was
applied by hand directly to indelis_test to unblock this test run; the
underlying migration-chain bug is reported separately and must be fixed
before these tests (or any fresh `alembic upgrade head`) will pass in
a clean environment / CI.
"""
from __future__ import annotations
from uuid import uuid4
import pytest
import pytest_asyncio
from httpx import AsyncClient
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from unittest.mock import AsyncMock, MagicMock, patch
from src.apps.settings.services.email_template_service import EmailTemplateService
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest_asyncio.fixture
async def tenant_account(db_session: AsyncSession):
"""Fresh tenant account for each test (unique subdomain avoids collisions)."""
from src.apps.tenants.models.account import Account
account = Account(
organization_name="Email Template Test Cemetery",
subdomain=f"email-tpl-test-{uuid4().hex[:8]}",
contact_email=f"admin-{uuid4().hex[:6]}@emailtpltest.com",
plan="starter",
status="active",
)
db_session.add(account)
await db_session.flush()
return account
@pytest_asyncio.fixture
async def other_tenant_account(db_session: AsyncSession):
"""A second, unrelated tenant — used for cross-tenant isolation tests."""
from src.apps.tenants.models.account import Account
account = Account(
organization_name="Other Cemetery",
subdomain=f"other-tpl-test-{uuid4().hex[:8]}",
contact_email=f"admin-{uuid4().hex[:6]}@othertpltest.com",
plan="starter",
status="active",
)
db_session.add(account)
await db_session.flush()
return account
async def _make_user(db_session: AsyncSession, tenant_account, role: str):
from src.apps.auth.models.user import User
from src.core.security import hash_password
user = User(
tenant_id=tenant_account.id,
email=f"{role}-{uuid4().hex[:6]}@emailtpltest.com",
password_hash=hash_password("TestPassword123!"),
first_name=role.capitalize(),
last_name="User",
role=role,
status="active",
)
db_session.add(user)
await db_session.flush()
return user
async def _headers_for(user, tenant_account) -> dict:
from src.core.security import create_access_token, build_token_payload
token = create_access_token(build_token_payload(user, tenant_account))
return {
"Authorization": f"Bearer {token}",
"X-Tenant-ID": str(tenant_account.id),
}
@pytest_asyncio.fixture
async def admin_user(db_session: AsyncSession, tenant_account):
return await _make_user(db_session, tenant_account, "administrator")
@pytest_asyncio.fixture
async def manager_user(db_session: AsyncSession, tenant_account):
return await _make_user(db_session, tenant_account, "manager")
@pytest_asyncio.fixture
async def staff_user(db_session: AsyncSession, tenant_account):
return await _make_user(db_session, tenant_account, "staff")
@pytest_asyncio.fixture
async def view_only_user(db_session: AsyncSession, tenant_account):
return await _make_user(db_session, tenant_account, "view_only")
@pytest_asyncio.fixture
async def admin_headers(admin_user, tenant_account):
return await _headers_for(admin_user, tenant_account)
@pytest_asyncio.fixture
async def manager_headers(manager_user, tenant_account):
return await _headers_for(manager_user, tenant_account)
@pytest_asyncio.fixture
async def staff_headers(staff_user, tenant_account):
return await _headers_for(staff_user, tenant_account)
@pytest_asyncio.fixture
async def view_only_headers(view_only_user, tenant_account):
return await _headers_for(view_only_user, tenant_account)
@pytest_asyncio.fixture
async def other_admin_user(db_session: AsyncSession, other_tenant_account):
return await _make_user(db_session, other_tenant_account, "administrator")
@pytest_asyncio.fixture
async def other_admin_headers(other_admin_user, other_tenant_account):
return await _headers_for(other_admin_user, other_tenant_account)
async def _seed(db_session: AsyncSession, tenant_account):
"""Seed the 17 system templates for a tenant and flush (INDL-54: was 8)."""
created = await EmailTemplateService.seed_system_templates(db_session, tenant_account.id)
await db_session.flush()
return created
def _create_payload(**overrides) -> dict:
payload = {
"name": f"Custom template {uuid4().hex[:8]}",
"subject": "A custom subject",
"body": "A custom body with {{cemetery_name}}.",
"status": "active",
}
payload.update(overrides)
return payload
# ---------------------------------------------------------------------------
# T-01 — list returns the 8 seeded system templates
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_list_returns_17_seeded_system_templates_for_fresh_tenant(
client: AsyncClient, db_session: AsyncSession, tenant_account, admin_headers
):
await _seed(db_session, tenant_account)
resp = await client.get("/api/v1/settings/email-templates", headers=admin_headers)
assert resp.status_code == 200, resp.text
body = resp.json()
assert body["success"] is True
items = body["data"]["items"]
# INDL-54 extended the seeded catalog from 8 to 17.
assert body["data"]["total"] == 17
assert len(items) == 17
assert all(item["is_system"] is True for item in items)
keys = {item["template_key"] for item in items}
assert keys == {
"contract_signed",
"invoice_payment_received",
"invoice_overdue_reminder",
"service_scheduled_family",
"field_crew_briefing",
"tribute_moderated",
"deed_transferred",
"new_user_welcome",
# INDL-54 additions:
"staff_invitation",
"staff_invitation_resend",
"contact_enquiry_reply",
"lead_qualified",
"proposal_sent",
"proposal_resent",
"contract_pdf_confirmation",
"service_reminder_24h",
"news_notification",
}
# PRD: "deed transferred" seeds OFF by default.
deed = next(item for item in items if item["template_key"] == "deed_transferred")
assert deed["status"] == "inactive"
# ---------------------------------------------------------------------------
# T-09 / SEC-01 — create as administrator
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_create_as_administrator_returns_201_and_appears_in_list(
client: AsyncClient, db_session: AsyncSession, tenant_account, admin_headers
):
await _seed(db_session, tenant_account)
payload = _create_payload()
resp = await client.post(
"/api/v1/settings/email-templates", json=payload, headers=admin_headers
)
assert resp.status_code == 201, resp.text
created = resp.json()["data"]
assert created["name"] == payload["name"]
assert created["is_system"] is False
assert created["template_key"] is None
list_resp = await client.get("/api/v1/settings/email-templates", headers=admin_headers)
names = {item["name"] for item in list_resp.json()["data"]["items"]}
assert payload["name"] in names
# 17 seeded system templates + 1 newly created custom template.
assert list_resp.json()["data"]["total"] == 18
# ---------------------------------------------------------------------------
# SEC-01 / AC-29 — create forbidden for staff, manager, view_only
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_create_as_staff_returns_403(client: AsyncClient, staff_headers):
resp = await client.post(
"/api/v1/settings/email-templates", json=_create_payload(), headers=staff_headers
)
assert resp.status_code == 403, resp.text
@pytest.mark.asyncio
async def test_create_as_manager_returns_403(client: AsyncClient, manager_headers):
resp = await client.post(
"/api/v1/settings/email-templates", json=_create_payload(), headers=manager_headers
)
assert resp.status_code == 403, resp.text
@pytest.mark.asyncio
async def test_create_as_view_only_returns_403(client: AsyncClient, view_only_headers):
resp = await client.post(
"/api/v1/settings/email-templates", json=_create_payload(), headers=view_only_headers
)
assert resp.status_code == 403, resp.text
# ---------------------------------------------------------------------------
# AC-28 / T-10 — duplicate name -> 409
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_create_with_duplicate_name_returns_409_with_exact_message(
client: AsyncClient, db_session: AsyncSession, tenant_account, admin_headers
):
seeded = await _seed(db_session, tenant_account)
existing_name = seeded[0].name
resp = await client.post(
"/api/v1/settings/email-templates",
json=_create_payload(name=existing_name),
headers=admin_headers,
)
assert resp.status_code == 409, resp.text
assert resp.json()["message"] == "A template with this name already exists"
# ---------------------------------------------------------------------------
# SEC-07 / API1 — cross-tenant GET -> 404, not 403
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_get_template_belonging_to_different_tenant_returns_404(
client: AsyncClient,
db_session: AsyncSession,
tenant_account,
admin_headers,
other_admin_headers,
):
seeded = await _seed(db_session, tenant_account)
target_id = str(seeded[0].id)
resp = await client.get(
f"/api/v1/settings/email-templates/{target_id}", headers=other_admin_headers
)
assert resp.status_code == 404, resp.text
assert resp.status_code != 403
# ---------------------------------------------------------------------------
# SEC-08 / AC-14 / PUT note — tenant_id / is_system / template_key immutable
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_put_with_protected_fields_in_body_does_not_mutate_them(
client: AsyncClient, db_session: AsyncSession, tenant_account, admin_headers, other_tenant_account
):
seeded = await _seed(db_session, tenant_account)
template = seeded[0]
assert template.is_system is True
original_template_key = template.template_key
resp = await client.put(
f"/api/v1/settings/email-templates/{template.id}",
json={
"subject": "Updated subject",
"body": "Updated body {{cemetery_name}}",
"tenant_id": str(other_tenant_account.id),
"is_system": False,
"template_key": "hacked_key",
},
headers=admin_headers,
)
assert resp.status_code == 200, resp.text
get_resp = await client.get(
f"/api/v1/settings/email-templates/{template.id}", headers=admin_headers
)
data = get_resp.json()["data"]
assert data["subject"] == "Updated subject"
assert data["is_system"] is True
assert data["template_key"] == original_template_key
# ---------------------------------------------------------------------------
# AC-08 / T-02 — PATCH toggles status only
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_patch_status_toggles_and_leaves_subject_body_unchanged(
client: AsyncClient, db_session: AsyncSession, tenant_account, admin_headers
):
seeded = await _seed(db_session, tenant_account)
deed = next(t for t in seeded if t.template_key == "deed_transferred")
assert deed.status == "inactive"
original_subject = deed.subject
original_body = deed.body
resp = await client.patch(
f"/api/v1/settings/email-templates/{deed.id}",
json={"status": "active"},
headers=admin_headers,
)
assert resp.status_code == 200, resp.text
assert resp.json()["data"]["status"] == "active"
get_resp = await client.get(
f"/api/v1/settings/email-templates/{deed.id}", headers=admin_headers
)
data = get_resp.json()["data"]
assert data["status"] == "active"
assert data["subject"] == original_subject
assert data["body"] == original_body
# ---------------------------------------------------------------------------
# AC-09 / DELETE — system template cannot be deleted
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_delete_system_template_returns_403(
client: AsyncClient, db_session: AsyncSession, tenant_account, admin_headers
):
seeded = await _seed(db_session, tenant_account)
template = seeded[0]
resp = await client.delete(
f"/api/v1/settings/email-templates/{template.id}", headers=admin_headers
)
assert resp.status_code == 403, resp.text
# ---------------------------------------------------------------------------
# AC-10 — delete custom template soft-deletes and hides it from list
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_delete_custom_template_soft_deletes_and_hides_from_list(
client: AsyncClient, db_session: AsyncSession, tenant_account, admin_headers
):
await _seed(db_session, tenant_account)
create_resp = await client.post(
"/api/v1/settings/email-templates", json=_create_payload(), headers=admin_headers
)
template_id = create_resp.json()["data"]["id"]
del_resp = await client.delete(
f"/api/v1/settings/email-templates/{template_id}", headers=admin_headers
)
assert del_resp.status_code == 200, del_resp.text
from src.apps.settings.models.email_template import EmailTemplate
row = (
await db_session.execute(
select(EmailTemplate).where(EmailTemplate.id == template_id)
)
).scalar_one()
assert row.deleted_at is not None
list_resp = await client.get("/api/v1/settings/email-templates", headers=admin_headers)
ids = {item["id"] for item in list_resp.json()["data"]["items"]}
assert template_id not in ids
# ---------------------------------------------------------------------------
# SEC-10 / T rate limit — 6th test-send within window -> 429
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_send_test_sixth_call_within_window_returns_429(
client: AsyncClient, db_session: AsyncSession, tenant_account, admin_headers
):
"""Mocks the Redis client the router obtains via get_redis(), matching the
project's established pattern (see tests/apps/payments/test_payment_link.py),
since a real Redis instance is not asserted as available in this test
environment. This tests the router's counting/threshold logic in isolation
from actual Redis network reachability.
"""
seeded = await _seed(db_session, tenant_account)
template_id = str(seeded[0].id)
call_count = {"n": 0}
async def fake_incr(key):
call_count["n"] += 1
return call_count["n"]
mock_redis = MagicMock()
mock_redis.incr = AsyncMock(side_effect=fake_incr)
mock_redis.expire = AsyncMock()
mock_redis.ttl = AsyncMock(return_value=3600)
mock_redis.aclose = AsyncMock()
with patch(
"src.apps.settings.router.get_redis", AsyncMock(return_value=mock_redis)
), patch("src.core.config.settings.SMTP_HOST", "smtp.example.com"), patch(
"src.apps.settings.services.email_template_service._dispatch", AsyncMock()
):
responses = []
for _ in range(6):
resp = await client.post(
f"/api/v1/settings/email-templates/{template_id}/test",
headers=admin_headers,
)
responses.append(resp)
for r in responses[:5]:
assert r.status_code == 200, r.text
assert responses[5].status_code == 429, responses[5].text
assert "Retry-After" in responses[5].headers
# ---------------------------------------------------------------------------
# Custom to_email override — product decision to allow sending a test to any
# address (not just the requesting admin's own email), keeping the rate
# limit + audit log as the abuse controls.
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_send_test_with_custom_to_email_overrides_default(
client: AsyncClient, db_session: AsyncSession, tenant_account, admin_headers
):
seeded = await _seed(db_session, tenant_account)
template_id = str(seeded[0].id)
mock_redis = MagicMock()
mock_redis.incr = AsyncMock(return_value=1)
mock_redis.expire = AsyncMock()
mock_redis.aclose = AsyncMock()
mock_dispatch = AsyncMock()
with patch(
"src.apps.settings.router.get_redis", AsyncMock(return_value=mock_redis)
), patch("src.core.config.settings.SMTP_HOST", "smtp.example.com"), patch(
"src.apps.settings.services.email_template_service._dispatch", mock_dispatch
):
resp = await client.post(
f"/api/v1/settings/email-templates/{template_id}/test",
json={"to_email": "someone-else@example.com"},
headers=admin_headers,
)
assert resp.status_code == 200, resp.text
assert resp.json()["message"] == "Test email sent to someone-else@example.com"
mock_dispatch.assert_awaited_once()
assert mock_dispatch.call_args.kwargs["to"] == "someone-else@example.com"
@pytest.mark.asyncio
async def test_send_test_without_to_email_defaults_to_requesting_user(
client: AsyncClient, db_session: AsyncSession, tenant_account, admin_headers, admin_user
):
seeded = await _seed(db_session, tenant_account)
template_id = str(seeded[0].id)
mock_redis = MagicMock()
mock_redis.incr = AsyncMock(return_value=1)
mock_redis.expire = AsyncMock()
mock_redis.aclose = AsyncMock()
mock_dispatch = AsyncMock()
with patch(
"src.apps.settings.router.get_redis", AsyncMock(return_value=mock_redis)
), patch("src.core.config.settings.SMTP_HOST", "smtp.example.com"), patch(
"src.apps.settings.services.email_template_service._dispatch", mock_dispatch
):
resp = await client.post(
f"/api/v1/settings/email-templates/{template_id}/test",
headers=admin_headers,
)
assert resp.status_code == 200, resp.text
assert resp.json()["message"] == f"Test email sent to {admin_user.email}"
assert mock_dispatch.call_args.kwargs["to"] == admin_user.email
@pytest.mark.asyncio
async def test_send_test_with_invalid_to_email_returns_422(
client: AsyncClient, db_session: AsyncSession, tenant_account, admin_headers
):
seeded = await _seed(db_session, tenant_account)
template_id = str(seeded[0].id)
resp = await client.post(
f"/api/v1/settings/email-templates/{template_id}/test",
json={"to_email": "not-an-email"},
headers=admin_headers,
)
assert resp.status_code == 422, resp.text
# ---------------------------------------------------------------------------
# M-1 security fix — Redis unreachable on test-send -> fail CLOSED (503),
# not fail-open/unlimited. Regression test for the rate limiter bypass found
# by Security/Tester review.
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_send_test_returns_503_when_redis_unreachable(
client: AsyncClient, db_session: AsyncSession, tenant_account, admin_headers
):
seeded = await _seed(db_session, tenant_account)
template_id = str(seeded[0].id)
with patch(
"src.apps.settings.router.get_redis", AsyncMock(return_value=None)
), patch("src.core.config.settings.SMTP_HOST", "smtp.example.com"), patch(
"src.apps.settings.services.email_template_service._dispatch", AsyncMock()
):
resp = await client.post(
f"/api/v1/settings/email-templates/{template_id}/test",
headers=admin_headers,
)
assert resp.status_code == 503, resp.text
# ---------------------------------------------------------------------------
# AC-15 / Risks & Notes — SMTP not configured -> 503
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_send_test_returns_503_when_smtp_not_configured(
client: AsyncClient, db_session: AsyncSession, tenant_account, admin_headers
):
seeded = await _seed(db_session, tenant_account)
template_id = str(seeded[0].id)
mock_redis = MagicMock()
mock_redis.incr = AsyncMock(return_value=1)
mock_redis.expire = AsyncMock()
mock_redis.aclose = AsyncMock()
with patch(
"src.apps.settings.router.get_redis", AsyncMock(return_value=mock_redis)
), patch("src.core.config.settings.SMTP_HOST", ""):
resp = await client.post(
f"/api/v1/settings/email-templates/{template_id}/test",
headers=admin_headers,
)
assert resp.status_code == 503, resp.text
# ---------------------------------------------------------------------------
# SEC-02 — injection payloads rejected with 422
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
@pytest.mark.parametrize(
"field,value",
[
("subject", "Hello {{ 7*7 }}"),
("body", "Hello {{ 7*7 }}"),
("subject", ""),
("body", ""),
("subject", "Click javascript:alert(1)"),
("body", "Click javascript:alert(1)"),
# H-1 regression: previously-unfiltered event-handler attributes.
("body", "