"""8H contact notification ownership freeze.""" from __future__ import annotations from datetime import UTC, datetime from pathlib import Path from src.domain.connectors.invoke_capabilities import CAPABILITY_CONVERSION_PATTERN, FORMS_FAMILY_PLAN from src.domain.connectors.schemas import InvokeEmailSendRequest from src.domain.forms.actions import ( FormCapabilityInvokeResult, FormPostSubmitContext, FormPostSubmitPlan, RecordingCapabilityInvoker, execute_post_submit_actions, ) from src.domain.forms.canonical import contact_block_compatible_definition from src.domain.forms.composition import _contact_plan, _quote_plan from src.domain.forms.contact_notifications import ( ADMIN_TEMPLATE_KEY, CONFIRMATION_TEMPLATE_KEY, CONTACT_NOTIFICATION_OWNERSHIP, CONTACT_NOTIFICATION_TOPOLOGIES, LEGACY_ADMIN_IDEMPOTENCY_PREFIX, expected_forms_email_send_count, forms_contact_notification_actions, ) from src.domain.forms.email_template import RecordingEmailTemplateRenderer from src.domain.forms.persistence import FORMS_PERSISTENCE from src.domain.forms.runtime import accept_form_submission _FIXED = datetime(2026, 9, 14, 2, 0, tzinfo=UTC) _FSB = "fsb_" + ("d" * 32) _CTX = FormPostSubmitContext(project_id="prj_8h") NOTIFY = Path("seeds/function_library/contact-capture-notify.yaml") CONTACT_JS = Path("src/domain/codegen/runtime/p2hs-contact.mjs") def _accept(values): result = accept_form_submission( contact_block_compatible_definition(), values, clock=lambda: _FIXED, submission_id_factory=lambda: _FSB, ) assert result.accepted is True return result.submission def _run(plan, values, renderer=None): invoker = RecordingCapabilityInvoker() result = execute_post_submit_actions( _accept(values), plan, context=_CTX, invoker=invoker, definition=contact_block_compatible_definition(), email_template_renderer=renderer, ) return result, invoker def test_8h_ownership_freeze() -> None: own = CONTACT_NOTIFICATION_OWNERSHIP assert own["slice_id"] == "8h_contact_notification_path_convergence" assert own["new_project_owner"] == "forms_post_submit" assert own["forms_writes_contact_submission"] is False assert own["pack_notify_on_forms_persist"] is False assert own["legacy_generation"] == "deprecated_for_new_projects" assert own["legacy_runtime"] == "compatibility_only" assert own["fallback_on_forms"] is False assert FORMS_PERSISTENCE["contact_submission_is_canonical"] is False forms = CONTACT_NOTIFICATION_TOPOLOGIES["forms_backed_contact"] assert forms["admin_owner"] == "forms_post_submit" assert forms["confirmation_owner"] == "forms_post_submit" assert forms["pack_notify"] is False legacy = CONTACT_NOTIFICATION_TOPOLOGIES["legacy_api_contact"] assert legacy["admin_owner"] == "contact-capture-notify" assert legacy["confirmation_owner"] == "contact-capture-notify" assert legacy["status"] == "compatibility_only" assert legacy["generation"] == "deprecated_for_new_projects" assert LEGACY_ADMIN_IDEMPOTENCY_PREFIX in legacy["admin_dedupe"] entity = CONTACT_NOTIFICATION_TOPOLOGIES["entity_backed_contact_capture"] assert entity["admin_owner"] == "contact-capture-notify" assert CAPABILITY_CONVERSION_PATTERN["phase_8h_status"] == "complete" assert CAPABILITY_CONVERSION_PATTERN["next_program_slice"] == ( "9_app_block_modernization" ) assert FORMS_FAMILY_PLAN["next_slice"] == "9_app_block_modernization" def test_admin_only_one_render_one_send() -> None: config = {"notify_to_email": "ops@example.com"} assert expected_forms_email_send_count(config) == 1 renderer = RecordingEmailTemplateRenderer() result, invoker = _run( FormPostSubmitPlan(actions=forms_contact_notification_actions(config)), {"name": "Ada", "email": "ada@example.com", "message": "Hi"}, renderer, ) assert [item.status for item in result.actions] == ["succeeded"] assert len(renderer.calls) == 1 assert renderer.calls[0]["template_key"] == ADMIN_TEMPLATE_KEY assert len(invoker.calls) == 1 assert invoker.calls[0]["capability"] == "email.send" assert "template_key" not in invoker.calls[0]["payload"] def test_confirmation_only_one_render_one_send() -> None: config = {"send_confirmation_email": True} assert expected_forms_email_send_count(config) == 1 renderer = RecordingEmailTemplateRenderer() result, invoker = _run( FormPostSubmitPlan(actions=forms_contact_notification_actions(config)), {"name": "Ada", "email": "ada@example.com", "message": "Hi"}, renderer, ) assert len(renderer.calls) == 1 assert renderer.calls[0]["template_key"] == CONFIRMATION_TEMPLATE_KEY assert len(invoker.calls) == 1 assert invoker.calls[0]["payload"]["to"] == "ada@example.com" def test_both_notifications_two_renders_two_sends() -> None: config = {"notify_to_email": "ops@example.com", "send_confirmation_email": True} assert expected_forms_email_send_count(config) == 2 renderer = RecordingEmailTemplateRenderer() result, invoker = _run( _contact_plan(config), {"name": "Ada", "email": "ada@example.com", "message": "Hi"}, renderer, ) email_calls = [call for call in invoker.calls if call["capability"] == "email.send"] assert len(renderer.calls) == 2 assert [call["template_key"] for call in renderer.calls] == [ ADMIN_TEMPLATE_KEY, CONFIRMATION_TEMPLATE_KEY, ] assert len(email_calls) == 2 assert {item.status for item in result.actions if item.action_type == "email.send"} == { "succeeded" } def test_neither_notification_zero_sends() -> None: config = {} assert expected_forms_email_send_count(config) == 0 renderer = RecordingEmailTemplateRenderer() _, invoker = _run(_contact_plan(config), {"name": "Ada", "email": "ada@example.com", "message": "Hi"}, renderer) assert renderer.calls == [] assert [call["capability"] for call in invoker.calls] == ["crm.contact.upsert"] def test_omitted_action_is_suppression_not_template_enabled() -> None: actions = forms_contact_notification_actions({"notify_to_email": "ops@example.com"}) assert [action.key for action in actions] == ["owner_notify"] renderer = RecordingEmailTemplateRenderer() _, invoker = _run(FormPostSubmitPlan(actions=actions), {"name": "Ada", "email": "ada@example.com", "message": "Hi"}, renderer) assert len(invoker.calls) == 1 assert renderer.calls[0]["template_key"] == ADMIN_TEMPLATE_KEY def test_render_failure_does_not_send_and_has_no_forms_fallback() -> None: renderer = RecordingEmailTemplateRenderer(error_code="missing_required_variable") result, invoker = _run( FormPostSubmitPlan( actions=forms_contact_notification_actions({"notify_to_email": "ops@example.com"}) ), {"name": "Ada", "email": "ada@example.com", "message": "Hi"}, renderer, ) assert result.actions[0].status == "failed" assert result.actions[0].error_code == "missing_required_variable" assert invoker.calls == [] def test_send_failure_is_delivery_not_rerender() -> None: renderer = RecordingEmailTemplateRenderer() invoker = RecordingCapabilityInvoker( { "email.send": FormCapabilityInvokeResult( status="failed", error_code="email_unbound", ) } ) result = execute_post_submit_actions( _accept({"name": "Ada", "email": "ada@example.com", "message": "Hi"}), FormPostSubmitPlan( actions=forms_contact_notification_actions({"notify_to_email": "ops@example.com"}) ), context=_CTX, invoker=invoker, definition=contact_block_compatible_definition(), email_template_renderer=renderer, ) assert len(renderer.calls) == 1 assert result.actions[0].status == "failed" assert result.actions[0].error_code == "email_unbound" assert invoker.calls[0]["payload"]["subject"] == "Rendered subject" assert set(invoker.calls[0]["payload"]) <= set(InvokeEmailSendRequest.model_fields) def test_forms_and_pack_do_not_share_a_persist_event() -> None: assert CONTACT_NOTIFICATION_OWNERSHIP["forms_writes_contact_submission"] is False plan = _contact_plan({"notify_to_email": "ops@example.com", "send_confirmation_email": True}) assert all(action.action_type != "contact-capture-notify" for action in plan.actions) yaml_text = NOTIFY.read_text(encoding="utf-8") assert "record.created" in yaml_text assert "contact_submission" in yaml_text assert "contact_capture.admin_notify" in yaml_text quote = _quote_plan({"notify_to_email": "ops@example.com"}) assert quote.actions[0].template.template_key == ADMIN_TEMPLATE_KEY def test_legacy_adapter_uses_canonical_render_and_pack_idempotency() -> None: source = CONTACT_JS.read_text(encoding="utf-8") assert "contact_capture.admin_notify" in source assert "/email-templates/" in source assert "contact-notify:" in source assert "request_type:" in source assert "email.send" in source assert "We've received your message" not in source assert CONTACT_NOTIFICATION_OWNERSHIP["legacy_admin_fallback"] == "caller_resilience"