138 lines
3.9 KiB
Python
138 lines
3.9 KiB
Python
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, 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
|