From e1dde0bf43b53bc1fd6fe73e3f17b7046c9ab6d1 Mon Sep 17 00:00:00 2001 From: codex Date: Tue, 21 Apr 2026 08:07:51 -0300 Subject: [PATCH] test(bstein-home): cover admin access routes --- backend/tests/test_admin_access.py | 226 +++++++++++++++++++++++++++++ 1 file changed, 226 insertions(+) create mode 100644 backend/tests/test_admin_access.py diff --git a/backend/tests/test_admin_access.py b/backend/tests/test_admin_access.py new file mode 100644 index 0000000..1ecd1c3 --- /dev/null +++ b/backend/tests/test_admin_access.py @@ -0,0 +1,226 @@ +from __future__ import annotations + +from contextlib import contextmanager +from datetime import datetime, timezone +from typing import Any + +from flask import Flask, g, jsonify + +from atlas_portal.routes import admin_access as admin + + +class DummyResult: + def __init__(self, row: dict[str, Any] | None = None, rows: list[dict[str, Any]] | None = None) -> None: + self.row = row + self.rows = rows or [] + + def fetchone(self) -> dict[str, Any] | None: + return self.row + + def fetchall(self) -> list[dict[str, Any]]: + return self.rows + + +class DummyConn: + def __init__( + self, + *, + row: dict[str, Any] | None = None, + rows: list[dict[str, Any]] | None = None, + fail: bool = False, + ) -> None: + self.row = row + self.rows = rows or [] + self.fail = fail + self.executed: list[tuple[str, object | None]] = [] + + def execute(self, query: str, params: object | None = None) -> DummyResult: + self.executed.append((query, params)) + if self.fail: + raise RuntimeError("database failed") + return DummyResult(row=self.row, rows=self.rows) + + +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, "method": method, "path": path, "payload": payload}) + + +class DummyAdmin: + def __init__(self, *, ready: bool = True, groups: list[str] | None = None, fail: bool = False) -> None: + self._ready = ready + self.groups = groups or [] + self.fail = fail + + def ready(self) -> bool: + return self._ready + + def list_group_names(self) -> list[str]: + if self.fail: + raise RuntimeError("keycloak failed") + return self.groups + + +def make_client(monkeypatch, *, conn: DummyConn | None = None, ariadne: DummyAriadne | None = None, is_admin: bool = True): + app = Flask(__name__) + app.secret_key = "test" + active_conn = conn or DummyConn() + active_ariadne = ariadne or DummyAriadne() + + monkeypatch.setattr(admin, "require_auth", lambda fn: fn) + monkeypatch.setattr( + admin, + "require_portal_admin", + lambda: (True, None) if is_admin else (False, (jsonify({"error": "forbidden"}), 403)), + ) + monkeypatch.setattr(admin, "configured", lambda: True) + monkeypatch.setattr(admin, "ariadne_client", active_ariadne) + monkeypatch.setattr(admin, "admin_client", lambda: DummyAdmin(groups=["dev", "admin", "quality"])) + + @contextmanager + def connect(): + yield active_conn + + monkeypatch.setattr(admin, "connect", connect) + + @app.before_request + def set_user() -> None: + g.keycloak_username = "brad" + + admin.register(app) + return app.test_client(), active_conn, active_ariadne + + +def test_admin_list_requests_preflight_proxy_and_database_paths(monkeypatch) -> None: + client, _conn, _ariadne = make_client(monkeypatch, is_admin=False) + assert client.get("/api/admin/access/requests").status_code == 403 + + client, _conn, _ariadne = make_client(monkeypatch) + monkeypatch.setattr(admin, "configured", lambda: False) + assert client.get("/api/admin/access/requests").status_code == 503 + + ariadne = DummyAriadne(enabled=True) + client, _conn, proxied = make_client(monkeypatch, ariadne=ariadne) + assert client.get("/api/admin/access/requests").get_json()["proxied"] is True + assert proxied.calls == [("GET", "/api/admin/access/requests", None)] + + now = datetime(2026, 4, 20, tzinfo=timezone.utc) + conn = DummyConn( + rows=[ + { + "request_code": "alice~CODE", + "username": "alice", + "contact_email": "alice@example.dev", + "first_name": "Alice", + "last_name": "Atlas", + "created_at": now, + "note": "please", + } + ] + ) + client, _conn, _ariadne = make_client(monkeypatch, conn=conn) + data = client.get("/api/admin/access/requests").get_json() + assert data["requests"][0]["id"] == "alice~CODE" + assert data["requests"][0]["created_at"].startswith("2026-04-20T00:00:00") + + broken = DummyConn(fail=True) + client, _conn, _ariadne = make_client(monkeypatch, conn=broken) + assert client.get("/api/admin/access/requests").status_code == 502 + + +def test_admin_flags_paths(monkeypatch) -> None: + client, _conn, _ariadne = make_client(monkeypatch, is_admin=False) + assert client.get("/api/admin/access/flags").status_code == 403 + + ariadne = DummyAriadne(enabled=True) + client, _conn, proxied = make_client(monkeypatch, ariadne=ariadne) + assert client.get("/api/admin/access/flags").get_json()["proxied"] is True + assert proxied.calls == [("GET", "/api/admin/access/flags", None)] + + client, _conn, _ariadne = make_client(monkeypatch) + monkeypatch.setattr(admin, "admin_client", lambda: DummyAdmin(ready=False)) + assert client.get("/api/admin/access/flags").status_code == 503 + + monkeypatch.setattr(admin, "admin_client", lambda: DummyAdmin(groups=["dev", "admin", "quality"], fail=True)) + assert client.get("/api/admin/access/flags").status_code == 502 + + monkeypatch.setattr(admin.settings, "PORTAL_ADMIN_GROUPS", {"admin"}) + monkeypatch.setattr(admin, "admin_client", lambda: DummyAdmin(groups=["dev", "admin", "quality"])) + assert client.get("/api/admin/access/flags").get_json() == {"flags": ["dev", "quality"]} + + +def test_admin_approve_paths(monkeypatch) -> None: + client, _conn, _ariadne = make_client(monkeypatch, is_admin=False) + assert client.post("/api/admin/access/requests/alice/approve", json={}).status_code == 403 + + client, _conn, _ariadne = make_client(monkeypatch) + monkeypatch.setattr(admin, "configured", lambda: False) + assert client.post("/api/admin/access/requests/alice/approve", json={}).status_code == 503 + + ariadne = DummyAriadne(enabled=True) + client, _conn, proxied = make_client(monkeypatch, ariadne=ariadne) + assert client.post("/api/admin/access/requests/alice space/approve", json={"flags": ["dev"]}).get_json()["proxied"] + assert proxied.calls == [ + ("POST", "/api/admin/access/requests/alice%20space/approve", {"flags": ["dev"]}) + ] + + conn = DummyConn(row={"request_code": "alice~CODE"}) + provisioned: list[str] = [] + monkeypatch.setattr(admin, "provision_access_request", lambda code: provisioned.append(code)) + client, active_conn, _ariadne = make_client(monkeypatch, conn=conn) + response = client.post("/api/admin/access/requests/alice/approve", json={"flags": ["dev", 7], "note": " ok "}) + assert response.get_json() == {"ok": True, "request_code": "alice~CODE"} + assert provisioned == ["alice~CODE"] + assert active_conn.executed[0][1] == ("brad", ["dev"], "ok", "alice") + + monkeypatch.setattr(admin, "provision_access_request", lambda code: (_ for _ in ()).throw(RuntimeError("boom"))) + client, _conn, _ariadne = make_client(monkeypatch, conn=conn) + assert client.post("/api/admin/access/requests/alice/approve", json={}).status_code == 200 + + client, _conn, _ariadne = make_client(monkeypatch, conn=DummyConn(row=None)) + assert client.post("/api/admin/access/requests/alice/approve", json={}).get_json() == { + "ok": True, + "request_code": "", + } + + client, _conn, _ariadne = make_client(monkeypatch, conn=DummyConn(fail=True)) + assert client.post("/api/admin/access/requests/alice/approve", json={}).status_code == 502 + + +def test_admin_deny_paths(monkeypatch) -> None: + client, _conn, _ariadne = make_client(monkeypatch, is_admin=False) + assert client.post("/api/admin/access/requests/alice/deny", json={}).status_code == 403 + + client, _conn, _ariadne = make_client(monkeypatch) + monkeypatch.setattr(admin, "configured", lambda: False) + assert client.post("/api/admin/access/requests/alice/deny", json={}).status_code == 503 + + ariadne = DummyAriadne(enabled=True) + client, _conn, proxied = make_client(monkeypatch, ariadne=ariadne) + assert client.post("/api/admin/access/requests/alice space/deny", json={"note": "no"}).get_json()["proxied"] + assert proxied.calls == [("POST", "/api/admin/access/requests/alice%20space/deny", {"note": "no"})] + + conn = DummyConn(row={"request_code": "alice~DENY"}) + client, active_conn, _ariadne = make_client(monkeypatch, conn=conn) + assert client.post("/api/admin/access/requests/alice/deny", json={"note": " no "}).get_json() == { + "ok": True, + "request_code": "alice~DENY", + } + assert active_conn.executed[0][1] == ("brad", "no", "alice") + + client, _conn, _ariadne = make_client(monkeypatch, conn=DummyConn(row=None)) + assert client.post("/api/admin/access/requests/alice/deny", json={}).get_json() == { + "ok": True, + "request_code": "", + } + + client, _conn, _ariadne = make_client(monkeypatch, conn=DummyConn(fail=True)) + assert client.post("/api/admin/access/requests/alice/deny", json={}).status_code == 502