From 86e9a2d82b2d69068c916417f02ba514c406e7a9 Mon Sep 17 00:00:00 2001 From: codex Date: Tue, 21 Apr 2026 04:08:02 -0300 Subject: [PATCH] test(ariadne): cover provisioning account edges --- .../test_provisioning_accounts_edges.py | 172 ++++++++++++++++++ 1 file changed, 172 insertions(+) create mode 100644 tests/unit/manager/test_provisioning_accounts_edges.py diff --git a/tests/unit/manager/test_provisioning_accounts_edges.py b/tests/unit/manager/test_provisioning_accounts_edges.py new file mode 100644 index 0000000..b5f9e9f --- /dev/null +++ b/tests/unit/manager/test_provisioning_accounts_edges.py @@ -0,0 +1,172 @@ +from tests.unit.manager.provisioning_helpers import * + + +def _settings(**overrides): + base = { + "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", + "vaultwarden_admin_rate_limit_backoff_sec": 600.0, + } + base.update(overrides) + return types.SimpleNamespace(**base) + + +def _ctx(**overrides): + base = { + "request_code": "REQ_EDGE", + "username": "alice", + "first_name": "", + "last_name": "", + "contact_email": "alice@example.com", + "email_verified_at": datetime.now(timezone.utc), + "status": "accounts_building", + "initial_password": "temp", + "revealed_at": None, + "attempted_at": None, + "approval_flags": [], + "user_id": "user-1", + "mailu_email": "alice@bstein.dev", + } + base.update(overrides) + return prov.RequestContext(**base) + + +def _task_params(conn): + return [ + params + for query, params in conn.executed + if "INSERT INTO access_request_tasks" in query and isinstance(params, tuple) and len(params) >= 4 + ] + + +def _manager(monkeypatch, settings=None, admin=None): + monkeypatch.setattr(prov, "settings", settings or _settings()) + monkeypatch.setattr(prov, "keycloak_admin", admin or DummyAdmin()) + return prov.ProvisioningManager(DummyDB({}), DummyStorage()) + + +def test_provisioning_account_helper_edges(monkeypatch) -> None: + class Admin(DummyAdmin): + def __init__(self): + super().__init__() + self.attr_calls = [] + + def set_user_attribute(self, username, key, value): + self.attr_calls.append((username, key, value)) + + def get_user(self, user_id): + raise RuntimeError("missing") + + def update_user_safe(self, user_id, payload): + raise RuntimeError("should not update") + + admin = Admin() + manager = _manager(monkeypatch, admin=admin) + + manager._set_vaultwarden_attrs("", "alice@example.com", "present") + assert admin.attr_calls == [] + + assert manager._ready_for_retry(_ctx(status="approved")) is True + assert manager._fetch_full_user("missing", {"id": "fallback"}) == {"id": "fallback"} + manager._ensure_contact_email(_ctx(), {"email": "alice@example.com"}) + manager._ensure_mailu_attrs(_ctx(), {"attributes": "not-a-dict"}) + + +def test_create_user_fallback_finds_user_by_email(monkeypatch) -> None: + class Admin(DummyAdmin): + def find_user(self, username): + return None + + def create_user(self, payload): + raise RuntimeError("duplicate") + + def find_user_by_email(self, email): + return {"id": "by-email", "username": "alice"} + + manager = _manager(monkeypatch, admin=Admin()) + assert manager._create_or_fetch_user(_ctx())["id"] == "by-email" + + +def test_nextcloud_sync_uses_summary_detail(monkeypatch) -> None: + class Summary: + detail = "imap settings rejected" + + manager = _manager(monkeypatch, settings=_settings(nextcloud_namespace="nextcloud")) + monkeypatch.setattr( + prov.nextcloud, + "sync_mail", + lambda username, wait=True: {"status": "error", "summary": Summary()}, + ) + + conn = DummyConn({}) + manager._sync_nextcloud_mail(conn, _ctx()) + + assert any(params[2] == "error" and params[3] == "imap settings rejected" for params in _task_params(conn)) + + +def test_grandfathered_vaultwarden_rate_limit(monkeypatch) -> None: + manager = _manager(monkeypatch) + monkeypatch.setattr( + prov.vaultwarden, + "find_user_by_email", + lambda _email: VaultwardenLookup(False, "rate_limited", "slow down"), + ) + + conn = DummyConn({}) + manager._handle_vaultwarden_grandfathered(conn, _ctx(), datetime.now(timezone.utc)) + + assert any(params[2] == "pending" and "rate limited until" in (params[3] or "") for params in _task_params(conn)) + + +def test_grandfathered_vaultwarden_missing(monkeypatch) -> None: + manager = _manager(monkeypatch) + monkeypatch.setattr( + prov.vaultwarden, + "find_user_by_email", + lambda _email: VaultwardenLookup(True, "missing", "not found"), + ) + + conn = DummyConn({}) + manager._handle_vaultwarden_grandfathered(conn, _ctx(), datetime.now(timezone.utc)) + + assert any( + params[2] == "error" and params[3] == "vaultwarden account not found for recovery email" + for params in _task_params(conn) + ) + + +def test_grandfathered_vaultwarden_lookup_failure(monkeypatch) -> None: + manager = _manager(monkeypatch) + monkeypatch.setattr( + prov.vaultwarden, + "find_user_by_email", + lambda _email: VaultwardenLookup(False, "bad_gateway", "upstream down"), + ) + + conn = DummyConn({}) + manager._handle_vaultwarden_grandfathered(conn, _ctx(), datetime.now(timezone.utc)) + + assert any(params[2] == "error" and params[3] == "upstream down" for params in _task_params(conn)) + + +def test_vaultwarden_invite_skips_when_retry_not_due(monkeypatch) -> None: + manager = _manager(monkeypatch) + monkeypatch.setattr(manager, "_vaultwarden_retry_due", lambda conn, request_code: False) + monkeypatch.setattr( + prov.vaultwarden, + "invite_user", + lambda _email: (_ for _ in ()).throw(RuntimeError("should not invite")), + ) + + conn = DummyConn({}) + manager._ensure_vaultwarden_invite(conn, _ctx()) + + assert _task_params(conn) == []