codex-app-server/cloud/tests/test_server_oauth_push.py

"""OAuth 与推送的 HTTP 端点测试。

不联网、不向真实 GitHub/Google/推送服务发请求；OAuth 回调成功
路径由模块级单测覆盖，这里只测 HTTP 层的失败路径与订阅分发。
"""

import http.client
import json
import os
import sys
import tempfile
import threading
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


class OAuthPushEndpointsTestCase(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.owner_id, self.owner_key = self.auth.create_user("owner")
        self._old_admins = os.environ.get("CODEX_GLOBAL_ADMINS")
        os.environ["CODEX_GLOBAL_ADMINS"] = self.owner_id
        self.outsider_id, self.outsider_key = self.auth.create_user("outsider")
        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()
        status, self.project = self._req(
            "POST", "/projects", {"name": "demo"}, self.owner_key)
        self.assertEqual(status, 201)

    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()
        if getattr(self, "_old_admins", None) is None:
            os.environ.pop("CODEX_GLOBAL_ADMINS", None)
        else:
            os.environ["CODEX_GLOBAL_ADMINS"] = self._old_admins

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

    # -- OAuth -----------------------------------------------------------
    def test_oauth_login_returns_url_without_secret(self):
        status, body = self._req(
            "GET",
            "/oauth/github/login?client_id=test-client"
            "&redirect_uri=http%3A%2F%2F127.0.0.1%2Fcb")
        self.assertEqual(status, 200)
        self.assertIn("github.com/login/oauth/authorize",
                      body["authorization_url"])
        self.assertIn("test-client", body["authorization_url"])
        self.assertNotIn("secret", body["authorization_url"])
        self.assertTrue(body["state"])

    def test_oauth_login_unknown_provider_400(self):
        status, _ = self._req(
            "GET",
            "/oauth/nope/login?client_id=x"
            "&redirect_uri=http%3A%2F%2F127.0.0.1%2Fcb")
        self.assertEqual(status, 400)

    def test_oauth_callback_bad_state_400_no_session(self):
        status, body = self._req(
            "POST", "/oauth/github/callback",
            {"code": "fake-code", "state": "no-such-state",
             "client_id": "test-client",
             "redirect_uri": "http://127.0.0.1/cb"})
        self.assertEqual(status, 400)
        self.assertIn("state", body["error"])

    def test_oauth_callback_good_state_no_secret_400(self):
        store = OAuthStateStore(self.db_path)
        try:
            state = store.create_state(
                "github", "http://127.0.0.1/cb")
        finally:
            store.conn.close()
        # 确保环境里没有真实 secret（本测试不依赖它）。
        old = os.environ.pop("GITHUB_OAUTH_CLIENT_SECRET", None)
        try:
            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"})
        finally:
            if old is not None:
                os.environ["GITHUB_OAUTH_CLIENT_SECRET"] = old
        self.assertEqual(status, 400)
        self.assertIn("client_secret", body["error"])

    # -- 推送 -------------------------------------------------------------
    def test_push_subscribe_list_dispatch_and_delete(self):
        pid = self.project["id"]
        status, sub = self._req(
            "POST", "/projects/%s/push/subscriptions" % pid,
            {"channel": "webhook", "target": "http://127.0.0.1:1/hook",
             "preference": "instant"},
            self.owner_key)
        self.assertEqual(status, 201, sub)
        status, body = self._req(
            "GET", "/projects/%s/push/subscriptions" % pid,
            key=self.owner_key)
        self.assertEqual(status, 200)
        self.assertEqual(len(body["subscriptions"]), 1)
        # 内置真实发送端会真 POST：目标不可达时分发应报投递失败
        # （不是“无 sender”，也不是静默成功），sent 为空。
        status, body = self._req(
            "POST", "/admin/push/dispatch",
            {"event_type": "failed", "project_id": pid,
             "payload": {"task_id": "t1"}},
            self.owner_key)
        self.assertEqual(status, 200)
        self.assertEqual(body["sent"], [])
        self.assertEqual(len(body["errors"]), 1)
        self.assertIn("delivery failed", body["errors"][0]["reason"])
        # outsider 不能往别人项目里订阅/分发。
        status, _ = self._req(
            "POST", "/projects/%s/push/subscriptions" % pid,
            {"channel": "webhook", "target": "https://example.invalid/x"},
            self.outsider_key)
        self.assertEqual(status, 403)
        # 删除：DELETE 方法须可用，且删后列表为空。
        conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
        try:
            conn.request("DELETE",
                         "/projects/%s/push/subscriptions/%s" % (pid, sub["id"]),
                         headers={"X-API-Key": self.owner_key})
            resp = conn.getresponse()
            resp.read()
            self.assertEqual(resp.status, 200)
        finally:
            conn.close()
        status, body = self._req(
            "GET", "/projects/%s/push/subscriptions" % pid,
            key=self.owner_key)
        self.assertEqual(body["subscriptions"], [])

    def test_push_dispatch_requires_auth(self):
        status, _ = self._req(
            "POST", "/admin/push/dispatch",
            {"event_type": "completed", "project_id": self.project["id"]})
        self.assertEqual(status, 401)


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