ariadne/tests/unit/manager/test_provisioning_welcome_email.py

from tests.unit.manager.provisioning_helpers import *


def test_provisioning_complete_event_failure(monkeypatch) -> None:
    dummy_settings = types.SimpleNamespace(
        mailu_domain="bstein.dev",
        mailu_sync_url="",
        mailu_mailbox_wait_timeout_sec=1.0,
        nextcloud_namespace="",
        nextcloud_mail_sync_cronjob="",
        provision_retry_cooldown_sec=0.0,
        default_user_groups=["dev"],
        allowed_flag_groups=[],
        welcome_email_enabled=False,
        portal_public_base_url="https://bstein.dev",
    )
    monkeypatch.setattr(prov, "settings", dummy_settings)
    _patch_mailu_ready(monkeypatch, dummy_settings)
    monkeypatch.setattr(prov, "keycloak_admin", DummyAdmin())
    monkeypatch.setattr(prov.mailu, "wait_for_mailbox", lambda email, timeout: True)
    monkeypatch.setattr(prov.wger, "sync_user", lambda *args, **kwargs: {"status": "ok"})
    monkeypatch.setattr(prov.firefly, "sync_user", lambda *args, **kwargs: {"status": "ok"})
    monkeypatch.setattr(prov.vaultwarden, "invite_user", lambda email: VaultwardenInvite(True, "invited"))
    monkeypatch.setattr(prov.ProvisioningManager, "_all_tasks_ok", lambda *args, **kwargs: True)

    class Storage(DummyStorage):
        def record_event(self, event_type, detail):
            if event_type == "provision_complete":
                raise RuntimeError("fail")
            return None

    row = {
        "username": "alice",
        "contact_email": "alice@example.com",
        "email_verified_at": datetime.now(timezone.utc),
        "status": "accounts_building",
        "initial_password": "temp",
        "initial_password_revealed_at": None,
        "provision_attempted_at": None,
        "approval_flags": [],
    }

    outcome = prov.ProvisioningManager(DummyDB(row), Storage()).provision_access_request("REQ_DONE")
    assert outcome.status == "awaiting_onboarding"

def test_provisioning_pending_event_failure(monkeypatch) -> None:
    dummy_settings = types.SimpleNamespace(
        mailu_domain="bstein.dev",
        mailu_sync_url="",
        mailu_mailbox_wait_timeout_sec=1.0,
        nextcloud_namespace="",
        nextcloud_mail_sync_cronjob="",
        provision_retry_cooldown_sec=0.0,
        default_user_groups=["dev"],
        allowed_flag_groups=[],
        welcome_email_enabled=False,
        portal_public_base_url="https://bstein.dev",
    )
    monkeypatch.setattr(prov, "settings", dummy_settings)
    _patch_mailu_ready(monkeypatch, dummy_settings)
    monkeypatch.setattr(prov, "keycloak_admin", DummyAdmin())
    monkeypatch.setattr(prov.mailu, "wait_for_mailbox", lambda email, timeout: True)
    monkeypatch.setattr(prov.wger, "sync_user", lambda *args, **kwargs: {"status": "ok"})
    monkeypatch.setattr(prov.firefly, "sync_user", lambda *args, **kwargs: {"status": "ok"})
    monkeypatch.setattr(prov.vaultwarden, "invite_user", lambda email: VaultwardenInvite(True, "invited"))
    monkeypatch.setattr(prov.ProvisioningManager, "_all_tasks_ok", lambda *args, **kwargs: False)

    class Storage(DummyStorage):
        def record_event(self, event_type, detail):
            if event_type == "provision_pending":
                raise RuntimeError("fail")
            return None

    row = {
        "username": "alice",
        "contact_email": "alice@example.com",
        "email_verified_at": datetime.now(timezone.utc),
        "status": "accounts_building",
        "initial_password": "temp",
        "initial_password_revealed_at": None,
        "provision_attempted_at": None,
        "approval_flags": [],
    }

    outcome = prov.ProvisioningManager(DummyDB(row), Storage()).provision_access_request("REQ_PENDING")
    assert outcome.status == "accounts_building"

def test_send_welcome_email_already_sent(monkeypatch) -> None:
    dummy_settings = types.SimpleNamespace(
        welcome_email_enabled=True,
        portal_public_base_url="https://bstein.dev",
    )
    monkeypatch.setattr(prov, "settings", dummy_settings)
    _patch_mailu_ready(monkeypatch, dummy_settings)

    class DB(DummyDB):
        def fetchone(self, query, params=None):
            return {"welcome_email_sent_at": datetime.now(timezone.utc)}

    sent = {"called": False}

    def mark_sent(_code):
        sent["called"] = True

    manager = prov.ProvisioningManager(DB({}), DummyStorage())
    monkeypatch.setattr(manager._storage, "mark_welcome_sent", mark_sent)
    monkeypatch.setattr(prov.mailer, "send_welcome", lambda *args, **kwargs: None)
    manager._send_welcome_email("REQ_WELCOME", "alice", "alice@example.com")
    assert sent["called"] is False

def test_send_welcome_email_marks_sent(monkeypatch) -> None:
    dummy_settings = types.SimpleNamespace(
        welcome_email_enabled=True,
        portal_public_base_url="https://bstein.dev",
    )
    monkeypatch.setattr(prov, "settings", dummy_settings)
    _patch_mailu_ready(monkeypatch, dummy_settings)

    class DB(DummyDB):
        def fetchone(self, query, params=None):
            return None

    sent = {"called": False}

    def mark_sent(_code):
        sent["called"] = True

    manager = prov.ProvisioningManager(DB({}), DummyStorage())
    monkeypatch.setattr(manager._storage, "mark_welcome_sent", mark_sent)
    monkeypatch.setattr(prov.mailer, "send_welcome", lambda *args, **kwargs: None)
    manager._send_welcome_email("REQ_WELCOME", "alice", "alice@example.com")
    assert sent["called"] is True