"""Tests for email credential delivery.""" import pytest from dagster import build_op_context from simpl_open_credential_delivery.ops.send_credentials_email import ( send_credentials_email, ) from simpl_open_credential_delivery.services.email_service import ( EmailDeliveryService, ) class FakeSMTP: def __init__(self, host, port, timeout=None): self.host = host self.port = port self.timeout = timeout self.started_tls = False self.logged_in = False self.sent_messages = [] def __enter__(self): return self def __exit__(self, exc_type, exc, tb): return False def starttls(self): self.started_tls = True def login(self, username, password): self.logged_in = True def send_message(self, message): self.sent_messages.append(message) def test_send_credentials_email_success(monkeypatch, op_context): fake_smtp = FakeSMTP("smtp.example.local", 587) monkeypatch.setenv("SMTP_HOST", "smtp.example.local") monkeypatch.setenv("SMTP_PORT", "587") monkeypatch.setenv("SMTP_FROM_ADDRESS", "no-reply@example.local") monkeypatch.setenv("SMTP_USE_TLS", "true") monkeypatch.setenv("EMAIL_SUBJECT_TEMPLATE", "Subject {repository_identifier}") monkeypatch.setenv("EMAIL_BODY_TEMPLATE", "Hello {consumer_email} {access_token}") monkeypatch.setattr("smtplib.SMTP", lambda host, port, timeout=None: fake_smtp) result = send_credentials_email( op_context, { "repository_identifier": "simpl-open/platform-automation", "consumer_email": "consumer@example.local", "access_level": "read-only", "username": "application-consumer", "access_token": "token-123", "repository_access_url": "https://gitea.example.local/simpl-open/platform-automation", }, ) assert result["email_delivery_status"] == "SENT" assert result["email_message_id"] == "" assert fake_smtp.started_tls is True assert fake_smtp.sent_messages def test_email_delivery_service_requires_sender(monkeypatch): monkeypatch.setenv("SMTP_HOST", "smtp.example.local") monkeypatch.setenv("SMTP_PORT", "587") monkeypatch.delenv("SMTP_FROM_ADDRESS", raising=False) with pytest.raises(Exception, match="SMTP_FROM_ADDRESS"): EmailDeliveryService.from_environment()