"""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.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() 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()