from __future__ import annotations

import os
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch

from backend.sql_catalog import server
from backend.sql_catalog.auth import (
    Auth0Authenticator,
    Auth0Settings,
    AuthError,
    AuthPrincipal,
)


class Auth0AuthenticatorTest(unittest.TestCase):
    def setUp(self) -> None:
        self.authenticator = Auth0Authenticator(
            Auth0Settings(
                domain="tenant.auth0.com",
                allowed_email_domains=frozenset({"pingcap.com"}),
                allowed_connections=frozenset({"google-oauth2"}),
            )
        )

    def authenticate(self, subject: str, **profile_overrides: object):
        profile = {
            "sub": subject,
            "email": "engineer@pingcap.com",
            "email_verified": True,
            "name": "PingCAP Engineer",
        }
        profile.update(profile_overrides)
        with patch.object(self.authenticator, "_userinfo", return_value=profile):
            return self.authenticator.authenticate("Bearer access-token")

    def test_accepts_verified_pingcap_google_identity(self) -> None:
        principal = self.authenticate("google-oauth2|123")

        self.assertEqual("engineer@pingcap.com", principal.email)
        self.assertEqual("google-oauth2", principal.connection)

    def test_rejects_non_google_connection(self) -> None:
        with self.assertRaises(AuthError) as error:
            self.authenticate("auth0|123")

        self.assertEqual(403, error.exception.status)
        self.assertEqual("account_not_allowed", error.exception.code)

    def test_rejects_email_outside_pingcap_domain(self) -> None:
        with self.assertRaises(AuthError) as error:
            self.authenticate("google-oauth2|123", email="engineer@example.com")

        self.assertEqual(403, error.exception.status)

    def test_rejects_unverified_email(self) -> None:
        with self.assertRaises(AuthError) as error:
            self.authenticate("google-oauth2|123", email_verified=False)

        self.assertEqual(403, error.exception.status)

    def test_requires_bearer_token(self) -> None:
        with self.assertRaises(AuthError) as error:
            self.authenticator.authenticate(None)

        self.assertEqual(401, error.exception.status)
        self.assertEqual("authentication_required", error.exception.code)


class ApplicationUserTest(unittest.TestCase):
    def setUp(self) -> None:
        self.temp_dir = tempfile.TemporaryDirectory()
        self.previous_db_path = server.DB_PATH
        self.previous_bootstrap_email = os.environ.get("AUTH0_BOOTSTRAP_EMAIL")
        os.environ["AUTH0_BOOTSTRAP_EMAIL"] = "catalog.owner@pingcap.com"
        server.DB_PATH = Path(self.temp_dir.name) / "auth.sqlite3"
        with server._AUTH_USER_CACHE_LOCK:
            server._AUTH_USER_CACHE.clear()
        server.init_db()

    def tearDown(self) -> None:
        server.DB_PATH = self.previous_db_path
        if self.previous_bootstrap_email is None:
            os.environ.pop("AUTH0_BOOTSTRAP_EMAIL", None)
        else:
            os.environ["AUTH0_BOOTSTRAP_EMAIL"] = self.previous_bootstrap_email
        with server._AUTH_USER_CACHE_LOCK:
            server._AUTH_USER_CACHE.clear()
        self.temp_dir.cleanup()

    def test_first_login_binds_bootstrap_user_and_request_context(self) -> None:
        principal = AuthPrincipal(
            subject="google-oauth2|bootstrap-user",
            email="catalog.owner@pingcap.com",
            email_verified=True,
            name="Shenjing Zhao",
            picture_url="https://example.com/avatar.png",
            connection="google-oauth2",
        )

        user = server.ensure_authenticated_user(principal)
        token = server.REQUEST_USER.set(user)
        try:
            payload = server.bind_request_actor(
                {"actor": "forged-user", "created_by": "forged-user"}
            )
            self.assertEqual(user["id"], payload["actor"])
            self.assertEqual(user["id"], payload["created_by"])
            self.assertEqual(user["id"], server.current_user_id())
        finally:
            server.REQUEST_USER.reset(token)

        with server.get_conn() as conn:
            stored = conn.execute(
                "SELECT * FROM users WHERE id = ?", (user["id"],)
            ).fetchone()
        self.assertEqual("admin", stored["role"])
        self.assertEqual(principal.subject, stored["auth0_subject"])
        self.assertEqual(1, stored["email_verified"])

    def test_user_directory_excludes_auth0_subject(self) -> None:
        directory = server.list_users()

        self.assertEqual(1, directory["total"])
        self.assertEqual("catalog.owner@pingcap.com", directory["items"][0]["email"])
        self.assertNotIn("auth0_subject", directory["items"][0])

    def test_bootstrap_moves_legacy_personal_scope_ownership(self) -> None:
        with server.get_conn() as conn:
            timestamp = server.now_iso()
            conn.execute(
                """
                INSERT INTO cluster_scopes
                    (id, name, description, owner_id, visibility, membership_mode,
                     definition_json, include_deleted, is_recommended, created_at, updated_at)
                VALUES ('scope-legacy', 'Legacy', '', ?, 'private', 'snapshot', '{}', 1, 0, ?, ?)
                """,
                (server.LEGACY_USER_ID, timestamp, timestamp),
            )
            server.bootstrap_auth_user(conn)
            scope = conn.execute(
                "SELECT owner_id FROM cluster_scopes WHERE id = 'scope-legacy'"
            ).fetchone()
            user = conn.execute(
                "SELECT id FROM users WHERE email = 'catalog.owner@pingcap.com'"
            ).fetchone()

        self.assertEqual(user["id"], scope["owner_id"])


if __name__ == "__main__":
    unittest.main()