330 lines
14 KiB
Python
330 lines
14 KiB
Python
from tests.unit.manager.provisioning_helpers import *
|
|
|
|
|
|
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"
|