1569 lines
58 KiB
Python
1569 lines
58 KiB
Python
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 _patch_mailu_ready(monkeypatch, settings, value=None) -> None:
|
|
if value is None:
|
|
value = bool(getattr(settings, "mailu_sync_url", ""))
|
|
monkeypatch.setattr(prov.mailu, "ready", lambda: value)
|
|
|
|
|
|
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)
|
|
_patch_mailu_ready(monkeypatch, 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)
|
|
_patch_mailu_ready(monkeypatch, 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)
|
|
_patch_mailu_ready(monkeypatch, 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)
|
|
_patch_mailu_ready(monkeypatch, 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)
|
|
_patch_mailu_ready(monkeypatch, 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)
|
|
_patch_mailu_ready(monkeypatch, 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)
|
|
_patch_mailu_ready(monkeypatch, 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)
|
|
_patch_mailu_ready(monkeypatch, 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)
|
|
_patch_mailu_ready(monkeypatch, 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)
|
|
_patch_mailu_ready(monkeypatch, 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)
|
|
_patch_mailu_ready(monkeypatch, 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)
|
|
_patch_mailu_ready(monkeypatch, 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)
|
|
_patch_mailu_ready(monkeypatch, 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)
|
|
_patch_mailu_ready(monkeypatch, 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)
|
|
_patch_mailu_ready(monkeypatch, 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)
|
|
_patch_mailu_ready(monkeypatch, 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)
|
|
_patch_mailu_ready(monkeypatch, 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)
|
|
_patch_mailu_ready(monkeypatch, 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)
|
|
_patch_mailu_ready(monkeypatch, 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)
|
|
_patch_mailu_ready(monkeypatch, 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)
|
|
_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.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)
|
|
_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 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)
|
|
_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, *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)
|
|
_patch_mailu_ready(monkeypatch, 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)
|
|
_patch_mailu_ready(monkeypatch, 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)
|
|
_patch_mailu_ready(monkeypatch, 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)
|
|
_patch_mailu_ready(monkeypatch, 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)
|
|
_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)
|
|
|
|
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)
|
|
_patch_mailu_ready(monkeypatch, 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)
|
|
_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
|