from __future__ import annotations import time import unittest from contextlib import contextmanager from datetime import datetime, timezone from pathlib import Path from typing import Any from unittest.mock import AsyncMock, patch from starlette.requests import Request from . import auth_sessions from .config import settings from .routes import auth as auth_routes @contextmanager def patched_settings(**values: Any): previous = {key: getattr(settings, key) for key in values} for key, value in values.items(): setattr(settings, key, value) try: yield finally: for key, value in previous.items(): setattr(settings, key, value) def managed_user( *, user_id: str = "f646404b-a9e5-4422-a97d-548c074334d5", email: str = "yunchan@twentyoz.kr", account_status: str = "approved", ) -> auth_sessions.ManagedUser: return auth_sessions.ManagedUser( user_id=user_id, email=email, display_name="Yun Chan", role="admin", admin_access=True, account_status=account_status, cohort_ids=["owner"], affiliation="TwentyOZ", legal_name="Yun Chan", department="", grade_level="", phone="", contact_address="", nickname="윤찬", self_introduction="관리자", avatar_url="", consent_at=1.0, profile_completed_at=1.0, terms_agreed_at=1.0, privacy_agreed_at=1.0, terms_version="v1", privacy_version="v1", created_at=1.0, last_seen_at=1.0, ) def alias_row(*, active: bool = True, status: str = "approved") -> dict[str, Any]: user = managed_user(account_status=status) return { "user_id": user.user_id, "email": user.email, "display_name": user.display_name, "role": "admin", "admin_access": user.admin_access, "learner_feedback_enabled": True, "account_status": status, "cohort": "owner", "affiliation": user.affiliation, "legal_name": user.legal_name, "department": user.department, "grade_level": user.grade_level, "phone": user.phone, "contact_address": user.contact_address, "nickname": user.nickname, "self_introduction": user.self_introduction, "avatar_url": user.avatar_url, "consent_at": datetime.fromtimestamp(1.0, tz=timezone.utc), "profile_completed_at": datetime.fromtimestamp(1.0, tz=timezone.utc), "terms_agreed_at": datetime.fromtimestamp(1.0, tz=timezone.utc), "privacy_agreed_at": datetime.fromtimestamp(1.0, tz=timezone.utc), "terms_version": user.terms_version, "privacy_version": user.privacy_version, "created_at": datetime.fromtimestamp(1.0, tz=timezone.utc), "last_seen_at": datetime.fromtimestamp(1.0, tz=timezone.utc), "is_active": active, } class FakeAcquire: def __init__(self, conn: Any) -> None: self.conn = conn async def __aenter__(self) -> Any: return self.conn async def __aexit__(self, exc_type: Any, exc: Any, tb: Any) -> None: return None class FakePool: def __init__(self, conn: Any) -> None: self.conn = conn def acquire(self) -> FakeAcquire: return FakeAcquire(self.conn) class AuthIdentityAliasTest(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self) -> None: auth_routes._oauth_states.clear() auth_sessions._sessions.clear() auth_sessions._users.clear() auth_sessions._email_index.clear() auth_sessions._auth_identity_alias_index.clear() auth_sessions._inactive_emails.clear() async def asyncTearDown(self) -> None: auth_routes._oauth_states.clear() auth_sessions._sessions.clear() auth_sessions._users.clear() auth_sessions._email_index.clear() auth_sessions._auth_identity_alias_index.clear() auth_sessions._inactive_emails.clear() async def test_explicit_alias_resolves_canonical_user_without_email_merge(self) -> None: class Conn: query = "" args: tuple[Any, ...] = () async def fetchrow(self, query: str, *args: Any) -> dict[str, Any]: self.query = query self.args = args return alias_row() conn = Conn() with patch.object(auth_sessions, "get_pool", return_value=FakePool(conn)): user = await auth_sessions.get_managed_user_by_auth_alias( "google:gmail-subject" ) self.assertIsNotNone(user) self.assertEqual(user.user_id, managed_user().user_id) self.assertEqual(user.email, "yunchan@twentyoz.kr") self.assertIn("app.auth_identity_alias", conn.query) self.assertEqual(conn.args, ("google:gmail-subject",)) async def test_alias_to_inactive_or_suspended_user_fails_closed(self) -> None: class Conn: def __init__(self, row: dict[str, Any]) -> None: self.row = row async def fetchrow(self, query: str, *args: Any) -> dict[str, Any]: return self.row for row in (alias_row(active=False), alias_row(status="suspended")): with ( self.subTest(row=row), patch.object(auth_sessions, "get_pool", return_value=FakePool(Conn(row))), ): with self.assertRaises(auth_sessions.InactiveUserError): await auth_sessions.get_managed_user_by_auth_alias( "google:gmail-subject" ) async def test_alias_session_keeps_login_email_and_canonical_entitlements(self) -> None: class Conn: query = "" args: tuple[Any, ...] = () async def execute(self, query: str, *args: Any) -> str: self.query = query self.args = args return "INSERT 0 1" conn = Conn() canonical = managed_user() with ( patched_settings( environment="prod", auth_super_admin_emails=[canonical.email], ), patch.object(auth_sessions, "get_pool", return_value=FakePool(conn)), ): _, session_user = await auth_sessions.create_session( email="yunchan8804@gmail.com", display_name="Yun Chan", role="admin", external_id="google:gmail-subject", account_status="approved", managed_user=canonical, ) self.assertEqual(session_user.user_id, canonical.user_id) self.assertEqual(session_user.email, "yunchan8804@gmail.com") self.assertTrue(session_user.super_admin) self.assertTrue(session_user.admin_access) self.assertIn("login_email", conn.query) self.assertEqual(conn.args[1], canonical.user_id) self.assertEqual(conn.args[5], "yunchan8804@gmail.com") async def test_restored_session_uses_login_email_but_canonical_super_admin(self) -> None: canonical = managed_user() class Conn: query = "" async def fetchrow(self, query: str, *args: Any) -> dict[str, Any]: self.query = query return { "expires_at": datetime.fromtimestamp( time.time() + 3600, tz=timezone.utc ), "user_id": canonical.user_id, "login_email": "yunchan8804@gmail.com", "canonical_email": canonical.email, "display_name": canonical.display_name, "role": "admin", "admin_access": True, "learner_feedback_enabled": True, "account_status": "approved", "cohort": "owner", "consent_at": datetime.fromtimestamp(1.0, tz=timezone.utc), "profile_completed_at": datetime.fromtimestamp( 1.0, tz=timezone.utc ), } async def execute(self, query: str, *args: Any) -> str: return "UPDATE 1" conn = Conn() with ( patched_settings(auth_super_admin_emails=[canonical.email]), patch.object(auth_sessions, "get_pool", return_value=FakePool(conn)), ): restored = await auth_sessions.get_session("opaque-session") self.assertIsNotNone(restored) self.assertEqual(restored.email, "yunchan8804@gmail.com") self.assertTrue(restored.super_admin) self.assertIn("s.login_email", conn.query) self.assertIn("canonical_email", conn.query) async def test_me_response_uses_canonical_entitlement_not_login_email(self) -> None: canonical = managed_user() session_user = auth_sessions.SessionUser( user_id=canonical.user_id, email="yunchan8804@gmail.com", display_name=canonical.display_name, role="admin", admin_access=True, super_admin=True, account_status="approved", cohort_ids=canonical.cohort_ids, consent_at=canonical.consent_at, profile_completed_at=canonical.profile_completed_at, expires_at=time.time() + 3600, ) with ( patched_settings(auth_super_admin_emails=[canonical.email]), patch.object( auth_routes, "get_managed_user", AsyncMock(return_value=canonical), ), ): response = await auth_routes._me_response(session_user) self.assertEqual(response.email, "yunchan8804@gmail.com") self.assertTrue(response.admin_access) self.assertTrue(response.super_admin) self.assertIsNotNone(response.onboarding_completed_at) async def test_google_callback_resolves_alias_before_email_and_redirects_admin( self, ) -> None: state = "gmail-alias-state" auth_routes._oauth_states[state] = auth_routes.OAuthState( code_verifier="verifier", next_path="/learn", created_at=time.time(), ) canonical = managed_user() class FakeResponse: status_code = 200 def __init__(self, payload: dict[str, Any]) -> None: self.payload = payload def json(self) -> dict[str, Any]: return self.payload class FakeAsyncClient: def __init__(self, *args: Any, **kwargs: Any) -> None: pass async def __aenter__(self) -> "FakeAsyncClient": return self async def __aexit__(self, exc_type: Any, exc: Any, tb: Any) -> None: return None async def post(self, url: str, **kwargs: Any) -> FakeResponse: return FakeResponse({"id_token": "id-token"}) async def get(self, url: str, **kwargs: Any) -> FakeResponse: return FakeResponse( { "aud": "google-client", "iss": "https://accounts.google.com", "email": "yunchan8804@gmail.com", "email_verified": True, "name": "Yun Chan", "sub": "gmail-subject", } ) request = Request( { "type": "http", "method": "GET", "path": "/auth/callback", "headers": [(b"host", b"api-vignette.test")], } ) create_session = AsyncMock(return_value=("opaque-session", object())) email_lookup = AsyncMock(side_effect=AssertionError("email lookup must not run")) with ( patched_settings( environment="prod", oauth_google_client_id="google-client", oauth_google_client_secret="google-secret", frontend_base_url="https://vignette.test", auth_super_admin_emails=[canonical.email], ), patch.object(auth_routes.httpx, "AsyncClient", FakeAsyncClient), patch.object( auth_routes, "get_managed_user_by_auth_alias", AsyncMock(return_value=canonical), ), patch.object(auth_routes, "get_managed_user_by_email", email_lookup), patch.object(auth_routes, "create_session", create_session), ): response = await auth_routes.callback( request, code="auth-code", state=state, oauth_state_cookie=state, ) self.assertEqual(response.headers["location"], "https://vignette.test/admin") email_lookup.assert_not_awaited() create_session.assert_awaited_once_with( email="yunchan8804@gmail.com", display_name="Yun Chan", role="admin", cohort_ids=["owner"], external_id="google:gmail-subject", account_status="approved", managed_user=canonical, ) def test_migration_makes_alias_runtime_read_only(self) -> None: migration = ( Path(__file__).resolve().parents[3] / "infra" / "db" / "init" / "19_auth_identity_alias.sql" ).read_text(encoding="utf-8") self.assertIn("CREATE TABLE IF NOT EXISTS app.auth_identity_alias", migration) self.assertIn("ENABLE ROW LEVEL SECURITY", migration) self.assertIn("FOR SELECT", migration) self.assertNotIn("FOR INSERT", migration) self.assertNotIn("FOR UPDATE", migration) self.assertNotIn("FOR DELETE", migration) self.assertIn("ADD COLUMN IF NOT EXISTS login_email", migration)