From 6c7dabe7ab31ed3e0c4ee493071dc77287e23722 Mon Sep 17 00:00:00 2001 From: codex Date: Tue, 21 Apr 2026 08:21:59 -0300 Subject: [PATCH] test(bstein-home): cover account action routes --- backend/tests/test_account_actions.py | 264 ++++++++++++++++++++++++++ 1 file changed, 264 insertions(+) create mode 100644 backend/tests/test_account_actions.py diff --git a/backend/tests/test_account_actions.py b/backend/tests/test_account_actions.py new file mode 100644 index 0000000..4746c5a --- /dev/null +++ b/backend/tests/test_account_actions.py @@ -0,0 +1,264 @@ +from __future__ import annotations + +from types import SimpleNamespace +from typing import Any + +from flask import Flask, g, jsonify + +from atlas_portal.routes import account_actions as actions + + +class DummyAriadne: + def __init__(self, enabled: bool = False) -> None: + self._enabled = enabled + self.calls: list[tuple[str, str, object | None]] = [] + + def enabled(self) -> bool: + return self._enabled + + def proxy(self, method: str, path: str, payload: object | None = None): + self.calls.append((method, path, payload)) + return jsonify({"proxied": True, "path": path, "payload": payload}) + + +class DummyAdmin: + def __init__( + self, + *, + ready: bool = True, + user: dict[str, Any] | None = None, + fail_find: bool = False, + fail_set: bool = False, + ) -> None: + self._ready = ready + self.user = user if user is not None else {"attributes": {}} + self.fail_find = fail_find + self.fail_set = fail_set + self.attributes: list[tuple[str, str, str]] = [] + + def ready(self) -> bool: + return self._ready + + def find_user(self, username: str) -> dict[str, Any] | None: + if self.fail_find: + raise RuntimeError("lookup failed") + return self.user + + def set_user_attribute(self, username: str, key: str, value: str) -> None: + if self.fail_set: + raise RuntimeError("write failed") + self.attributes.append((username, key, value)) + + +class MailuClient: + status_code = 200 + raises = False + + def __init__(self, *, timeout: int) -> None: + self.timeout = timeout + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + def post(self, url: str, json: dict[str, Any] | None = None): + if self.raises: + raise RuntimeError("mailu unavailable") + return SimpleNamespace(status_code=self.status_code) + + +def make_client( + monkeypatch, + *, + admin: DummyAdmin | None = None, + ariadne: DummyAriadne | None = None, + account_ok: bool = True, + username: str = "alice", + email: str = "alice@example.dev", +): + app = Flask(__name__) + active_admin = admin or DummyAdmin() + active_ariadne = ariadne or DummyAriadne() + + monkeypatch.setattr(actions, "require_auth", lambda fn: fn) + monkeypatch.setattr( + actions, + "require_account_access", + lambda: (True, None) if account_ok else (False, (jsonify({"error": "forbidden"}), 403)), + ) + monkeypatch.setattr(actions, "admin_client", lambda: active_admin) + monkeypatch.setattr(actions, "ariadne_client", active_ariadne) + monkeypatch.setattr(actions, "random_password", lambda length=16: f"pw-{length}") + monkeypatch.setattr(actions.settings, "MAILU_DOMAIN", "bstein.dev") + monkeypatch.setattr(actions.settings, "MAILU_SYNC_URL", "") + monkeypatch.setattr(actions, "trigger_nextcloud_mail_sync", lambda user, wait=True: {"status": "ok", "wait": wait}) + monkeypatch.setattr(actions, "trigger_wger_user_sync", lambda *args, **kwargs: {"status": "ok"}) + monkeypatch.setattr(actions, "trigger_firefly_user_sync", lambda *args, **kwargs: {"status": "ok"}) + + @app.before_request + def set_user() -> None: + g.keycloak_username = username + g.keycloak_email = email + + actions.register_account_actions(app) + return app.test_client(), active_admin, active_ariadne + + +def test_tcp_check_success_and_failure(monkeypatch) -> None: + class SocketContext: + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + monkeypatch.setattr(actions.socket, "create_connection", lambda *args, **kwargs: SocketContext()) + assert actions._tcp_check("host", 443, 1) is True + assert actions._tcp_check("", 443, 1) is False + assert actions._tcp_check("host", 0, 1) is False + monkeypatch.setattr(actions.socket, "create_connection", lambda *args, **kwargs: (_ for _ in ()).throw(OSError())) + assert actions._tcp_check("host", 443, 1) is False + + +def test_mailu_rotate_paths(monkeypatch) -> None: + client, _admin, _ariadne = make_client(monkeypatch, account_ok=False) + assert client.post("/api/account/mailu/rotate").status_code == 403 + + ariadne = DummyAriadne(enabled=True) + client, _admin, proxied = make_client(monkeypatch, ariadne=ariadne) + assert client.post("/api/account/mailu/rotate").get_json()["proxied"] is True + assert proxied.calls == [("POST", "/api/account/mailu/rotate", None)] + + client, _admin, _ariadne = make_client(monkeypatch, admin=DummyAdmin(ready=False)) + assert client.post("/api/account/mailu/rotate").status_code == 503 + + client, _admin, _ariadne = make_client(monkeypatch, username="") + assert client.post("/api/account/mailu/rotate").status_code == 400 + + client, _admin, _ariadne = make_client(monkeypatch, admin=DummyAdmin(fail_set=True)) + assert client.post("/api/account/mailu/rotate").status_code == 502 + + client, admin, _ariadne = make_client(monkeypatch) + data = client.post("/api/account/mailu/rotate").get_json() + assert data["password"] == "pw-16" + assert data["sync_enabled"] is False + assert data["nextcloud_sync"] == {"status": "ok", "wait": True} + assert admin.attributes[0] == ("alice", "mailu_app_password", "pw-16") + + for status_code, expected_error in ((200, ""), (503, "sync status 503")): + client, _admin, _ariadne = make_client(monkeypatch) + monkeypatch.setattr(actions.settings, "MAILU_SYNC_URL", "https://mailu-sync.example.dev") + MailuClient.status_code = status_code + MailuClient.raises = False + monkeypatch.setattr(actions, "httpx", SimpleNamespace(Client=MailuClient)) + data = client.post("/api/account/mailu/rotate").get_json() + assert data["sync_enabled"] is True + assert data["sync_error"] == expected_error + + client, _admin, _ariadne = make_client(monkeypatch) + monkeypatch.setattr(actions.settings, "MAILU_SYNC_URL", "https://mailu-sync.example.dev") + MailuClient.raises = True + monkeypatch.setattr(actions, "httpx", SimpleNamespace(Client=MailuClient)) + assert client.post("/api/account/mailu/rotate").get_json()["sync_error"] == "sync request failed" + MailuClient.raises = False + + client, _admin, _ariadne = make_client(monkeypatch) + monkeypatch.setattr(actions, "trigger_nextcloud_mail_sync", lambda *args, **kwargs: (_ for _ in ()).throw(RuntimeError())) + assert client.post("/api/account/mailu/rotate").get_json()["nextcloud_sync"] == {"status": "error"} + + +def test_wger_reset_and_rotation_check_paths(monkeypatch) -> None: + client, _admin, _ariadne = make_client(monkeypatch, account_ok=False) + assert client.post("/api/account/wger/reset").status_code == 403 + + ariadne = DummyAriadne(enabled=True) + client, _admin, proxied = make_client(monkeypatch, ariadne=ariadne) + assert client.post("/api/account/wger/reset").get_json()["path"] == "/api/account/wger/reset" + assert client.post("/api/account/wger/rotation/check").get_json()["path"] == "/api/account/wger/rotation/check" + assert len(proxied.calls) == 2 + + client, _admin, _ariadne = make_client(monkeypatch, admin=DummyAdmin(ready=False)) + assert client.post("/api/account/wger/reset").status_code == 503 + client, _admin, _ariadne = make_client(monkeypatch, username="") + assert client.post("/api/account/wger/reset").status_code == 400 + + client, _admin, _ariadne = make_client(monkeypatch, admin=DummyAdmin(user={"attributes": {"mailu_email": ["mail@example.dev"]}})) + assert client.post("/api/account/wger/reset").get_json()["password"] == "pw-16" + + client, _admin, _ariadne = make_client(monkeypatch, admin=DummyAdmin(user={"attributes": {"mailu_email": "mail@example.dev"}})) + assert client.post("/api/account/wger/reset").status_code == 200 + + client, _admin, _ariadne = make_client(monkeypatch, admin=DummyAdmin(fail_find=True)) + assert client.post("/api/account/wger/reset").status_code == 200 + + client, _admin, _ariadne = make_client(monkeypatch) + monkeypatch.setattr(actions, "trigger_wger_user_sync", lambda *args, **kwargs: {"status": "failed"}) + assert client.post("/api/account/wger/reset").status_code == 502 + + client, _admin, _ariadne = make_client(monkeypatch, admin=DummyAdmin(fail_set=True)) + assert client.post("/api/account/wger/reset").status_code == 502 + + client, _admin, _ariadne = make_client(monkeypatch) + assert client.post("/api/account/wger/rotation/check").status_code == 503 + + +def test_firefly_reset_and_rotation_check_paths(monkeypatch) -> None: + client, _admin, _ariadne = make_client(monkeypatch, account_ok=False) + assert client.post("/api/account/firefly/reset").status_code == 403 + + ariadne = DummyAriadne(enabled=True) + client, _admin, proxied = make_client(monkeypatch, ariadne=ariadne) + assert client.post("/api/account/firefly/reset").get_json()["path"] == "/api/account/firefly/reset" + assert client.post("/api/account/firefly/rotation/check").get_json()["path"] == "/api/account/firefly/rotation/check" + assert len(proxied.calls) == 2 + + client, _admin, _ariadne = make_client(monkeypatch, admin=DummyAdmin(ready=False)) + assert client.post("/api/account/firefly/reset").status_code == 503 + client, _admin, _ariadne = make_client(monkeypatch, username="") + assert client.post("/api/account/firefly/reset").status_code == 400 + + client, _admin, _ariadne = make_client(monkeypatch, admin=DummyAdmin(user={"attributes": {"mailu_email": ["mail@example.dev"]}})) + assert client.post("/api/account/firefly/reset").get_json()["password"] == "pw-24" + + client, _admin, _ariadne = make_client(monkeypatch, admin=DummyAdmin(user={"attributes": {"mailu_email": "mail@example.dev"}})) + assert client.post("/api/account/firefly/reset").status_code == 200 + + client, _admin, _ariadne = make_client(monkeypatch, admin=DummyAdmin(fail_find=True)) + assert client.post("/api/account/firefly/reset").status_code == 200 + + client, _admin, _ariadne = make_client(monkeypatch) + monkeypatch.setattr(actions, "trigger_firefly_user_sync", lambda *args, **kwargs: {"status": "failed"}) + assert client.post("/api/account/firefly/reset").status_code == 502 + + client, _admin, _ariadne = make_client(monkeypatch, admin=DummyAdmin(fail_set=True)) + assert client.post("/api/account/firefly/reset").status_code == 502 + + client, _admin, _ariadne = make_client(monkeypatch) + assert client.post("/api/account/firefly/rotation/check").status_code == 503 + + +def test_nextcloud_mail_sync_paths(monkeypatch) -> None: + client, _admin, _ariadne = make_client(monkeypatch, account_ok=False) + assert client.post("/api/account/nextcloud/mail/sync").status_code == 403 + + ariadne = DummyAriadne(enabled=True) + client, _admin, proxied = make_client(monkeypatch, ariadne=ariadne) + assert client.post("/api/account/nextcloud/mail/sync", json={"wait": False}).get_json()["proxied"] is True + assert proxied.calls == [("POST", "/api/account/nextcloud/mail/sync", {"wait": False})] + + client, _admin, _ariadne = make_client(monkeypatch, admin=DummyAdmin(ready=False)) + assert client.post("/api/account/nextcloud/mail/sync").status_code == 503 + client, _admin, _ariadne = make_client(monkeypatch, username="") + assert client.post("/api/account/nextcloud/mail/sync").status_code == 400 + + client, _admin, _ariadne = make_client(monkeypatch) + assert client.post("/api/account/nextcloud/mail/sync", json={"wait": False}).get_json() == { + "status": "ok", + "wait": False, + } + + client, _admin, _ariadne = make_client(monkeypatch) + monkeypatch.setattr(actions, "trigger_nextcloud_mail_sync", lambda *args, **kwargs: (_ for _ in ()).throw(RuntimeError("sync failed"))) + assert client.post("/api/account/nextcloud/mail/sync").status_code == 502