""" Standalone test for the SendGrid email flow. Tests both EmailChannel.is_configured() and EmailChannel.send() directly. Usage: cd cuatrolabs-notification-ms python test_email_flow.py --to recipient@example.com """ import asyncio import argparse import sys # --------------------------------------------------------------------------- # Config — edit these or pass via CLI args # --------------------------------------------------------------------------- TEST_CONFIG = { "enabled": True, "api_key": "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", "from_email": "no-reply@example.com", "from_name": "Cuatro Notifications", "reply_to": "support@example.com", } TEST_TEMPLATE_DATA = { "subject": "Test Email — Cuatro Notification MS", "body": "Hello {{ name }},\n\nThis is a test email from the Cuatro notification service.\n\nRegards,\nCuatro Team", "html_body": "

Hello {{ name }},

This is a test email from the Cuatro notification service.

Regards,
Cuatro Team

", "name": "Test User", } # --------------------------------------------------------------------------- async def run_test(recipient: str): # Patch sys.path so we can import app modules without installing the package import os sys.path.insert(0, os.path.dirname(__file__)) # Minimal stub for app.core.logging so we don't need the full app stack import types import logging logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s") core_mod = types.ModuleType("app.core") core_logging_mod = types.ModuleType("app.core.logging") core_logging_mod.get_logger = lambda name: logging.getLogger(name) sys.modules.setdefault("app", types.ModuleType("app")) sys.modules["app.core"] = core_mod sys.modules["app.core.logging"] = core_logging_mod from app.channels.email import EmailChannel print("\n=== SendGrid Email Flow Test ===\n") # 1. is_configured check configured = EmailChannel.is_configured(TEST_CONFIG) print(f"[1] is_configured: {'✓ PASS' if configured else '✗ FAIL'}") if not configured: print(" Channel config is missing required fields (enabled, api_key, from_email).") return # 2. Send test email print(f"[2] Sending email to: {recipient}") success, message, provider_id = await EmailChannel.send( recipient=recipient, template_name="test_notification", template_data=TEST_TEMPLATE_DATA, channel_config=TEST_CONFIG, ) status = "✓ PASS" if success else "✗ FAIL" print(f" Result : {status}") print(f" Message : {message}") if provider_id: print(f" Provider ID: {provider_id}") # 3. Misconfigured channel test (should fail gracefully) print("\n[3] Testing misconfigured channel (no api_key)...") bad_config = {"enabled": True, "from_email": "no-reply@example.com"} configured_bad = EmailChannel.is_configured(bad_config) print(f" is_configured with missing api_key: {'✗ FAIL (should be False)' if configured_bad else '✓ PASS (correctly False)'}") # 4. Disabled channel test print("\n[4] Testing disabled channel...") disabled_config = {**TEST_CONFIG, "enabled": False} configured_disabled = EmailChannel.is_configured(disabled_config) print(f" is_configured when disabled: {'✗ FAIL (should be False)' if configured_disabled else '✓ PASS (correctly False)'}") print("\n=== Done ===\n") if __name__ == "__main__": parser = argparse.ArgumentParser(description="Test SendGrid email flow") parser.add_argument( "--to", required=True, help="Recipient email address for the test send", ) args = parser.parse_args() asyncio.run(run_test(args.to))