feat: web command console v1 (GET /console) (#57)

Co-authored-by: host <host@test.local>
This commit is contained in:
openqjandhost authored and GitHub committed 2026-10-27 01:02:24 +08:00
1 parent c6e633d162
commit e40c2c33bb
4 files changed
+441

No files matched your search

+91
View File
@@ -0,0 +1,91 @@
"""命令台 /console 单测:只验页面壳与路由,不跑浏览器 JS。
页面本身不走鉴权(只是个壳),数据一律凭 X-API-Key 调接口,
所以 /console 无 key 也应 200 出 HTML;真正的鉴权由各 API
端点负责(已有单测覆盖)。
"""
import http.client
import os
import sys
import tempfile
import threading
import unittest
from pathlib import Path
try:
from cloud.auth import AuthStore
from cloud.server import create_server
except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from cloud.auth import AuthStore
from cloud.server import create_server
class ConsoleTestCase(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.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()
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 _get(self, path, key=None):
conn = http.client.HTTPConnection(
"127.0.0.1", self.port, timeout=10)
headers = {}
if key is not None:
headers["X-API-Key"] = key
try:
conn.request("GET", path, headers=headers)
resp = conn.getresponse()
return resp.status, dict(resp.getheaders()), resp.read()
finally:
conn.close()
def test_console_serves_html_without_key(self):
status, headers, body = self._get("/console")
self.assertEqual(status, 200)
self.assertIn("text/html", headers.get("Content-Type", ""))
text = body.decode("utf-8")
self.assertIn("Codex 命令台", text)
# 壳必须引用它实际调的接口,防 JS 与服务端路由漂移。
for marker in ("/health", "/projects", "/tasks", "/workers",
"X-API-Key"):
self.assertIn(marker, text)
def test_console_trailing_slash_same_page(self):
status, _headers, body = self._get("/console/")
self.assertEqual(status, 200)
self.assertIn("Codex 命令台", body.decode("utf-8"))
def test_console_subpath_not_routed(self):
# 未知子路径先过鉴权(无 key 时 401),有 key 时才落到 404;
# 绝不能把 /console/extra 也当成壳 serve 出去。
status, _headers, _body = self._get("/console/extra")
self.assertEqual(status, 401)
status, _headers, _body = self._get("/console/extra", self.owner_key)
self.assertEqual(status, 404)
def test_console_does_not_leak_data_without_key(self):
# 壳里只有静态文本,没有任何任务/项目数据(数据须凭 key 调 API)。
_status, _headers, body = self._get("/console")
text = body.decode("utf-8")
self.assertNotIn(self.owner_id, text)
if __name__ == "__main__":
unittest.main()