test(bstein-home): cover admin access routes

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