test: cover provisioning group flags

tests/test_provisioning.py

@@ -0,0 +1,137 @@
+from __future__ import annotations
+
+from contextlib import contextmanager
+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):
+        self._row = row
+        self.executed = []
+
+    def execute(self, query, params=None):
+        self.executed.append((query, params))
+        if "pg_try_advisory_lock" in query:
+            return DummyResult({"locked": True})
+        if "SELECT username" in query:
+            return DummyResult(self._row)
+        return DummyResult()
+
+
+class DummyDB:
+    def __init__(self, row):
+        self._row = row
+
+    @contextmanager
+    def connection(self):
+        yield DummyConn(self._row)
+
+    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 mark_welcome_sent(self, *args, **kwargs):
+        return None
+
+
+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: 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