feat: 운영 안정성과 세션 음성 경험 개선

apps/api/app/test_auth_providers.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 import unittest
 import base64
 from contextlib import contextmanager
+from datetime import datetime, timezone
 from typing import Any
 from urllib.parse import parse_qs, urlencode, urlsplit
 
@@ -390,6 +391,36 @@ class AuthProviderScaffoldTest(unittest.IsolatedAsyncioTestCase):
             hoonjung_role = auth_routes._role_for_email("hoonjungkoo@hs.ac.kr")
             self.assertEqual(hoonjung_role, Role.ADMIN)
 
+    async def test_existing_primary_role_super_admin_persists_admin_access_on_login(
+        self,
+    ) -> None:
+        email = "persistent-root@twentyoz.kr"
+        existing = _managed_user(email=email, role="learner", admin_access=False)
+        auth_sessions._users[existing.user_id] = existing
+        auth_sessions._email_index[email] = existing.user_id
+
+        with (
+            patched_settings(
+                environment="dev",
+                auth_super_admin_emails=[email],
+                auth_admin_emails=[],
+                auth_teacher_emails=[],
+                auth_approved_emails=[],
+            ),
+            patch.object(auth_sessions, "get_pool", side_effect=RuntimeError("no db")),
+        ):
+            _, user = await auth_sessions.create_session(
+                email=email,
+                display_name="Persistent Root",
+                role="learner",
+                external_id="google:persistent-root",
+            )
+
+        self.assertEqual(user.role, "learner")
+        self.assertTrue(user.super_admin)
+        self.assertTrue(user.admin_access)
+        self.assertTrue(auth_sessions._users[existing.user_id].admin_access)
+
     async def test_pending_provider_user_enqueues_account_approval_notification(self) -> None:
         with (
             patched_settings(
@@ -544,6 +575,69 @@ class AuthProviderScaffoldTest(unittest.IsolatedAsyncioTestCase):
             self.assertTrue(updated.admin_access)
             self.assertEqual(updated.role, "learner")
 
+    async def test_super_admin_oauth_upsert_casts_nullable_admin_access_as_boolean(
+        self,
+    ) -> None:
+        class RecordingConn:
+            def __init__(self) -> None:
+                self.queries: list[str] = []
+
+            async def fetchrow(self, query: str, *args: Any) -> dict[str, Any] | None:
+                self.queries.append(query)
+                if len(self.queries) == 1:
+                    return None
+                now = datetime.now(timezone.utc)
+                return {
+                    "user_id": "00000000-0000-0000-0000-000000000605",
+                    "email": "yunchan@twentyoz.kr",
+                    "display_name": "Yun Chan",
+                    "role": "learner",
+                    "admin_access": True,
+                    "account_status": "approved",
+                    "cohort": "",
+                    "affiliation": "",
+                    "created_at": now,
+                    "last_seen_at": now,
+                }
+
+        class RecordingAcquire:
+            def __init__(self, conn: RecordingConn) -> None:
+                self.conn = conn
+
+            async def __aenter__(self) -> RecordingConn:
+                return self.conn
+
+            async def __aexit__(self, exc_type, exc, tb) -> None:
+                return None
+
+        class RecordingPool:
+            def __init__(self, conn: RecordingConn) -> None:
+                self.conn = conn
+
+            def acquire(self) -> RecordingAcquire:
+                return RecordingAcquire(self.conn)
+
+        conn = RecordingConn()
+        with (
+            patched_settings(
+                auth_super_admin_emails=["yunchan@twentyoz.kr"],
+                auth_admin_emails=[],
+            ),
+            patch.object(auth_sessions, "get_pool", return_value=RecordingPool(conn)),
+        ):
+            user = await auth_sessions.upsert_managed_user(
+                auth_sessions.ManagedUserUpsertInput(
+                    email="yunchan@twentyoz.kr",
+                    display_name="Yun Chan",
+                    role="learner",
+                    external_id="google:111856072590637505974",
+                )
+            )
+
+        self.assertTrue(user.admin_access)
+        self.assertIn("$7::boolean", conn.queries[0])
+        self.assertIn("$9::boolean", conn.queries[1])
+
     async def test_auth_config_allows_dev_login_from_configured_tailnet_forwarded_host(self) -> None:
         request = _request(
             [