feat(07-01): localized emails, locale-aware templates API, language preference endpoint

- email.py: send_invite_email() adds language param (en/es/pt), sends localized subject+body
- templates.py: list_templates()/get_template() accept ?locale= param, merge translations on response
- portal.py: PATCH /api/portal/users/me/language endpoint persists language preference
- portal.py: /api/portal/auth/verify response includes user.language field
- portal.py: AuthVerifyResponse adds language field (default 'en')
- test_portal_auth.py: fix _make_user mock to set language='en' (auto-fix Rule 1)
- test_language_preference.py: 4 integration tests for language preference endpoint
- test_templates_i18n.py: 5 integration tests for locale-aware templates (all passing)
This commit is contained in:
2026-03-25 16:27:14 -06:00
parent 7a3a4f0fdd
commit 9654982433
6 changed files with 648 additions and 38 deletions

View File

@@ -0,0 +1,186 @@
"""
Integration tests for the language preference PATCH endpoint.
Tests:
- PATCH /api/portal/users/me/language with valid language returns 200
- PATCH with unsupported language returns 400
- PATCH to "pt" then GET /api/portal/auth/verify includes language="pt"
- PATCH without auth returns 401
Uses the same pattern as existing integration tests:
- Session override via app.dependency_overrides
- X-Portal-User-Id / X-Portal-User-Role header injection for auth
- db_session fixture from tests/conftest.py (Alembic migrations applied)
"""
from __future__ import annotations
import uuid
import bcrypt
import pytest
import pytest_asyncio
from fastapi import FastAPI
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession
from shared.api.portal import portal_router
from shared.db import get_session
from shared.models.auth import PortalUser
# ---------------------------------------------------------------------------
# App factory
# ---------------------------------------------------------------------------
def make_app(session: AsyncSession) -> FastAPI:
"""Build a minimal FastAPI test app with portal router."""
app = FastAPI()
app.include_router(portal_router)
async def override_get_session(): # type: ignore[return]
yield session
app.dependency_overrides[get_session] = override_get_session
return app
# ---------------------------------------------------------------------------
# Header helpers
# ---------------------------------------------------------------------------
def auth_headers(user_id: uuid.UUID, role: str = "customer_admin") -> dict[str, str]:
return {
"X-Portal-User-Id": str(user_id),
"X-Portal-User-Role": role,
}
# ---------------------------------------------------------------------------
# DB helpers
# ---------------------------------------------------------------------------
async def _create_user(
session: AsyncSession,
role: str = "customer_admin",
language: str = "en",
) -> PortalUser:
suffix = uuid.uuid4().hex[:8]
hashed = bcrypt.hashpw(b"testpassword123", bcrypt.gensalt()).decode()
user = PortalUser(
id=uuid.uuid4(),
email=f"langtest-{suffix}@example.com",
hashed_password=hashed,
name=f"Language Test User {suffix}",
role=role,
language=language,
)
session.add(user)
await session.flush()
return user
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest_asyncio.fixture
async def lang_client(db_session: AsyncSession) -> AsyncClient:
"""HTTP client with portal router mounted."""
app = make_app(db_session)
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
yield c
@pytest_asyncio.fixture
async def lang_user(db_session: AsyncSession) -> PortalUser:
"""Create a customer_admin user with default language 'en'."""
user = await _create_user(db_session, role="customer_admin", language="en")
await db_session.commit()
return user
# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_patch_language_valid(
lang_client: AsyncClient,
lang_user: PortalUser,
) -> None:
"""PATCH /api/portal/users/me/language with valid language returns 200 and new language."""
response = await lang_client.patch(
"/api/portal/users/me/language",
json={"language": "es"},
headers=auth_headers(lang_user.id),
)
assert response.status_code == 200
data = response.json()
assert data["language"] == "es"
@pytest.mark.asyncio
async def test_patch_language_invalid(
lang_client: AsyncClient,
lang_user: PortalUser,
) -> None:
"""PATCH with unsupported language 'fr' returns 400."""
response = await lang_client.patch(
"/api/portal/users/me/language",
json={"language": "fr"},
headers=auth_headers(lang_user.id),
)
assert response.status_code == 400
assert "fr" in response.json()["detail"].lower() or "unsupported" in response.json()["detail"].lower()
@pytest.mark.asyncio
async def test_patch_language_persists(
lang_client: AsyncClient,
lang_user: PortalUser,
db_session: AsyncSession,
) -> None:
"""PATCH to 'pt', then GET /api/portal/auth/verify includes language='pt'."""
# First PATCH to "pt"
patch_response = await lang_client.patch(
"/api/portal/users/me/language",
json={"language": "pt"},
headers=auth_headers(lang_user.id),
)
assert patch_response.status_code == 200
assert patch_response.json()["language"] == "pt"
# Verify via /auth/verify — need to pass email+password
# Re-fetch user to get credentials, then call auth/verify
verify_response = await lang_client.post(
"/api/portal/auth/verify",
json={"email": lang_user.email, "password": "testpassword123"},
)
assert verify_response.status_code == 200
verify_data = verify_response.json()
assert verify_data["language"] == "pt", (
f"Expected language='pt' in auth/verify response, got: {verify_data.get('language')!r}"
)
@pytest.mark.asyncio
async def test_patch_language_unauthenticated(
lang_client: AsyncClient,
) -> None:
"""PATCH without auth headers returns 401 or 422 (missing required headers)."""
response = await lang_client.patch(
"/api/portal/users/me/language",
json={"language": "es"},
)
# FastAPI raises 422 when required headers are missing entirely (before auth guard runs).
# Both 401 and 422 are acceptable rejections of unauthenticated requests.
assert response.status_code in (401, 422)

View File

@@ -0,0 +1,226 @@
"""
Integration tests for locale-aware template API endpoints.
Tests:
- GET /api/portal/templates (no locale) returns English fields
- GET /api/portal/templates?locale=es returns Spanish-translated fields
- GET /api/portal/templates?locale=pt returns Portuguese-translated fields
- GET /api/portal/templates?locale=fr falls back to English
- Translated fields overlay English base, English values still in DB
Uses the same pattern as existing integration tests:
- Session override via app.dependency_overrides
- X-Portal-User-Id / X-Portal-User-Role header injection
- db_session fixture from tests/conftest.py (Alembic migrations applied)
"""
from __future__ import annotations
import uuid
import pytest
import pytest_asyncio
from fastapi import FastAPI
from httpx import ASGITransport, AsyncClient
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from shared.api.portal import portal_router
from shared.api.templates import templates_router
from shared.db import get_session
from shared.models.tenant import AgentTemplate
# ---------------------------------------------------------------------------
# App factory
# ---------------------------------------------------------------------------
def make_app(session: AsyncSession) -> FastAPI:
"""Build a minimal FastAPI test app with portal + templates routers."""
app = FastAPI()
app.include_router(portal_router)
app.include_router(templates_router)
async def override_get_session(): # type: ignore[return]
yield session
app.dependency_overrides[get_session] = override_get_session
return app
# ---------------------------------------------------------------------------
# Header helpers
# ---------------------------------------------------------------------------
def platform_admin_headers(user_id: uuid.UUID) -> dict[str, str]:
return {
"X-Portal-User-Id": str(user_id),
"X-Portal-User-Role": "platform_admin",
}
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest_asyncio.fixture
async def i18n_client(db_session: AsyncSession) -> AsyncClient:
"""HTTP client with portal + templates router mounted."""
app = make_app(db_session)
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
yield c
@pytest_asyncio.fixture
def admin_user_id() -> uuid.UUID:
"""Fixed UUID for a fake platform_admin — no DB row needed for header-based auth."""
return uuid.UUID("00000000-0000-0000-0000-000000000099")
# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_list_templates_default_locale(
i18n_client: AsyncClient,
admin_user_id: uuid.UUID,
) -> None:
"""GET /api/portal/templates (no locale param) returns English fields."""
headers = platform_admin_headers(admin_user_id)
response = await i18n_client.get("/api/portal/templates", headers=headers)
assert response.status_code == 200
templates = response.json()
assert len(templates) >= 7
# All returned names should be English (Customer Support Rep is the first by sort_order)
names = {t["name"] for t in templates}
assert "Customer Support Rep" in names, f"Expected English template name, got: {names}"
@pytest.mark.asyncio
async def test_list_templates_spanish(
i18n_client: AsyncClient,
admin_user_id: uuid.UUID,
) -> None:
"""GET /api/portal/templates?locale=es returns Spanish-translated name/description/persona."""
headers = platform_admin_headers(admin_user_id)
response = await i18n_client.get("/api/portal/templates?locale=es", headers=headers)
assert response.status_code == 200
templates = response.json()
assert len(templates) >= 7
# Find the Customer Support Rep template (ID: 000...001)
support_tmpl = next(
(t for t in templates if "soporte" in t["name"].lower() or "support" in t["name"].lower()),
None,
)
assert support_tmpl is not None, f"Customer support template not found in: {[t['name'] for t in templates]}"
# Spanish name should differ from English
assert support_tmpl["name"] != "Customer Support Rep", (
f"Expected Spanish translation, got English name: {support_tmpl['name']!r}"
)
# Spanish description should be present and non-empty
assert len(support_tmpl["description"]) > 10
@pytest.mark.asyncio
async def test_list_templates_portuguese(
i18n_client: AsyncClient,
admin_user_id: uuid.UUID,
) -> None:
"""GET /api/portal/templates?locale=pt returns Portuguese-translated fields."""
headers = platform_admin_headers(admin_user_id)
response = await i18n_client.get("/api/portal/templates?locale=pt", headers=headers)
assert response.status_code == 200
templates = response.json()
assert len(templates) >= 7
# Find the Customer Support Rep template
support_tmpl = next(
(t for t in templates if "suporte" in t["name"].lower() or "support" in t["name"].lower()),
None,
)
assert support_tmpl is not None, f"Customer support template not found in: {[t['name'] for t in templates]}"
# Portuguese name should differ from English
assert support_tmpl["name"] != "Customer Support Rep", (
f"Expected Portuguese translation, got English name: {support_tmpl['name']!r}"
)
# Portuguese description should be present and non-empty
assert len(support_tmpl["description"]) > 10
@pytest.mark.asyncio
async def test_list_templates_unsupported_locale(
i18n_client: AsyncClient,
admin_user_id: uuid.UUID,
) -> None:
"""GET /api/portal/templates?locale=fr falls back to English."""
headers = platform_admin_headers(admin_user_id)
response = await i18n_client.get("/api/portal/templates?locale=fr", headers=headers)
assert response.status_code == 200
templates = response.json()
assert len(templates) >= 7
# Names should be English (fallback)
names = {t["name"] for t in templates}
assert "Customer Support Rep" in names, (
f"Expected English fallback for unsupported locale 'fr', got names: {names}"
)
@pytest.mark.asyncio
async def test_template_translations_overlay(
i18n_client: AsyncClient,
admin_user_id: uuid.UUID,
db_session: AsyncSession,
) -> None:
"""Translated fields overlay English, English base fields still in DB."""
headers = platform_admin_headers(admin_user_id)
# Get Spanish-translated templates
es_response = await i18n_client.get("/api/portal/templates?locale=es", headers=headers)
assert es_response.status_code == 200
es_templates = es_response.json()
# Get English templates (default)
en_response = await i18n_client.get("/api/portal/templates", headers=headers)
assert en_response.status_code == 200
en_templates = en_response.json()
# Find the support template in both
es_support = next((t for t in es_templates if "soporte" in t["name"].lower()), None)
en_support = next((t for t in en_templates if t["name"] == "Customer Support Rep"), None)
assert es_support is not None, "Spanish support template not found"
assert en_support is not None, "English support template not found"
# They should share the same template ID
assert es_support["id"] == en_support["id"], "Template IDs should match across locales"
# Names should differ between locales
assert es_support["name"] != en_support["name"], (
"Spanish and English names should differ for Customer Support Rep template"
)
# English base values must still be present in DB (not overwritten)
result = await db_session.execute(
select(AgentTemplate).where(
AgentTemplate.id == uuid.UUID(en_support["id"])
)
)
tmpl_orm = result.scalar_one_or_none()
assert tmpl_orm is not None
assert tmpl_orm.name == "Customer Support Rep", (
f"DB English name should be unchanged, got: {tmpl_orm.name!r}"
)

View File

@@ -29,6 +29,7 @@ def _make_user(role: str, email: str = "test@example.com") -> PortalUser:
user.email = email
user.name = "Test User"
user.role = role
user.language = "en"
# Real bcrypt hash for password "testpassword"
user.hashed_password = bcrypt.hashpw(b"testpassword", bcrypt.gensalt()).decode()
user.created_at = datetime.now(tz=timezone.utc)