"""OAuth 真实浏览器登录的 HTTP 端点测试。

不联网、不向真实 GitHub/Google 发请求：server 的注入点
（server.oauth_http_client / server.oauth_userinfo_client）挂 fake；
真实 userinfo 端点本身只由 cloud.oauth.default_userinfo_client
拼 URL，不在这里调真实网络。测试里出现的 token、key 一律是
fake 占位，绝无真实 client_secret、真实 token、真实 GitHub 账号。
"""

import http.client
import os
import re
import sys
import tempfile
import threading
import time
import unittest
from pathlib import Path

try:
    from cloud.auth import AuthStore
    from cloud.oauth import OAuthStateStore
    from cloud.server import create_server
except ImportError:
    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
    from cloud.auth import AuthStore
    from cloud.oauth import OAuthStateStore
    from cloud.server import create_server


def _fake_token_client(result=None, error=None):
    def fake(token_url, payload):
        if error is not None:
            return {"error": error}
        return dict(result if result is not None else
                    {"access_token": "fake-token-not-real"})
    return fake


def _fake_userinfo(result=None, exc=None):
    def fake(access_token):
        if exc is not None:
            raise exc
        return dict(result if result is not None else
                    {"id": 4242, "login": "gh-user",
                     "email": "gh-user@example.invalid"})
    return fake


class OAuthRealLoginTestCase(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.db_path = os.path.join(self.tmp.name, "control.db")
        self.auth = AuthStore(self.db_path)
        self.server = create_server(self.db_path, "127.0.0.1", 0)
        self.port = self.server.server_address[1]
        self.thread = threading.Thread(
            target=self.server.serve_forever, daemon=True)
        self.thread.start()
        # 默认挂 fake：token 响应故意不带身份字段，逼走 userinfo 路径。
        self.server.oauth_http_client = _fake_token_client()
        self.server.oauth_userinfo_client = _fake_userinfo()

    def tearDown(self):
        self.server.shutdown()
        self.server.server_close()
        self.thread.join(timeout=5)
        try:
            self.auth.conn.close()
        except Exception:
            pass
        self.tmp.cleanup()

    def _req(self, method, path, body=None):
        conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
        try:
            headers = {}
            payload = None
            if body is not None:
                import json
                payload = json.dumps(body).encode("utf-8")
                headers["Content-Type"] = "application/json"
            conn.request(method, path, body=payload, headers=headers)
            resp = conn.getresponse()
            data = resp.read().decode("utf-8", "replace")
        finally:
            conn.close()
        return resp.status, data

    def _login_state(self):
        status, _ = self._req(
            "GET",
            "/oauth/github/login?client_id=test-client"
            "&redirect_uri=http%3A%2F%2F127.0.0.1%2Fcb")
        self.assertEqual(status, 200)
        store = OAuthStateStore(self.db_path)
        try:
            row = store.conn.execute(
                "SELECT state FROM oauth_states ORDER BY created_at DESC"
                " LIMIT 1").fetchone()
        finally:
            store.conn.close()
        return row[0]

    def _api_key_in_page(self, page):
        match = re.search(r"<code>([^<]+)</code>", page)
        self.assertIsNotNone(match, page)
        return match.group(1)

    # -- GET 回调成功：发 key、key 可用、轮换旧 key 失效 -------------
    def test_get_callback_success_issues_key_and_rotation(self):
        state = self._login_state()
        status, page = self._req(
            "GET", "/oauth/github/callback?code=fake-code&state=" + state)
        self.assertEqual(status, 200, page)
        self.assertIn("gh-user", page)
        self.assertIn("new user: yes", page)
        self.assertNotIn("fake-token-not-real", page)
        self.assertNotIn("client_secret", page)
        key1 = self._api_key_in_page(page)
        user_id = self.auth.authenticate(key1)
        self.assertIsNotNone(user_id)

        # 同一身份第二次登录：新 key、旧 key 失效、审计有轮换记录。
        state2 = self._login_state()
        status, page2 = self._req(
            "GET", "/oauth/github/callback?code=fake-code&state=" + state2)
        self.assertEqual(status, 200, page2)
        self.assertIn("new user: no", page2)
        self.assertIn("rotated", page2)
        key2 = self._api_key_in_page(page2)
        self.assertNotEqual(key1, key2)
        self.assertIsNone(self.auth.authenticate(key1))
        self.assertEqual(self.auth.authenticate(key2), user_id)
        actions = [row["action"] for row in self.auth.list_audit("", limit=50)]
        self.assertIn("oauth_login", actions)
        self.assertIn("api_key_rotated", actions)

    def test_post_callback_success_returns_key_once(self):
        state = self._login_state()
        import json
        status, body = self._req(
            "POST", "/oauth/github/callback",
            {"code": "fake-code", "state": state,
             "client_id": "test-client",
             "redirect_uri": "http://127.0.0.1/cb"})
        parsed = json.loads(body)
        self.assertEqual(status, 200, body)
        self.assertIn("api_key", parsed)
        self.assertFalse(parsed["api_key_rotated"])
        self.assertEqual(self.auth.authenticate(parsed["api_key"]),
                         parsed["user_id"])

    # -- GET 回调失败路径 ----------------------------------------------
    def test_get_callback_bad_state_400_no_user(self):
        status, page = self._req(
            "GET", "/oauth/github/callback?code=x&state=no-such-state")
        self.assertEqual(status, 400)
        self.assertNotIn("fake-token-not-real", page)
        count = self.auth.conn.execute(
            "SELECT COUNT(*) FROM users").fetchone()[0]
        self.assertEqual(count, 0)

    def test_get_callback_expired_state_400(self):
        state = self._login_state()
        store = OAuthStateStore(self.db_path)
        try:
            store.conn.execute(
                "UPDATE oauth_states SET created_at = ? WHERE state = ?",
                (time.time() - 601, state))
            store.conn.commit()
        finally:
            store.conn.close()
        status, _ = self._req(
            "GET", "/oauth/github/callback?code=x&state=" + state)
        self.assertEqual(status, 400)

    def test_get_callback_state_provider_mismatch_400(self):
        state = self._login_state()
        status, _ = self._req(
            "GET", "/oauth/google/callback?code=x&state=" + state)
        self.assertEqual(status, 400)

    def test_get_callback_token_exchange_failure_400_no_user(self):
        self.server.oauth_http_client = _fake_token_client(
            error="bad_verification_code")
        state = self._login_state()
        status, _ = self._req(
            "GET", "/oauth/github/callback?code=bad&state=" + state)
        self.assertEqual(status, 400)
        count = self.auth.conn.execute(
            "SELECT COUNT(*) FROM users").fetchone()[0]
        self.assertEqual(count, 0)

    def test_get_callback_userinfo_failure_400_no_user(self):
        self.server.oauth_userinfo_client = _fake_userinfo(exc=RuntimeError("down"))
        state = self._login_state()
        status, page = self._req(
            "GET", "/oauth/github/callback?code=x&state=" + state)
        self.assertEqual(status, 400, page)
        self.assertIn("userinfo", page)
        count = self.auth.conn.execute(
            "SELECT COUNT(*) FROM users").fetchone()[0]
        self.assertEqual(count, 0)

    def test_get_callback_userinfo_empty_400_no_user(self):
        self.server.oauth_userinfo_client = _fake_userinfo(result={})
        state = self._login_state()
        status, _ = self._req(
            "GET", "/oauth/github/callback?code=x&state=" + state)
        self.assertEqual(status, 400)
        count = self.auth.conn.execute(
            "SELECT COUNT(*) FROM users").fetchone()[0]
        self.assertEqual(count, 0)

    def test_get_callback_secret_missing_400_before_network(self):
        # 真实路径（不注入 fake）：无 secret 时换 token 前就拒绝，
        # 不会走到真实网络；fake userinfo 在此路径下不被使用。
        self.server.oauth_http_client = None
        self.server.oauth_userinfo_client = None
        old = os.environ.pop("GITHUB_OAUTH_CLIENT_SECRET", None)
        try:
            state = self._login_state()
            status, page = self._req(
                "GET", "/oauth/github/callback?code=x&state=" + state)
        finally:
            if old is not None:
                os.environ["GITHUB_OAUTH_CLIENT_SECRET"] = old
        self.assertEqual(status, 400, page)
        self.assertIn("client_secret", page)

    # -- AuthStore 轮换本身 ---------------------------------------------
    def test_rotate_api_key_old_key_dead(self):
        user_id, old_key = self.auth.create_user("someone")
        new_key = self.auth.rotate_api_key(user_id)
        self.assertNotEqual(old_key, new_key)
        self.assertIsNone(self.auth.authenticate(old_key))
        self.assertEqual(self.auth.authenticate(new_key), user_id)
        with self.assertRaises(ValueError):
            self.auth.rotate_api_key("no-such-user")


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