"""Tests for Task 5.10 — Template Engine (Message Templates). Covers: - MessageTemplate model: creation, str, unique constraint - render_template service: Jinja2 rendering with context, language fallback - render_template_subject: renders subject field - Services: create_message_template, update_message_template, delete_message_template - Selectors: list_message_templates, get_message_template - API: GET/POST /templates/messages/ - API: GET/PATCH/DELETE /templates/messages/{id}/ - API: POST /templates/messages/{id}/preview/ - API: unauthenticated access returns 401/403 - Notification system integration: DB template takes priority over in-code registry """ from __future__ import annotations import json import pytest from django.test import Client from simorgh.apps.platform_core.models import MessageChannel, MessageTemplate from simorgh.apps.platform_core.selectors import ( get_message_template, list_message_templates, ) from simorgh.apps.platform_core.services import ( PlatformCoreError, create_message_template, delete_message_template, render_template, render_template_subject, update_message_template, ) from tests.factories import AdminUserFactory, TenantFactory pytestmark = pytest.mark.django_db # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- SIMPLE_BODY = "Hello {{ name }}, your ticket #{{ ticket_id }} is ready." SIMPLE_SUBJECT = "Ticket #{{ ticket_id }} update" SIMPLE_CONTEXT = {"name": "Alice", "ticket_id": "TK-0042"} def _make_org_node(tenant): from simorgh.apps.organizations.services import create_node return create_node(tenant_id=tenant.pk, name="HQ-Template") def _make_tmpl(tenant, org_node, **kwargs): defaults = dict( tenant_id=tenant.pk, organization_node_id=org_node.pk, code="welcome", name="Welcome Email", channel=MessageChannel.EMAIL, subject=SIMPLE_SUBJECT, body=SIMPLE_BODY, language="en", ) defaults.update(kwargs) # need an actor if "actor" not in defaults: defaults["actor"] = AdminUserFactory() return create_message_template(**defaults) def _auth_client(user, tenant): from django.test import RequestFactory from simorgh.apps.memberships.models import Membership client = Client() client.force_login(user) # Attach tenant to session (middleware normally does this) session = client.session session["tenant_id"] = tenant.pk session.save() return client def _api_post(client, url, data, tenant, user): return client.post( f"/api/v1/{url}", data=json.dumps(data), content_type="application/json", HTTP_X_TENANT_ID=str(tenant.pk), ) def _api_get(client, url, tenant): return client.get( f"/api/v1/{url}", HTTP_X_TENANT_ID=str(tenant.pk), ) def _api_patch(client, url, data, tenant): return client.patch( f"/api/v1/{url}", data=json.dumps(data), content_type="application/json", HTTP_X_TENANT_ID=str(tenant.pk), ) def _api_delete(client, url, tenant): return client.delete( f"/api/v1/{url}", HTTP_X_TENANT_ID=str(tenant.pk), ) # --------------------------------------------------------------------------- # Model — unit tests # --------------------------------------------------------------------------- class TestMessageTemplateModel: def test_str_representation(self): tenant = TenantFactory() org_node = _make_org_node(tenant) actor = AdminUserFactory() tmpl = _make_tmpl(tenant, org_node, actor=actor) assert str(tmpl) == "welcome [email][en]" def test_default_channel_is_email(self): tenant = TenantFactory() org_node = _make_org_node(tenant) actor = AdminUserFactory() tmpl = _make_tmpl(tenant, org_node, actor=actor) assert tmpl.channel == MessageChannel.EMAIL def test_unique_constraint_tenant_code_language(self): from django.db import IntegrityError tenant = TenantFactory() org_node = _make_org_node(tenant) actor = AdminUserFactory() _make_tmpl(tenant, org_node, actor=actor, code="dup", language="en") with pytest.raises((IntegrityError, PlatformCoreError)): _make_tmpl(tenant, org_node, actor=actor, code="dup", language="en") def test_same_code_different_language_allowed(self): tenant = TenantFactory() org_node = _make_org_node(tenant) actor = AdminUserFactory() t1 = _make_tmpl(tenant, org_node, actor=actor, code="greet", language="en") t2 = _make_tmpl(tenant, org_node, actor=actor, code="greet", language="fa") assert t1.pk != t2.pk def test_message_channel_choices(self): assert MessageChannel.EMAIL == "email" assert MessageChannel.SMS == "sms" assert MessageChannel.PUSH == "push" assert MessageChannel.INAPP == "inapp" def test_variables_default_is_list(self): tenant = TenantFactory() org_node = _make_org_node(tenant) actor = AdminUserFactory() tmpl = _make_tmpl(tenant, org_node, actor=actor) assert isinstance(tmpl.variables, list) # --------------------------------------------------------------------------- # render_template — service # --------------------------------------------------------------------------- class TestRenderTemplate: def test_renders_body_with_context(self): tenant = TenantFactory() org_node = _make_org_node(tenant) actor = AdminUserFactory() _make_tmpl(tenant, org_node, actor=actor, code="order_confirm") rendered = render_template( "order_confirm", {"name": "Alice", "ticket_id": "TK-0042"}, tenant_id=tenant.pk, language="en", ) assert "Alice" in rendered assert "TK-0042" in rendered def test_language_fallback_to_en(self): """When no FA template exists, falls back to EN.""" tenant = TenantFactory() org_node = _make_org_node(tenant) actor = AdminUserFactory() _make_tmpl(tenant, org_node, actor=actor, code="fallback_test", language="en") rendered = render_template( "fallback_test", SIMPLE_CONTEXT, tenant_id=tenant.pk, language="fa", # FA doesn't exist ) assert "Alice" in rendered def test_raises_if_template_not_found(self): tenant = TenantFactory() with pytest.raises(PlatformCoreError, match="No active MessageTemplate"): render_template( "nonexistent_code", {}, tenant_id=tenant.pk, language="en", ) def test_raises_on_render_error(self): tenant = TenantFactory() org_node = _make_org_node(tenant) actor = AdminUserFactory() _make_tmpl( tenant, org_node, actor=actor, code="bad_syntax", body="{{ unclosed", ) with pytest.raises(PlatformCoreError, match="rendering failed"): render_template("bad_syntax", {}, tenant_id=tenant.pk, language="en") def test_inactive_template_not_used(self): tenant = TenantFactory() org_node = _make_org_node(tenant) actor = AdminUserFactory() tmpl = _make_tmpl(tenant, org_node, actor=actor, code="inactive_check") update_message_template(tmpl, actor=actor, is_active=False) with pytest.raises(PlatformCoreError, match="No active MessageTemplate"): render_template("inactive_check", {}, tenant_id=tenant.pk, language="en") class TestRenderTemplateSubject: def test_renders_subject(self): tenant = TenantFactory() org_node = _make_org_node(tenant) actor = AdminUserFactory() _make_tmpl(tenant, org_node, actor=actor, code="subj_test") result = render_template_subject( "subj_test", {"ticket_id": "TK-0001"}, tenant_id=tenant.pk, language="en" ) assert "TK-0001" in result def test_returns_empty_for_missing_template(self): tenant = TenantFactory() result = render_template_subject("no_code", {}, tenant_id=tenant.pk, language="en") assert result == "" def test_returns_empty_for_no_subject(self): tenant = TenantFactory() org_node = _make_org_node(tenant) actor = AdminUserFactory() _make_tmpl(tenant, org_node, actor=actor, code="no_subj", subject="") result = render_template_subject("no_subj", {}, tenant_id=tenant.pk, language="en") assert result == "" # --------------------------------------------------------------------------- # Services — CRUD # --------------------------------------------------------------------------- class TestMessageTemplateServices: def test_create_returns_instance(self): tenant = TenantFactory() org_node = _make_org_node(tenant) actor = AdminUserFactory() tmpl = _make_tmpl(tenant, org_node, actor=actor, code="svc_create") assert tmpl.pk is not None assert tmpl.tenant_id == tenant.pk def test_update_name(self): tenant = TenantFactory() org_node = _make_org_node(tenant) actor = AdminUserFactory() tmpl = _make_tmpl(tenant, org_node, actor=actor, code="svc_update") updated = update_message_template(tmpl, actor=actor, name="New Name") assert updated.name == "New Name" def test_update_is_active(self): tenant = TenantFactory() org_node = _make_org_node(tenant) actor = AdminUserFactory() tmpl = _make_tmpl(tenant, org_node, actor=actor, code="svc_active") updated = update_message_template(tmpl, actor=actor, is_active=False) assert updated.is_active is False def test_update_invalid_field_raises(self): tenant = TenantFactory() org_node = _make_org_node(tenant) actor = AdminUserFactory() tmpl = _make_tmpl(tenant, org_node, actor=actor, code="svc_inv_field") with pytest.raises(PlatformCoreError, match="Cannot update field"): update_message_template(tmpl, actor=actor, unknown_field="x") def test_delete_removes_record(self): tenant = TenantFactory() org_node = _make_org_node(tenant) actor = AdminUserFactory() tmpl = _make_tmpl(tenant, org_node, actor=actor, code="svc_delete") tmpl_id = tmpl.pk delete_message_template(tmpl, actor=actor) assert not MessageTemplate.objects.filter(pk=tmpl_id).exists() def test_variables_stored_as_list(self): tenant = TenantFactory() org_node = _make_org_node(tenant) actor = AdminUserFactory() vars_spec = [{"name": "user", "type": "str", "description": "Recipient name"}] tmpl = _make_tmpl(tenant, org_node, actor=actor, code="svc_vars", variables=vars_spec) tmpl.refresh_from_db() assert tmpl.variables == vars_spec # --------------------------------------------------------------------------- # Selectors # --------------------------------------------------------------------------- class TestMessageTemplateSelectors: def test_list_returns_tenant_scoped(self): t1 = TenantFactory() t2 = TenantFactory() o1 = _make_org_node(t1) o2 = _make_org_node(t2) a = AdminUserFactory() _make_tmpl(t1, o1, actor=a, code="sel_scope_1") _make_tmpl(t2, o2, actor=a, code="sel_scope_2") results = list(list_message_templates(t1.pk)) assert all(r.tenant_id == t1.pk for r in results) assert len(results) == 1 def test_list_filter_by_channel(self): tenant = TenantFactory() org_node = _make_org_node(tenant) actor = AdminUserFactory() _make_tmpl(tenant, org_node, actor=actor, code="sel_ch_email", channel="email") _make_tmpl(tenant, org_node, actor=actor, code="sel_ch_sms", channel="sms") emails = list(list_message_templates(tenant.pk, channel="email")) assert all(t.channel == "email" for t in emails) assert len(emails) == 1 def test_list_filter_by_language(self): tenant = TenantFactory() org_node = _make_org_node(tenant) actor = AdminUserFactory() _make_tmpl(tenant, org_node, actor=actor, code="sel_lang_en", language="en") _make_tmpl(tenant, org_node, actor=actor, code="sel_lang_fa", language="fa") fa_only = list(list_message_templates(tenant.pk, language="fa")) assert len(fa_only) == 1 assert fa_only[0].language == "fa" def test_list_filter_by_is_active(self): tenant = TenantFactory() org_node = _make_org_node(tenant) actor = AdminUserFactory() tmpl = _make_tmpl(tenant, org_node, actor=actor, code="sel_inactive") update_message_template(tmpl, actor=actor, is_active=False) active = list(list_message_templates(tenant.pk, is_active=True)) assert all(t.is_active for t in active) def test_get_returns_correct_template(self): tenant = TenantFactory() org_node = _make_org_node(tenant) actor = AdminUserFactory() tmpl = _make_tmpl(tenant, org_node, actor=actor, code="sel_get") found = get_message_template(tmpl.pk, tenant_id=tenant.pk) assert found is not None assert found.pk == tmpl.pk def test_get_returns_none_for_wrong_tenant(self): t1 = TenantFactory() t2 = TenantFactory() o1 = _make_org_node(t1) actor = AdminUserFactory() tmpl = _make_tmpl(t1, o1, actor=actor, code="sel_wrong_tenant") found = get_message_template(tmpl.pk, tenant_id=t2.pk) assert found is None def test_get_returns_none_for_missing(self): tenant = TenantFactory() found = get_message_template(999999, tenant_id=tenant.pk) assert found is None # --------------------------------------------------------------------------- # API — full-stack tests # --------------------------------------------------------------------------- def _make_tenant_with_user(): from simorgh.apps.memberships.models import Membership from simorgh.apps.iam.models import Role tenant = TenantFactory() org_node = _make_org_node(tenant) user = AdminUserFactory() role = Role.objects.create(tenant=tenant, code="admin-tmpl", name="Admin Template") mem = Membership.objects.create( tenant=tenant, organization_node=org_node, role=role, status="active", ) mem.users.add(user) return tenant, org_node, user class TestMessageTemplateAPI: def test_list_unauthenticated_returns_403(self): tenant = TenantFactory() client = Client() resp = client.get( "/api/v1/templates/messages/", HTTP_X_TENANT_ID=str(tenant.pk), ) assert resp.status_code in (401, 403) def test_list_empty(self): tenant, org_node, user = _make_tenant_with_user() client = Client() client.force_login(user) resp = client.get( "/api/v1/templates/messages/", HTTP_X_TENANT_ID=str(tenant.pk), ) assert resp.status_code == 200 assert resp.json() == [] def test_create_template(self): tenant, org_node, user = _make_tenant_with_user() client = Client() client.force_login(user) payload = { "code": "api_create", "name": "API Created", "channel": "email", "subject": "Hello {{ name }}", "body": "Dear {{ name }}, welcome!", "language": "en", } resp = client.post( "/api/v1/templates/messages/", data=json.dumps(payload), content_type="application/json", HTTP_X_TENANT_ID=str(tenant.pk), ) assert resp.status_code == 201 data = resp.json() assert data["code"] == "api_create" assert data["channel"] == "email" def test_create_missing_required_fields(self): tenant, org_node, user = _make_tenant_with_user() client = Client() client.force_login(user) resp = client.post( "/api/v1/templates/messages/", data=json.dumps({"channel": "email"}), content_type="application/json", HTTP_X_TENANT_ID=str(tenant.pk), ) assert resp.status_code == 400 def test_list_returns_created_templates(self): tenant, org_node, user = _make_tenant_with_user() actor = user _make_tmpl(tenant, org_node, actor=actor, code="list_test_a") _make_tmpl(tenant, org_node, actor=actor, code="list_test_b") client = Client() client.force_login(user) resp = client.get( "/api/v1/templates/messages/", HTTP_X_TENANT_ID=str(tenant.pk), ) assert resp.status_code == 200 codes = [t["code"] for t in resp.json()] assert "list_test_a" in codes assert "list_test_b" in codes def test_list_filter_by_channel(self): tenant, org_node, user = _make_tenant_with_user() actor = user _make_tmpl(tenant, org_node, actor=actor, code="fl_email", channel="email") _make_tmpl(tenant, org_node, actor=actor, code="fl_sms", channel="sms") client = Client() client.force_login(user) resp = client.get( "/api/v1/templates/messages/?channel=sms", HTTP_X_TENANT_ID=str(tenant.pk), ) assert resp.status_code == 200 assert all(t["channel"] == "sms" for t in resp.json()) def test_retrieve_template(self): tenant, org_node, user = _make_tenant_with_user() actor = user tmpl = _make_tmpl(tenant, org_node, actor=actor, code="api_get") client = Client() client.force_login(user) resp = client.get( f"/api/v1/templates/messages/{tmpl.pk}/", HTTP_X_TENANT_ID=str(tenant.pk), ) assert resp.status_code == 200 assert resp.json()["id"] == tmpl.pk def test_retrieve_wrong_tenant_returns_404(self): t1, o1, u1 = _make_tenant_with_user() t2, o2, u2 = _make_tenant_with_user() tmpl = _make_tmpl(t1, o1, actor=u1, code="api_404_check") client = Client() client.force_login(u2) resp = client.get( f"/api/v1/templates/messages/{tmpl.pk}/", HTTP_X_TENANT_ID=str(t2.pk), ) assert resp.status_code == 404 def test_patch_template(self): tenant, org_node, user = _make_tenant_with_user() actor = user tmpl = _make_tmpl(tenant, org_node, actor=actor, code="api_patch") client = Client() client.force_login(user) resp = client.patch( f"/api/v1/templates/messages/{tmpl.pk}/", data=json.dumps({"name": "Updated Name", "is_active": False}), content_type="application/json", HTTP_X_TENANT_ID=str(tenant.pk), ) assert resp.status_code == 200 data = resp.json() assert data["name"] == "Updated Name" assert data["is_active"] is False def test_delete_template(self): tenant, org_node, user = _make_tenant_with_user() actor = user tmpl = _make_tmpl(tenant, org_node, actor=actor, code="api_delete") client = Client() client.force_login(user) resp = client.delete( f"/api/v1/templates/messages/{tmpl.pk}/", HTTP_X_TENANT_ID=str(tenant.pk), ) assert resp.status_code == 204 assert not MessageTemplate.objects.filter(pk=tmpl.pk).exists() def test_preview_renders_body_and_subject(self): tenant, org_node, user = _make_tenant_with_user() actor = user tmpl = _make_tmpl(tenant, org_node, actor=actor, code="api_preview") client = Client() client.force_login(user) resp = client.post( f"/api/v1/templates/messages/{tmpl.pk}/preview/", data=json.dumps({"context": SIMPLE_CONTEXT}), content_type="application/json", HTTP_X_TENANT_ID=str(tenant.pk), ) assert resp.status_code == 200 data = resp.json() assert "Alice" in data["body"] assert "TK-0042" in data["subject"] def test_preview_empty_context(self): tenant, org_node, user = _make_tenant_with_user() actor = user tmpl = _make_tmpl( tenant, org_node, actor=actor, code="api_prev_empty", body="Hello world" ) client = Client() client.force_login(user) resp = client.post( f"/api/v1/templates/messages/{tmpl.pk}/preview/", data=json.dumps({}), content_type="application/json", HTTP_X_TENANT_ID=str(tenant.pk), ) assert resp.status_code == 200 assert resp.json()["body"] == "Hello world" def test_preview_bad_template_returns_400(self): tenant, org_node, user = _make_tenant_with_user() actor = user tmpl = _make_tmpl( tenant, org_node, actor=actor, code="api_prev_bad", body="{{ unclosed" ) client = Client() client.force_login(user) resp = client.post( f"/api/v1/templates/messages/{tmpl.pk}/preview/", data=json.dumps({"context": {}}), content_type="application/json", HTTP_X_TENANT_ID=str(tenant.pk), ) assert resp.status_code == 400 def test_preview_unauthenticated_returns_403(self): tenant, org_node, user = _make_tenant_with_user() tmpl = _make_tmpl(tenant, org_node, actor=user, code="api_prev_unauth") client = Client() resp = client.post( f"/api/v1/templates/messages/{tmpl.pk}/preview/", data=json.dumps({"context": {}}), content_type="application/json", HTTP_X_TENANT_ID=str(tenant.pk), ) assert resp.status_code in (401, 403) # --------------------------------------------------------------------------- # Notification system integration # --------------------------------------------------------------------------- class TestNotificationIntegration: def test_db_template_used_when_available(self): """MessageTemplate in DB overrides the in-code template for 'inapp' channel.""" from simorgh.apps.notifications.templates import ( NotificationTemplate, register_template, ) from simorgh.apps.notifications.services import dispatch from simorgh.apps.memberships.models import Membership from simorgh.apps.iam.models import Role tenant = TenantFactory() org_node = _make_org_node(tenant) user = AdminUserFactory() role = Role.objects.create(tenant=tenant, code="admin-notif-tmpl", name="Admin Notif") mem = Membership.objects.create( tenant=tenant, organization_node=org_node, role=role, status="active", ) mem.users.add(user) # Register an in-code template kind = "test.integration.msg_tmpl" try: register_template( NotificationTemplate( kind=kind, title_key="Fallback Title", body_key="Fallback Body", default_channels=("inapp",), ) ) except ValueError: pass # already registered in another test run # Create a DB override template create_message_template( tenant_id=tenant.pk, organization_node_id=org_node.pk, code=kind, name="DB Override", channel=MessageChannel.INAPP, subject="", body="DB body for {{ user_name }}", language="en", actor=user, ) from unittest.mock import patch with patch("simorgh.apps.notifications.services.deliver") as mock_deliver: mock_deliver.return_value = None from simorgh.apps.notifications.models import Notification notifications = dispatch( kind, recipients=[user], context={"user_name": "Bob"}, tenant_id=tenant.pk, organization_node_id=org_node.pk, ) # The notification body should come from the DB template assert len(notifications) >= 1 body = notifications[0].body assert "DB body for Bob" in body