from __future__ import annotations import types import pytest from ariadne.services import mailu as mailu_module from ariadne.services.mailu import ( MAILU_APP_PASSWORD_ATTR, MAILU_EMAIL_ATTR, MAILU_ENABLED_ATTR, MailuService, MailuSyncContext, MailuUserSyncResult, PasswordTooLongError, ) def _settings(**overrides): values = { "mailu_domain": "bstein.dev", "mailu_db_host": "localhost", "mailu_db_port": 5432, "mailu_db_name": "mailu", "mailu_db_user": "mailu", "mailu_db_password": "secret", "mailu_default_quota": 20_000_000_000, "mailu_system_users": [], "mailu_system_password": "", } values.update(overrides) return types.SimpleNamespace(**values) class _Admin: def __init__(self, *, ready: bool = True, fail_update: bool = False): self._ready = ready self.fail_update = fail_update self.calls = [] def ready(self): return self._ready def iter_users(self, page_size=200, brief=False): return [] def update_user_safe(self, user_id, payload): if self.fail_update: raise RuntimeError("update failed") self.calls.append((user_id, payload)) def set_user_attribute(self, username, key, value): self.calls.append((username, key, value)) class _Cursor: def __init__(self): self.executed = [] def execute(self, query, params=None): self.executed.append((query, params)) def fetchone(self): return None def __enter__(self): return self def __exit__(self, exc_type, exc, tb): return False class _Conn: def __init__(self): self.cursor_obj = _Cursor() def cursor(self): return self.cursor_obj def __enter__(self): return self def __exit__(self, exc_type, exc, tb): return False def _service(monkeypatch, **settings_overrides) -> MailuService: monkeypatch.setattr(mailu_module, "settings", _settings(**settings_overrides)) return MailuService() def test_mailu_helper_and_context_edges(monkeypatch) -> None: svc = _service(monkeypatch) assert mailu_module._extract_attr(None, MAILU_EMAIL_ATTR) is None assert mailu_module._extract_attr({MAILU_EMAIL_ATTR: ["", " alias@bstein.dev "]}, MAILU_EMAIL_ATTR) == "alias@bstein.dev" assert mailu_module._extract_attr({MAILU_EMAIL_ATTR: ["", " "]}, MAILU_EMAIL_ATTR) is None assert mailu_module._extract_attr({MAILU_EMAIL_ATTR: " alias@bstein.dev "}, MAILU_EMAIL_ATTR) == "alias@bstein.dev" assert mailu_module._display_name({"firstName": " Alice ", "lastName": " Example "}) == "Alice Example" assert mailu_module._domain_matches("ALICE@BSTEIN.DEV") assert not mailu_module._domain_matches("alice@example.com") assert mailu_module._password_too_long("x" * 100) assert svc.ready() updates = {} assert svc._mailu_enabled({}, updates) assert updates == {MAILU_ENABLED_ATTR: ["true"]} assert svc._mailu_enabled({MAILU_ENABLED_ATTR: ["YES"]}, {}) assert not svc._mailu_enabled({MAILU_ENABLED_ATTR: ["no"]}, {}) assert svc.resolve_mailu_email("alice", {}, "alice@example.com") == "alice@bstein.dev" monkeypatch.setattr(mailu_module, "random_password", lambda *_args: "generated") enabled, prepared_updates, password = svc._prepare_updates("alice", {}, "alice@bstein.dev") assert enabled assert prepared_updates[MAILU_EMAIL_ATTR] == ["alice@bstein.dev"] assert prepared_updates[MAILU_APP_PASSWORD_ATTR] == ["generated"] assert password == "generated" admin = _Admin() monkeypatch.setattr(mailu_module, "keycloak_admin", admin) assert svc._apply_updates("1", {}, "alice") assert svc._apply_updates("1", {MAILU_EMAIL_ATTR: ["alice@bstein.dev"]}, "alice") assert admin.calls monkeypatch.setattr(mailu_module, "keycloak_admin", _Admin(fail_update=True)) assert not svc._apply_updates("1", {MAILU_EMAIL_ATTR: ["alice@bstein.dev"]}, "alice") assert svc._should_skip_user({}, "") assert svc._should_skip_user({"enabled": False}, "alice") assert svc._should_skip_user({"serviceAccountClientId": "svc"}, "alice") assert svc._build_sync_context({})[1].skipped == 1 assert svc._build_sync_context({"username": "alice"})[1].failures == 1 monkeypatch.setattr(svc, "_apply_updates", lambda *_args: False) assert svc._build_sync_context({"id": "1", "username": "alice", "attributes": "bad"})[1].failures == 1 def test_mailu_retry_and_result_edges(monkeypatch) -> None: svc = _service(monkeypatch) ctx = MailuSyncContext("alice", "1", "alice@bstein.dev", "pw", 0, "Alice") monkeypatch.setattr(mailu_module, "random_password", lambda *_args: "retry") admin = _Admin() monkeypatch.setattr(mailu_module, "keycloak_admin", admin) calls = {"count": 0} def retrying_ensure(_conn, _email, password, _display): calls["count"] += 1 if calls["count"] == 1: raise PasswordTooLongError("too long") assert password == "retry" return True monkeypatch.setattr(svc, "_ensure_mailbox", retrying_ensure) assert svc._ensure_mailbox_with_retry(_Conn(), ctx) == (True, False, True) assert admin.calls def failing_retry(_conn, _email, _password, _display): raise PasswordTooLongError("too long") monkeypatch.setattr(svc, "_ensure_mailbox", failing_retry) monkeypatch.setattr(mailu_module, "keycloak_admin", _Admin(fail_update=True)) assert svc._ensure_mailbox_with_retry(_Conn(), ctx) == (False, True, True) monkeypatch.setattr(svc, "_ensure_mailbox", lambda *_args: (_ for _ in ()).throw(RuntimeError("boom"))) assert svc._ensure_mailbox_with_retry(_Conn(), ctx) == (False, True, False) assert svc._build_sync_result(1, False, True, False) == MailuUserSyncResult(failures=1, updated=1) assert svc._build_sync_result(0, True, False, False) == MailuUserSyncResult(processed=1, mailboxes=1) assert svc._build_sync_result(0, False, False, True) == MailuUserSyncResult(skipped=1, updated=1) assert svc._build_sync_result(0, False, False, False) == MailuUserSyncResult(skipped=1) def test_mailu_ensure_mailbox_edges(monkeypatch) -> None: svc = _service(monkeypatch) assert not svc._ensure_mailbox(_Conn(), "", "pw", "") assert not svc._ensure_mailbox(_Conn(), "invalid", "pw", "") assert not svc._ensure_mailbox(_Conn(), "alice@example.com", "pw", "") with pytest.raises(PasswordTooLongError): svc._ensure_mailbox(_Conn(), "alice@bstein.dev", "x" * 100, "") conn = _Conn() monkeypatch.setattr(mailu_module.bcrypt_sha256, "hash", lambda password: f"hashed:{password}") assert svc._ensure_mailbox(conn, "alice@bstein.dev", "pw", "Alice") query, params = conn.cursor_obj.executed[0] assert "INSERT INTO" in query assert params["localpart"] == "alice" assert params["display"] == "Alice" def raise_password_limit(_password): raise ValueError("password cannot be longer than 72 bytes") monkeypatch.setattr(mailu_module.bcrypt_sha256, "hash", raise_password_limit) with pytest.raises(PasswordTooLongError): svc._ensure_mailbox(_Conn(), "alice@bstein.dev", "pw", "") monkeypatch.setattr(mailu_module.bcrypt_sha256, "hash", lambda _password: (_ for _ in ()).throw(ValueError("other"))) with pytest.raises(ValueError, match="other"): svc._ensure_mailbox(_Conn(), "alice@bstein.dev", "pw", "") def test_mailu_system_sync_and_wait_edges(monkeypatch) -> None: svc = _service(monkeypatch) assert svc._ensure_system_mailboxes(_Conn()) == 0 svc = _service(monkeypatch, mailu_system_users=["ops@bstein.dev"], mailu_system_password="") assert svc._ensure_system_mailboxes(_Conn()) == 0 svc = _service(monkeypatch, mailu_system_users=["ops@bstein.dev"], mailu_system_password="x" * 100) assert svc._ensure_system_mailboxes(_Conn()) == 0 svc = _service(monkeypatch, mailu_system_users=["", "ops@bstein.dev"], mailu_system_password="pw") monkeypatch.setattr(svc, "_ensure_mailbox", lambda *_args: True) assert svc._ensure_system_mailboxes(_Conn()) == 1 monkeypatch.setattr(mailu_module, "keycloak_admin", _Admin(ready=False)) with pytest.raises(RuntimeError, match="keycloak"): svc.sync("schedule") monkeypatch.setattr(mailu_module, "keycloak_admin", _Admin()) svc = _service(monkeypatch, mailu_db_password="") with pytest.raises(RuntimeError, match="database"): svc.sync("schedule") svc = _service(monkeypatch) monkeypatch.setattr(svc, "mailbox_exists", lambda _email: False) ticks = iter([0, 1, 3]) monkeypatch.setattr(mailu_module.time, "time", lambda: next(ticks)) monkeypatch.setattr(mailu_module.time, "sleep", lambda _seconds: None) assert not svc.wait_for_mailbox("missing@bstein.dev", timeout_sec=2)