from __future__ import annotations from contextlib import contextmanager from datetime import datetime, timezone import types from ariadne.manager import provisioning as prov from ariadne.services.vaultwarden import VaultwardenInvite class DummyResult: def __init__(self, row=None): self._row = row def fetchone(self): return self._row def fetchall(self): return [] class DummyConn: def __init__(self, row, locked=True): self._row = row self._locked = locked self.executed = [] def execute(self, query, params=None): self.executed.append((query, params)) if "pg_try_advisory_lock" in query: return DummyResult({"locked": self._locked}) if "SELECT username" in query: return DummyResult(self._row) return DummyResult() class DummyDB: def __init__(self, row, locked=True): self._row = row self._locked = locked @contextmanager def connection(self): yield DummyConn(self._row, locked=self._locked) def fetchone(self, query, params=None): return None def fetchall(self, query, params=None): return [] class DummyStorage: def record_task_run(self, *args, **kwargs): return None def record_event(self, *args, **kwargs): return None def mark_welcome_sent(self, *args, **kwargs): return None def list_provision_candidates(self): return [] class DummyAdmin: def __init__(self): self.groups = [] def ready(self): return True def find_user(self, username): return {"id": "1"} def find_user_by_email(self, email): return None def get_user(self, user_id): return {"id": "1", "attributes": {}} def create_user(self, payload): return "1" def update_user(self, user_id, payload): return None def reset_password(self, user_id, password, temporary=False): return None def set_user_attribute(self, username, key, value): return None def get_group_id(self, group_name: str): return group_name def add_user_to_group(self, user_id, group_id): self.groups.append(group_id) def test_provisioning_filters_flag_groups(monkeypatch) -> None: dummy_settings = types.SimpleNamespace( mailu_domain="bstein.dev", mailu_sync_url="http://mailu", mailu_mailbox_wait_timeout_sec=1.0, nextcloud_namespace="nextcloud", nextcloud_mail_sync_cronjob="nextcloud-mail-sync", provision_retry_cooldown_sec=0.0, default_user_groups=["dev"], allowed_flag_groups=["demo", "test"], welcome_email_enabled=False, portal_public_base_url="https://bstein.dev", ) monkeypatch.setattr(prov, "settings", dummy_settings) admin = DummyAdmin() monkeypatch.setattr(prov, "keycloak_admin", admin) monkeypatch.setattr(prov.mailu, "sync", lambda reason, force=False: None) monkeypatch.setattr(prov.mailu, "wait_for_mailbox", lambda email, timeout: True) monkeypatch.setattr(prov.nextcloud, "sync_mail", lambda username, wait=True: {"status": "ok"}) monkeypatch.setattr(prov.wger, "sync_user", lambda username, email, password, wait=True: {"status": "ok"}) monkeypatch.setattr(prov.firefly, "sync_user", lambda email, password, wait=True: {"status": "ok"}) monkeypatch.setattr(prov.vaultwarden, "invite_user", lambda email: VaultwardenInvite(True, "invited")) monkeypatch.setattr(prov.ProvisioningManager, "_all_tasks_ok", lambda *args, **kwargs: True) monkeypatch.setattr(prov.ProvisioningManager, "_send_welcome_email", lambda *args, **kwargs: None) row = { "username": "alice", "contact_email": "alice@example.com", "email_verified_at": None, "status": "approved", "initial_password": None, "initial_password_revealed_at": None, "provision_attempted_at": None, "approval_flags": ["demo", "admin"], } db = DummyDB(row) storage = DummyStorage() manager = prov.ProvisioningManager(db, storage) manager.provision_access_request("REQ123") assert "dev" in admin.groups assert "demo" in admin.groups assert "admin" not in admin.groups def test_provisioning_creates_user_and_password(monkeypatch) -> None: dummy_settings = types.SimpleNamespace( mailu_domain="bstein.dev", mailu_sync_url="http://mailu", mailu_mailbox_wait_timeout_sec=1.0, nextcloud_namespace="nextcloud", nextcloud_mail_sync_cronjob="nextcloud-mail-sync", provision_retry_cooldown_sec=0.0, default_user_groups=["dev"], allowed_flag_groups=["demo"], welcome_email_enabled=False, portal_public_base_url="https://bstein.dev", ) monkeypatch.setattr(prov, "settings", dummy_settings) class Admin(DummyAdmin): def __init__(self): super().__init__() self.created_payload = None self.reset_calls = [] def find_user(self, username): return None def get_user(self, user_id): return { "id": user_id, "username": "alice", "requiredActions": ["CONFIGURE_TOTP"], "attributes": {}, } def create_user(self, payload): self.created_payload = payload return "user-123" def update_user(self, user_id, payload): return None def reset_password(self, user_id, password, temporary=False): self.reset_calls.append((user_id, temporary)) admin = Admin() monkeypatch.setattr(prov, "keycloak_admin", admin) monkeypatch.setattr(prov.mailu, "sync", lambda reason, force=False: None) monkeypatch.setattr(prov.mailu, "wait_for_mailbox", lambda email, timeout: True) monkeypatch.setattr(prov.nextcloud, "sync_mail", lambda username, wait=True: {"status": "ok"}) monkeypatch.setattr(prov.wger, "sync_user", lambda username, email, password, wait=True: {"status": "ok"}) monkeypatch.setattr(prov.firefly, "sync_user", lambda email, password, wait=True: {"status": "ok"}) monkeypatch.setattr(prov.vaultwarden, "invite_user", lambda email: VaultwardenInvite(True, "invited")) monkeypatch.setattr(prov.ProvisioningManager, "_all_tasks_ok", lambda *args, **kwargs: True) monkeypatch.setattr(prov.ProvisioningManager, "_send_welcome_email", lambda *args, **kwargs: None) row = { "username": "alice", "contact_email": "alice@example.com", "email_verified_at": datetime.now(timezone.utc), "status": "approved", "initial_password": None, "initial_password_revealed_at": None, "provision_attempted_at": None, "approval_flags": ["demo"], } db = DummyDB(row) storage = DummyStorage() manager = prov.ProvisioningManager(db, storage) outcome = manager.provision_access_request("REQ124") assert outcome.status == "awaiting_onboarding" assert admin.created_payload is not None assert admin.reset_calls def test_extract_attr_variants() -> None: assert prov._extract_attr("bad", "key") == "" assert prov._extract_attr({"key": ["value"]}, "key") == "value" assert prov._extract_attr({"key": ["", " "]}, "key") == "" assert prov._extract_attr({"key": "value"}, "key") == "value" assert prov._extract_attr({}, "key") == "" def test_provisioning_empty_request_code(monkeypatch) -> None: db = DummyDB({}) storage = DummyStorage() manager = prov.ProvisioningManager(db, storage) outcome = manager.provision_access_request("") assert outcome.status == "unknown" def test_provisioning_admin_not_ready(monkeypatch) -> None: monkeypatch.setattr(prov.keycloak_admin, "ready", lambda: False) db = DummyDB({}) storage = DummyStorage() manager = prov.ProvisioningManager(db, storage) outcome = manager.provision_access_request("REQ") assert outcome.status == "accounts_building" def test_provisioning_lock_not_acquired(monkeypatch) -> None: monkeypatch.setattr(prov.keycloak_admin, "ready", lambda: True) row = { "username": "alice", "contact_email": "alice@example.com", "email_verified_at": datetime.now(timezone.utc), "status": "approved", "initial_password": None, "initial_password_revealed_at": None, "provision_attempted_at": None, "approval_flags": [], } db = DummyDB(row, locked=False) storage = DummyStorage() manager = prov.ProvisioningManager(db, storage) outcome = manager.provision_access_request("REQ") assert outcome.status == "accounts_building" def test_provisioning_cooldown_short_circuit(monkeypatch) -> None: dummy_settings = types.SimpleNamespace( mailu_domain="bstein.dev", mailu_sync_url="http://mailu", mailu_mailbox_wait_timeout_sec=1.0, nextcloud_namespace="nextcloud", nextcloud_mail_sync_cronjob="nextcloud-mail-sync", provision_retry_cooldown_sec=9999.0, default_user_groups=["dev"], allowed_flag_groups=["demo"], welcome_email_enabled=False, portal_public_base_url="https://bstein.dev", provision_poll_interval_sec=1.0, ) monkeypatch.setattr(prov, "settings", dummy_settings) monkeypatch.setattr(prov.keycloak_admin, "ready", lambda: True) row = { "username": "alice", "contact_email": "alice@example.com", "email_verified_at": datetime.now(timezone.utc), "status": "accounts_building", "initial_password": None, "initial_password_revealed_at": None, "provision_attempted_at": datetime.now(timezone.utc), "approval_flags": [], } db = DummyDB(row, locked=True) storage = DummyStorage() manager = prov.ProvisioningManager(db, storage) outcome = manager.provision_access_request("REQ") assert outcome.status == "accounts_building" def test_provisioning_mailu_sync_disabled(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=["demo"], welcome_email_enabled=False, portal_public_base_url="https://bstein.dev", ) monkeypatch.setattr(prov, "settings", dummy_settings) class Admin(DummyAdmin): def __init__(self): super().__init__() self.updated_actions = [] def find_user(self, username): return {"id": "user-1"} def get_user(self, user_id): return { "id": user_id, "username": "alice", "requiredActions": ["CONFIGURE_TOTP"], "attributes": { "mailu_email": ["alice@bstein.dev"], "mailu_enabled": ["false"], "wger_password": ["pw"], "wger_password_updated_at": ["done"], "firefly_password": ["pw"], "firefly_password_updated_at": ["done"], }, } def update_user_safe(self, user_id, payload): self.updated_actions.append(payload) admin = Admin() monkeypatch.setattr(prov, "keycloak_admin", admin) monkeypatch.setattr(prov.wger, "sync_user", lambda username, email, password, wait=True: {"status": "ok"}) monkeypatch.setattr(prov.firefly, "sync_user", lambda email, password, wait=True: {"status": "ok"}) monkeypatch.setattr(prov.vaultwarden, "invite_user", lambda email: VaultwardenInvite(True, "invited")) monkeypatch.setattr(prov.ProvisioningManager, "_all_tasks_ok", lambda *args, **kwargs: False) 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": [], } db = DummyDB(row) storage = DummyStorage() manager = prov.ProvisioningManager(db, storage) outcome = manager.provision_access_request("REQ125") assert outcome.status == "accounts_building" assert admin.updated_actions def test_provisioning_sets_missing_email(monkeypatch) -> None: dummy_settings = types.SimpleNamespace( mailu_domain="bstein.dev", mailu_sync_url="http://mailu", mailu_mailbox_wait_timeout_sec=1.0, nextcloud_namespace="nextcloud", nextcloud_mail_sync_cronjob="nextcloud-mail-sync", 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) class Admin(DummyAdmin): def __init__(self): super().__init__() self.updated_actions = [] def find_user(self, username): return {"id": "user-1"} def get_user(self, user_id): return {"id": user_id, "username": "alice", "email": None, "attributes": {}} def update_user_safe(self, user_id, payload): self.updated_actions.append(payload) admin = Admin() monkeypatch.setattr(prov, "keycloak_admin", admin) monkeypatch.setattr(prov.mailu, "sync", lambda reason, force=False: None) monkeypatch.setattr(prov.mailu, "wait_for_mailbox", lambda email, timeout: True) monkeypatch.setattr(prov.nextcloud, "sync_mail", lambda username, wait=True: {"status": "ok"}) monkeypatch.setattr(prov.wger, "sync_user", lambda username, email, password, wait=True: {"status": "ok"}) monkeypatch.setattr(prov.firefly, "sync_user", lambda email, password, wait=True: {"status": "ok"}) monkeypatch.setattr(prov.vaultwarden, "invite_user", lambda email: VaultwardenInvite(True, "invited")) monkeypatch.setattr(prov.ProvisioningManager, "_all_tasks_ok", lambda *args, **kwargs: False) 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": [], } db = DummyDB(row) storage = DummyStorage() manager = prov.ProvisioningManager(db, storage) outcome = manager.provision_access_request("REQ126") assert outcome.status == "accounts_building" assert any("email" in payload for payload in admin.updated_actions) def test_provisioning_mailbox_not_ready(monkeypatch) -> None: dummy_settings = types.SimpleNamespace( mailu_domain="bstein.dev", mailu_sync_url="http://mailu", mailu_mailbox_wait_timeout_sec=1.0, nextcloud_namespace="nextcloud", nextcloud_mail_sync_cronjob="nextcloud-mail-sync", 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) admin = DummyAdmin() monkeypatch.setattr(prov, "keycloak_admin", admin) monkeypatch.setattr(prov.mailu, "sync", lambda reason, force=False: None) monkeypatch.setattr(prov.mailu, "wait_for_mailbox", lambda email, timeout: False) monkeypatch.setattr(prov.nextcloud, "sync_mail", lambda username, wait=True: {"status": "ok"}) monkeypatch.setattr(prov.wger, "sync_user", lambda username, email, password, wait=True: {"status": "ok"}) monkeypatch.setattr(prov.firefly, "sync_user", lambda email, password, wait=True: {"status": "ok"}) monkeypatch.setattr(prov.vaultwarden, "invite_user", lambda email: VaultwardenInvite(True, "invited")) monkeypatch.setattr(prov.ProvisioningManager, "_all_tasks_ok", lambda *args, **kwargs: False) row = { "username": "alice", "contact_email": "alice@example.com", "email_verified_at": datetime.now(timezone.utc), "status": "approved", "initial_password": "temp", "initial_password_revealed_at": None, "provision_attempted_at": None, "approval_flags": [], } db = DummyDB(row) storage = DummyStorage() manager = prov.ProvisioningManager(db, storage) outcome = manager.provision_access_request("REQ126") assert outcome.status == "accounts_building" def test_provisioning_sync_errors(monkeypatch) -> None: dummy_settings = types.SimpleNamespace( mailu_domain="bstein.dev", mailu_sync_url="http://mailu", mailu_mailbox_wait_timeout_sec=1.0, nextcloud_namespace="nextcloud", nextcloud_mail_sync_cronjob="nextcloud-mail-sync", 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) admin = DummyAdmin() monkeypatch.setattr(prov, "keycloak_admin", admin) monkeypatch.setattr(prov.mailu, "sync", lambda reason, force=False: None) monkeypatch.setattr(prov.mailu, "wait_for_mailbox", lambda email, timeout: True) monkeypatch.setattr(prov.nextcloud, "sync_mail", lambda username, wait=True: {"status": "error"}) monkeypatch.setattr(prov.wger, "sync_user", lambda username, email, password, wait=True: {"status": "error"}) monkeypatch.setattr(prov.firefly, "sync_user", lambda email, password, wait=True: {"status": "error"}) monkeypatch.setattr(prov.vaultwarden, "invite_user", lambda email: VaultwardenInvite(False, "error", "fail")) monkeypatch.setattr(prov.ProvisioningManager, "_all_tasks_ok", lambda *args, **kwargs: False) row = { "username": "alice", "contact_email": "alice@example.com", "email_verified_at": datetime.now(timezone.utc), "status": "approved", "initial_password": "temp", "initial_password_revealed_at": None, "provision_attempted_at": None, "approval_flags": [], } db = DummyDB(row) storage = DummyStorage() manager = prov.ProvisioningManager(db, storage) outcome = manager.provision_access_request("REQ127") assert outcome.status == "accounts_building" def test_provisioning_run_loop_processes_candidates(monkeypatch) -> None: dummy_settings = types.SimpleNamespace(provision_poll_interval_sec=0.0) monkeypatch.setattr(prov, "settings", dummy_settings) monkeypatch.setattr(prov.keycloak_admin, "ready", lambda: True) counts = [{"status": "pending", "count": 2}, {"status": "approved", "count": 1}] seen_counts = {} class DB: def fetchall(self, *_args, **_kwargs): return counts class Storage: def list_provision_candidates(self): return [types.SimpleNamespace(request_code="REQ1"), types.SimpleNamespace(request_code="REQ2")] manager = prov.ProvisioningManager(DB(), Storage()) calls: list[str] = [] monkeypatch.setattr(manager, "provision_access_request", lambda code: calls.append(code)) monkeypatch.setattr(prov, "set_access_request_counts", lambda payload: seen_counts.update(payload)) def fake_sleep(_): manager._stop_event.set() monkeypatch.setattr(prov.time, "sleep", fake_sleep) manager._stop_event.clear() manager._run_loop() assert calls == ["REQ1", "REQ2"] assert seen_counts.get("pending") == 2 def test_provisioning_run_loop_waits_for_admin(monkeypatch) -> None: dummy_settings = types.SimpleNamespace(provision_poll_interval_sec=0.0) monkeypatch.setattr(prov, "settings", dummy_settings) monkeypatch.setattr(prov.keycloak_admin, "ready", lambda: False) class DB: def fetchall(self, *_args, **_kwargs): return [] class Storage: def list_provision_candidates(self): raise AssertionError("should not list candidates") manager = prov.ProvisioningManager(DB(), Storage()) def fake_sleep(_): manager._stop_event.set() monkeypatch.setattr(prov.time, "sleep", fake_sleep) manager._stop_event.clear() manager._run_loop() def test_provisioning_missing_verified_email(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) class Admin(DummyAdmin): def find_user(self, username): return None admin = Admin() monkeypatch.setattr(prov, "keycloak_admin", admin) monkeypatch.setattr(prov.ProvisioningManager, "_all_tasks_ok", lambda *args, **kwargs: False) row = { "username": "alice", "contact_email": "alice@example.com", "email_verified_at": None, "status": "accounts_building", "initial_password": "temp", "initial_password_revealed_at": None, "provision_attempted_at": None, "approval_flags": [], } db = DummyDB(row) storage = DummyStorage() manager = prov.ProvisioningManager(db, storage) outcome = manager.provision_access_request("REQ200") assert outcome.status == "accounts_building" def test_provisioning_initial_password_missing(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) admin = DummyAdmin() monkeypatch.setattr(prov, "keycloak_admin", admin) monkeypatch.setattr(prov.mailu, "wait_for_mailbox", lambda email, timeout: True) monkeypatch.setattr(prov.wger, "sync_user", lambda username, email, password, wait=True: {"status": "ok"}) monkeypatch.setattr(prov.firefly, "sync_user", lambda email, password, wait=True: {"status": "ok"}) monkeypatch.setattr(prov.vaultwarden, "invite_user", lambda email: VaultwardenInvite(True, "invited")) monkeypatch.setattr(prov.ProvisioningManager, "_all_tasks_ok", lambda *args, **kwargs: False) row = { "username": "alice", "contact_email": "alice@example.com", "email_verified_at": datetime.now(timezone.utc), "status": "awaiting_onboarding", "initial_password": None, "initial_password_revealed_at": None, "provision_attempted_at": None, "approval_flags": [], } db = DummyDB(row) storage = DummyStorage() manager = prov.ProvisioningManager(db, storage) outcome = manager.provision_access_request("REQ201") assert outcome.status == "awaiting_onboarding" def test_provisioning_group_and_mailu_errors(monkeypatch) -> None: dummy_settings = types.SimpleNamespace( mailu_domain="bstein.dev", mailu_sync_url="http://mailu", mailu_mailbox_wait_timeout_sec=1.0, nextcloud_namespace="nextcloud", nextcloud_mail_sync_cronjob="nextcloud-mail-sync", 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) class Admin(DummyAdmin): def get_group_id(self, group_name: str): return None def set_user_attribute(self, username: str, key: str, value: str): if key == "mailu_app_password": raise RuntimeError("fail") return None admin = Admin() monkeypatch.setattr(prov, "keycloak_admin", admin) monkeypatch.setattr(prov.mailu, "sync", lambda reason, force=False: None) monkeypatch.setattr(prov.mailu, "wait_for_mailbox", lambda email, timeout: True) monkeypatch.setattr(prov.nextcloud, "sync_mail", lambda username, wait=True: (_ for _ in ()).throw(RuntimeError("fail"))) monkeypatch.setattr(prov.wger, "sync_user", lambda username, email, password, wait=True: {"status": "ok"}) monkeypatch.setattr(prov.firefly, "sync_user", lambda email, password, wait=True: {"status": "ok"}) monkeypatch.setattr(prov.vaultwarden, "invite_user", lambda email: VaultwardenInvite(True, "invited")) monkeypatch.setattr(prov.ProvisioningManager, "_all_tasks_ok", lambda *args, **kwargs: False) 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": [], } db = DummyDB(row) storage = DummyStorage() manager = prov.ProvisioningManager(db, storage) outcome = manager.provision_access_request("REQ202") assert outcome.status == "accounts_building" def test_provisioning_task_helpers() -> None: class Conn: def execute(self, *_args, **_kwargs): class Result: def fetchall(self): return [ {"task": "a", "status": "ok"}, {"task": "b", "status": "error"}, "bad", ] return Result() manager = prov.ProvisioningManager(DummyDB({}), DummyStorage()) statuses = manager._task_statuses(Conn(), "REQ") assert statuses == {"a": "ok", "b": "error"} assert manager._all_tasks_ok(Conn(), "REQ", ["a"]) is True assert manager._all_tasks_ok(Conn(), "REQ", ["b"]) is False def test_provisioning_ensure_task_rows_empty() -> None: manager = prov.ProvisioningManager(DummyDB({}), DummyStorage()) manager._ensure_task_rows(DummyConn({}, locked=True), "REQ", []) def test_provisioning_record_task_ignores_storage_errors(monkeypatch) -> None: class Storage: def record_event(self, *args, **kwargs): raise RuntimeError("fail") def record_task_run(self, *args, **kwargs): raise RuntimeError("fail") db = DummyDB({}) manager = prov.ProvisioningManager(db, Storage()) manager._record_task("REQ", "task", "ok", None, datetime.now(timezone.utc)) def test_provisioning_send_welcome_email_variants(monkeypatch) -> None: dummy_settings = types.SimpleNamespace( welcome_email_enabled=False, portal_public_base_url="https://bstein.dev", ) monkeypatch.setattr(prov, "settings", dummy_settings) manager = prov.ProvisioningManager(DummyDB({}), DummyStorage()) manager._send_welcome_email("REQ", "alice", "alice@example.com") dummy_settings = types.SimpleNamespace( welcome_email_enabled=True, portal_public_base_url="https://bstein.dev", ) monkeypatch.setattr(prov, "settings", dummy_settings) manager._send_welcome_email("REQ", "alice", "") class DB(DummyDB): def fetchone(self, *_args, **_kwargs): return None class Storage(DummyStorage): def mark_welcome_sent(self, *args, **kwargs): raise AssertionError("should not be called") manager = prov.ProvisioningManager(DB({}), Storage()) monkeypatch.setattr( prov.mailer, "send_welcome", lambda *args, **kwargs: (_ for _ in ()).throw(prov.MailerError("fail")), ) manager._send_welcome_email("REQ", "alice", "alice@example.com") def test_provisioning_start_stop(monkeypatch) -> None: class DummyThread: def __init__(self, target=None, name=None, daemon=None): self.started = False self.joined = False def is_alive(self) -> bool: return False def start(self) -> None: self.started = True def join(self, timeout=None) -> None: self.joined = True monkeypatch.setattr(prov.threading, "Thread", DummyThread) manager = prov.ProvisioningManager(DummyDB({}), DummyStorage()) manager.start() assert manager._thread.started is True manager.stop() assert manager._thread.joined is True def test_provisioning_start_skips_when_running() -> None: class LiveThread: def __init__(self): self.started = False def is_alive(self) -> bool: return True def start(self) -> None: self.started = True manager = prov.ProvisioningManager(DummyDB({}), DummyStorage()) manager._thread = LiveThread() manager.start() assert manager._thread.started is False def test_provisioning_run_loop_skips_when_admin_not_ready(monkeypatch) -> None: manager = prov.ProvisioningManager(DummyDB({}), DummyStorage()) monkeypatch.setattr(prov.keycloak_admin, "ready", lambda: False) monkeypatch.setattr(manager, "_sync_status_metrics", lambda: (_ for _ in ()).throw(RuntimeError("fail"))) monkeypatch.setattr( prov.time, "sleep", lambda *_args, **_kwargs: manager._stop_event.set(), ) manager._run_loop() def test_provisioning_run_loop_processes_candidates(monkeypatch) -> None: manager = prov.ProvisioningManager(DummyDB({}), DummyStorage()) monkeypatch.setattr(prov.keycloak_admin, "ready", lambda: True) monkeypatch.setattr(manager._storage, "list_provision_candidates", lambda: [types.SimpleNamespace(request_code="REQ")]) calls: list[str] = [] monkeypatch.setattr(manager, "provision_access_request", lambda code: calls.append(code)) monkeypatch.setattr( prov.time, "sleep", lambda *_args, **_kwargs: manager._stop_event.set(), ) manager._run_loop() assert calls == ["REQ"] def test_provisioning_sync_status_metrics(monkeypatch) -> None: db = DummyDB({}) db.fetchall = lambda *_args, **_kwargs: [{"status": "pending", "count": 2}] manager = prov.ProvisioningManager(db, DummyStorage()) captured = {} monkeypatch.setattr(prov, "set_access_request_counts", lambda payload: captured.update(payload)) manager._sync_status_metrics() assert captured["pending"] == 2 def test_provisioning_locked_returns_accounts_building(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) monkeypatch.setattr(prov, "keycloak_admin", DummyAdmin()) db = DummyDB({"username": "alice"}, locked=False) manager = prov.ProvisioningManager(db, DummyStorage()) outcome = manager.provision_access_request("REQ_LOCK") assert outcome.status == "accounts_building" def test_provisioning_missing_row_returns_unknown(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) monkeypatch.setattr(prov, "keycloak_admin", DummyAdmin()) db = DummyDB(None) manager = prov.ProvisioningManager(db, DummyStorage()) outcome = manager.provision_access_request("REQ_NONE") assert outcome.status == "unknown" def test_provisioning_denied_status_returns_denied(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) monkeypatch.setattr(prov, "keycloak_admin", DummyAdmin()) row = { "username": "alice", "contact_email": "alice@example.com", "email_verified_at": datetime.now(timezone.utc), "status": "denied", "initial_password": None, "initial_password_revealed_at": None, "provision_attempted_at": None, "approval_flags": [], } db = DummyDB(row) manager = prov.ProvisioningManager(db, DummyStorage()) outcome = manager.provision_access_request("REQ_DENIED") assert outcome.status == "denied" def test_provisioning_respects_retry_cooldown(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=60.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) monkeypatch.setattr(prov, "keycloak_admin", DummyAdmin()) 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": datetime.now(), "approval_flags": [], } db = DummyDB(row) manager = prov.ProvisioningManager(db, DummyStorage()) outcome = manager.provision_access_request("REQ_COOLDOWN") assert outcome.status == "accounts_building" def test_provisioning_updates_existing_user_attrs(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) class Admin(DummyAdmin): def __init__(self): super().__init__() self.update_payloads = [] self.attr_calls = [] def find_user(self, username): return {"id": "1"} def get_user(self, user_id): return { "id": user_id, "username": "alice", "email": "", "requiredActions": ["CONFIGURE_TOTP"], "attributes": {"mailu_enabled": ["false"]}, } def update_user_safe(self, user_id, payload): self.update_payloads.append(payload) def set_user_attribute(self, username, key, value): self.attr_calls.append((key, value)) admin = Admin() monkeypatch.setattr(prov, "keycloak_admin", admin) monkeypatch.setattr(prov.mailu, "sync", lambda reason, force=False: None) monkeypatch.setattr(prov.mailu, "wait_for_mailbox", lambda email, timeout: True) monkeypatch.setattr(prov.nextcloud, "sync_mail", lambda username, wait=True: {"status": "ok"}) monkeypatch.setattr(prov.wger, "sync_user", lambda username, email, password, wait=True: {"status": "ok"}) monkeypatch.setattr(prov.firefly, "sync_user", lambda email, password, wait=True: {"status": "ok"}) monkeypatch.setattr(prov.vaultwarden, "invite_user", lambda email: VaultwardenInvite(True, "invited")) monkeypatch.setattr(prov.ProvisioningManager, "_all_tasks_ok", lambda *args, **kwargs: False) 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": [], } db = DummyDB(row) storage = DummyStorage() manager = prov.ProvisioningManager(db, storage) outcome = manager.provision_access_request("REQ_ATTRS") assert outcome.status == "accounts_building" assert admin.update_payloads assert any(key == "mailu_email" for key, _value in admin.attr_calls) def test_provisioning_mailu_sync_failure(monkeypatch) -> None: dummy_settings = types.SimpleNamespace( mailu_domain="bstein.dev", mailu_sync_url="http://mailu", 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) monkeypatch.setattr(prov, "keycloak_admin", DummyAdmin()) monkeypatch.setattr(prov.mailu, "sync", lambda *args, **kwargs: (_ for _ in ()).throw(RuntimeError("fail"))) monkeypatch.setattr(prov.mailu, "wait_for_mailbox", lambda email, timeout: False) monkeypatch.setattr(prov.ProvisioningManager, "_all_tasks_ok", lambda *args, **kwargs: False) 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": [], } db = DummyDB(row) manager = prov.ProvisioningManager(db, DummyStorage()) outcome = manager.provision_access_request("REQ_MAILU") assert outcome.status == "accounts_building" def test_provisioning_nextcloud_sync_error(monkeypatch) -> None: dummy_settings = types.SimpleNamespace( mailu_domain="bstein.dev", mailu_sync_url="", mailu_mailbox_wait_timeout_sec=1.0, nextcloud_namespace="nextcloud", nextcloud_mail_sync_cronjob="nextcloud-mail-sync", 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) monkeypatch.setattr(prov, "keycloak_admin", DummyAdmin()) monkeypatch.setattr(prov.mailu, "wait_for_mailbox", lambda email, timeout: True) monkeypatch.setattr(prov.nextcloud, "sync_mail", lambda *args, **kwargs: {"status": "error"}) monkeypatch.setattr(prov.wger, "sync_user", lambda username, email, password, wait=True: {"status": "ok"}) monkeypatch.setattr(prov.firefly, "sync_user", lambda email, password, wait=True: {"status": "ok"}) monkeypatch.setattr(prov.vaultwarden, "invite_user", lambda email: VaultwardenInvite(True, "invited")) monkeypatch.setattr(prov.ProvisioningManager, "_all_tasks_ok", lambda *args, **kwargs: False) 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": [], } db = DummyDB(row) manager = prov.ProvisioningManager(db, DummyStorage()) outcome = manager.provision_access_request("REQ_NC") assert outcome.status == "accounts_building" def test_provisioning_wger_firefly_errors(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) monkeypatch.setattr(prov, "keycloak_admin", DummyAdmin()) monkeypatch.setattr(prov.mailu, "wait_for_mailbox", lambda email, timeout: True) monkeypatch.setattr(prov.wger, "sync_user", lambda username, email, password, wait=True: {"status": "error"}) monkeypatch.setattr(prov.firefly, "sync_user", lambda email, password, wait=True: {"status": "error"}) monkeypatch.setattr(prov.vaultwarden, "invite_user", lambda email: VaultwardenInvite(True, "invited")) monkeypatch.setattr(prov.ProvisioningManager, "_all_tasks_ok", lambda *args, **kwargs: False) 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": [], } db = DummyDB(row) manager = prov.ProvisioningManager(db, DummyStorage()) outcome = manager.provision_access_request("REQ_WGER") assert outcome.status == "accounts_building" def test_provisioning_start_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) 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, *args, **kwargs): raise RuntimeError("fail") 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": [], } manager = prov.ProvisioningManager(DummyDB(row), Storage()) outcome = manager.provision_access_request("REQ_EVENT") assert outcome.status == "accounts_building" def test_provisioning_missing_verified_email(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) class Admin(DummyAdmin): def find_user(self, username): return None monkeypatch.setattr(prov, "keycloak_admin", Admin()) row = { "username": "alice", "contact_email": "alice@example.com", "email_verified_at": None, "status": "accounts_building", "initial_password": "temp", "initial_password_revealed_at": None, "provision_attempted_at": None, "approval_flags": [], } outcome = prov.ProvisioningManager(DummyDB(row), DummyStorage()).provision_access_request("REQ_MISSING") assert outcome.status == "accounts_building" def test_provisioning_email_conflict(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) class Admin(DummyAdmin): def find_user(self, username): return None def find_user_by_email(self, email): return {"username": "other"} monkeypatch.setattr(prov, "keycloak_admin", Admin()) 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), DummyStorage()).provision_access_request("REQ_CONFLICT") assert outcome.status == "accounts_building" def test_provisioning_missing_contact_email(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) class Admin(DummyAdmin): def find_user(self, username): return None monkeypatch.setattr(prov, "keycloak_admin", Admin()) row = { "username": "alice", "contact_email": "", "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), DummyStorage()).provision_access_request("REQ_EMPTY") assert outcome.status == "accounts_building" def test_provisioning_user_id_missing(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) class Admin(DummyAdmin): def find_user(self, username): return {"id": ""} monkeypatch.setattr(prov, "keycloak_admin", Admin()) 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), DummyStorage()).provision_access_request("REQ_ID") assert outcome.status == "accounts_building" def test_provisioning_initial_password_revealed(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) 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) row = { "username": "alice", "contact_email": "alice@example.com", "email_verified_at": datetime.now(timezone.utc), "status": "accounts_building", "initial_password": None, "initial_password_revealed_at": datetime.now(timezone.utc), "provision_attempted_at": None, "approval_flags": [], } outcome = prov.ProvisioningManager(DummyDB(row), DummyStorage()).provision_access_request("REQ_REVEALED") assert outcome.status == "accounts_building" def test_provisioning_vaultwarden_attribute_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) class Admin(DummyAdmin): def set_user_attribute(self, username, key, value): raise RuntimeError("fail") monkeypatch.setattr(prov, "keycloak_admin", Admin()) 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) 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), DummyStorage()).provision_access_request("REQ_VAULT") assert outcome.status == "accounts_building" 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) 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) 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) 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) 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