admin/codex-app-server
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
code-to10-gapfill2
codex-app-server/cloud/tests/test_p1_security_server.py
T
host 935b530f84 split: handler mixins for oauth/identity/project/automation/fullauto (94 methods) on integration base 02835f6 
- CloudControlPlaneHandler mixes in OAuthMixin, IdentityMemoryMixin,
  ProjectMixin, AutomationMixin, FullAutoMixin; shared helpers move to
  cloud/handler_support.py with server.py re-exports (old import paths
  keep working, incl. _validate_env_branch for tests).
- Method bodies moved verbatim; routers, base plumbing, task/lifecycle,
  webhooks stay in server.py. No >100-line method body split.
- test_handler_split_conformance.py (6 tests) pins resolution, layer
  bans and method accounting (238 total, 144 + 94, no duplicates).
- test_p1_security_server.py: monkeypatch targets follow the defining
  module (handler_project MAX_PROJECTS_PER_USER, handler_identity
  relay helpers); assertions unchanged.

(cherry picked from commit 92165a2b097126470e4827327d96d592a253adf1)
2026-11-03 02:49:26 +00:00

608 lines
26 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""P1 控制面/控制台单测（server.py / console.html / sub2api.py / push.py）。
只覆盖本批 P1 控制面改动，不跑真实模型回合、不联网、不碰生产：
server 侧走本地临时端口 + 临时 SQLite；网关侧用假 request /
假 admin（key、token 全是测试假值）；推送 DNS 用注入的
resolver。绝不改 worker.py / coordinator.py / cost.py / host.py。
"""
import http.client
import json
import os
import sqlite3
import stat
import sys
import tempfile
import threading
import unittest
from pathlib import Path
from unittest import mock
try:
from cloud.auth import AuthStore
from cloud.push import dispatch_event, validate_push_target
from cloud.push import create_subscription
from cloud.server import create_server
from cloud import server as server_mod
# 拆分后补丁目标跟随方法定义模块：项目上限在 handler_project，
# relay 模型查询在 handler_identity（方法查本模块全局变量，
# 补丁 cloud.server 已不再生效，见 code-to10-v2-split 摘要）。
from cloud import handler_identity as handler_identity_mod
from cloud import handler_project as handler_project_mod
from cloud.sub2api import Sub2apiError
from cloud.sub2api import ensure_gateway_keys_table
from cloud.sub2api import get_user_gateway_key
from cloud.sub2api import rotate_user_gateway_key
from cloud.sub2api import validate_or_rotate_gateway_key
except ImportError:
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from cloud.auth import AuthStore
from cloud.push import dispatch_event, validate_push_target
from cloud.push import create_subscription
from cloud.server import create_server
from cloud import server as server_mod
from cloud import handler_identity as handler_identity_mod
from cloud import handler_project as handler_project_mod
from cloud.sub2api import Sub2apiError
from cloud.sub2api import ensure_gateway_keys_table
from cloud.sub2api import get_user_gateway_key
from cloud.sub2api import rotate_user_gateway_key
from cloud.sub2api import validate_or_rotate_gateway_key
def _req(port, method, path, body=None, key=None, raw_body=None):
conn = http.client.HTTPConnection("127.0.0.1", port, timeout=10)
hdrs = {}
if raw_body is not None:
payload = raw_body
elif body is not None:
payload = json.dumps(body).encode("utf-8")
hdrs["Content-Type"] = "application/json"
else:
payload = None
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 = data.decode("utf-8", "replace")
return resp.status, parsed
class _FakeGatewayAdmin:
"""假网关管理员：建号/授余额只记调用，不联网。"""
def __init__(self):
self.created = []
self.deleted = []
self.balances = []
def create_user(self, email, password, username):
sub_id = 1000 + len(self.created)
self.created.append((email, username))
return sub_id
def delete_user(self, sub_user_id):
self.deleted.append(sub_user_id)
return True
def set_balance(self, sub_user_id, amount):
self.balances.append((sub_user_id, amount))
def resolve_group_id(self):
return 2
class _FakeGatewayRequest:
"""假网关 HTTP：models 状态可调，其余走固定成功路径。"""
def __init__(self, models_status=200):
self.models_status = models_status
self.calls = []
def __call__(self, method, url, body=None, token=None):
self.calls.append((method, url))
if url.endswith("/v1/models"):
return self.models_status, {"data": []}
if url.endswith("/api/v1/auth/login"):
return 200, {"data": {"access_token": "fake-user-token",
"refresh_token": "fake-refresh",
"expires_in": 3600}}
if url.endswith("/api/v1/keys"):
return 201, {"data": {"id": "fake-key-id-1",
"key": "fake-gateway-key-1"}}
if "/balance" in url:
return 200, {"data": {}}
return 200, {"data": {}}
class Sub2apiP1TestCase(unittest.TestCase):
def setUp(self):
self.conn = sqlite3.connect(":memory:")
ensure_gateway_keys_table(self.conn)
self.admin = _FakeGatewayAdmin()
# 假 origin：请求本身全是 fake，不产生真实网络调用。
self._old_origin = os.environ.get("SUB2API_BASE_URL")
os.environ["SUB2API_BASE_URL"] = "http://127.0.0.1:9"
def tearDown(self):
self.conn.close()
if self._old_origin is None:
os.environ.pop("SUB2API_BASE_URL", None)
else:
os.environ["SUB2API_BASE_URL"] = self._old_origin
def _seed_row(self, key="fake-old-key"):
self.conn.execute(
"INSERT INTO user_gateway_keys(user_id, sub2api_user_id,"
" api_key, refresh_token, created_at, updated_at)"
" VALUES(?, ?, ?, ?, ?, ?)",
("u1", 42, key, "fake-refresh", 1.0, 1.0))
self.conn.commit()
def test_validate_no_row_provisions(self):
req = _FakeGatewayRequest()
status, key, sub_id, key_id = validate_or_rotate_gateway_key(
self.conn, "u1", admin=self.admin, request=req)
self.assertEqual(status, "provisioned")
self.assertEqual(key, "fake-gateway-key-1")
self.assertEqual(key_id, "fake-key-id-1")
self.assertIsNotNone(get_user_gateway_key(self.conn, "u1"))
def test_validate_healthy_key_not_rotated(self):
self._seed_row()
req = _FakeGatewayRequest(models_status=200)
status, key, _sub, _kid = validate_or_rotate_gateway_key(
self.conn, "u1", admin=self.admin, request=req)
self.assertEqual(status, "ok")
self.assertEqual(key, "fake-old-key")
self.assertEqual(self.admin.created, [])
def test_validate_dead_key_rotates_not_pending(self):
self._seed_row()
req = _FakeGatewayRequest(models_status=403)
status, key, _sub, _kid = validate_or_rotate_gateway_key(
self.conn, "u1", admin=self.admin, request=req)
self.assertEqual(status, "rotated")
self.assertEqual(key, "fake-gateway-key-1")
row = get_user_gateway_key(self.conn, "u1")
self.assertEqual(row["api_key"], "fake-gateway-key-1")
self.assertNotEqual(row["api_key"], "fake-old-key")
def test_validate_zero_balance_403_rotates(self):
# 余额为 0 时网关鉴权层回 403：与 401 同属死 key，必须轮换。
self._seed_row()
req = _FakeGatewayRequest(models_status=403)
status, _key, _sub, _kid = validate_or_rotate_gateway_key(
self.conn, "u1", admin=self.admin, request=req)
self.assertEqual(status, "rotated")
def test_validate_gateway_down_keeps_old_row(self):
self._seed_row()
req = _FakeGatewayRequest(models_status=500)
with self.assertRaises(Sub2apiError):
validate_or_rotate_gateway_key(
self.conn, "u1", admin=self.admin, request=req)
row = get_user_gateway_key(self.conn, "u1")
self.assertEqual(row["api_key"], "fake-old-key")
def test_rotate_replaces_row(self):
self._seed_row()
req = _FakeGatewayRequest()
status, key, sub_id, key_id = rotate_user_gateway_key(
self.conn, "u1", admin=self.admin, request=req)
self.assertEqual(status, "rotated")
self.assertEqual(key_id, "fake-key-id-1")
self.assertEqual(
get_user_gateway_key(self.conn, "u1")["api_key"], key)
class PushP1TestCase(unittest.TestCase):
def setUp(self):
self.conn = sqlite3.connect(":memory:")
def tearDown(self):
self.conn.close()
def test_literal_private_and_loopback_blocked(self):
for target in (
"http://127.0.0.1:9/hook",
"http://10.0.0.5/hook",
"http://172.16.4.4/hook",
"http://192.168.1.10/hook",
"http://169.254.169.254/latest/meta-data/",
"http://[::1]/hook"):
self.assertIsNotNone(
validate_push_target(target), target)
def test_metadata_hostname_blocked(self):
self.assertIsNotNone(validate_push_target(
"http://metadata.google.internal/hook"))
def test_hostname_resolved_to_private_blocked(self):
err = validate_push_target(
"https://hooks.example.com/hook",
resolver=lambda host: ["10.1.2.3"])
self.assertIsNotNone(err)
def test_hostname_public_resolves_ok(self):
err = validate_push_target(
"https://hooks.example.com/hook",
resolver=lambda host: ["93.184.216.34"])
self.assertIsNone(err)
def test_dns_failure_rejected(self):
def _boom(host):
raise OSError("no dns in test")
self.assertIsNotNone(validate_push_target(
"https://hooks.example.com/hook", resolver=_boom))
def test_dispatch_loopback_never_posts(self):
# 真实发送路径（sender=None，生产服务端走的路径）：
# loopback 目标在预检即被挡，不产生任何 POST。
create_subscription(self.conn, "proj-a", "user-1", "webhook",
"http://127.0.0.1:9/hook", "instant")
result = dispatch_event(self.conn, "failed", "proj-a",
{"task_id": "t1"}, sender=None)
self.assertEqual(result["sent"], [])
self.assertEqual(len(result["errors"]), 1)
self.assertIn("blocked", result["errors"][0]["reason"])
class ServerP1TestCase(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.member_id, self.member_key = self.auth.create_user("member")
self.outsider_id, self.outsider_key = self.auth.create_user(
"outsider")
self._old_admins = os.environ.get("CODEX_GLOBAL_ADMINS")
self._old_gateway = os.environ.get("CLOUD_GATEWAY")
self._old_backup_root = os.environ.get("CLOUD_BACKUP_ROOT")
self._old_backup_dir = os.environ.get("CLOUD_BACKUP_DIR")
os.environ["CODEX_GLOBAL_ADMINS"] = self.owner_id
os.environ.pop("CLOUD_GATEWAY", None)
os.environ.pop("CLOUD_BACKUP_ROOT", None)
os.environ.pop("CLOUD_BACKUP_DIR", None)
self._old_max_projects = handler_project_mod.MAX_PROJECTS_PER_USER
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()
s, proj = _req(self.port, "POST", "/projects",
{"name": "owner/repo"}, self.owner_key)
self.assertEqual(s, 201, proj)
self.project_id = proj["id"]
self.auth.set_member(
self.project_id, self.owner_id, self.member_id, "maintainer")
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()
handler_project_mod.MAX_PROJECTS_PER_USER = self._old_max_projects
if self._old_admins is None:
os.environ.pop("CODEX_GLOBAL_ADMINS", None)
else:
os.environ["CODEX_GLOBAL_ADMINS"] = self._old_admins
if self._old_gateway is None:
os.environ.pop("CLOUD_GATEWAY", None)
else:
os.environ["CLOUD_GATEWAY"] = self._old_gateway
if self._old_backup_root is None:
os.environ.pop("CLOUD_BACKUP_ROOT", None)
else:
os.environ["CLOUD_BACKUP_ROOT"] = self._old_backup_root
if self._old_backup_dir is None:
os.environ.pop("CLOUD_BACKUP_DIR", None)
else:
os.environ["CLOUD_BACKUP_DIR"] = self._old_backup_dir
# 恢复猴子补丁过的模块函数（relay 在 handler_identity，
# rotate 仍在 server；补丁目标跟随方法定义模块）。
for mod, name in ((handler_identity_mod, "_validate_sub2api_key"),
(handler_identity_mod, "_list_sub2api_models"),
(server_mod, "_rotate_sub2api_key")):
try:
mock.patch.object(mod, name).stop()
except Exception:
pass
# -- helpers ---------------------------------------------------
def _audit_actions(self, user_id=None):
conn = sqlite3.connect(self.db_path)
try:
rows = conn.execute(
"SELECT action, detail FROM audit_log"
+ (" WHERE user_id = ?" if user_id else ""),
(user_id,) if user_id else ()).fetchall()
finally:
conn.close()
return rows
def _set_quota(self, target, quota):
return _req(self.port, "POST", "/admin/quotas",
{"user_id": target, "quota_usd_micros": quota},
self.owner_key)
# -- 备份 -------------------------------------------------------
def test_backup_whitelist_and_mode(self):
root = os.path.join(self.tmp.name, "backup-root")
os.environ["CLOUD_BACKUP_ROOT"] = root
out_dir = os.path.join(root, "daily")
s, body = _req(self.port, "POST", "/admin/backup",
{"out_dir": out_dir}, self.owner_key)
self.assertEqual(s, 200, body)
self.assertTrue(os.path.exists(body["path"]))
self.assertEqual(stat.S_IMODE(os.stat(body["path"]).st_mode),
0o600)
details = [d for _a, d in self._audit_actions(self.owner_id)]
self.assertTrue(any("admin_backup" in a
for a, _d in self._audit_actions(
self.owner_id)))
self.assertTrue(any(out_dir in (d or "")
for d in details))
# 白名单外：系统目录与 .. 逃逸一律 400。
for bad in ("/etc", os.path.join(root, "..", "escape")):
s, body = _req(self.port, "POST", "/admin/backup",
{"out_dir": bad}, self.owner_key)
self.assertEqual(s, 400, (bad, body))
# -- workers 聚合 ------------------------------------------------
def test_workers_admin_full_member_aggregate_only(self):
s, _b = _req(self.port, "POST", "/workers/heartbeat",
{"worker_id": "w-p1", "executor": "docker",
"capacity": 2, "running": 0}, self.owner_key)
self.assertEqual(s, 200, _b)
s, body = _req(self.port, "GET", "/workers", key=self.owner_key)
self.assertEqual(s, 200)
self.assertIn("workers", body)
self.assertTrue(any(w.get("worker_id") == "w-p1"
for w in body["workers"]))
s, body = _req(self.port, "GET", "/workers", key=self.member_key)
self.assertEqual(s, 200)
self.assertNotIn("workers", body)
self.assertNotIn("w-p1", json.dumps(body))
self.assertEqual(body["workers_total"], 1)
self.assertEqual(body["workers_online"], 1)
self.assertEqual(body["workers_offline"], 0)
# -- github repos 过滤 -------------------------------------------
def test_github_repos_filtered_by_member_projects(self):
s, proj = _req(self.port, "POST", "/projects",
{"name": "with-repo",
"repo_url": "https://github.com/acme/alpha.git"},
self.owner_key)
self.assertEqual(s, 201, proj)
self.auth.set_member(proj["id"], self.owner_id, self.member_id,
"maintainer")
conn = sqlite3.connect(self.db_path)
try:
conn.execute(
"CREATE TABLE IF NOT EXISTS github_app_installations("
"installation_id TEXT, account_login TEXT,"
" repo_full_name TEXT, created_at REAL,"
" UNIQUE(installation_id, repo_full_name))")
conn.execute(
"INSERT OR IGNORE INTO github_app_installations"
" VALUES(?, ?, ?, ?)", ("7", "acme", "acme/alpha", 1.0))
conn.execute(
"INSERT OR IGNORE INTO github_app_installations"
" VALUES(?, ?, ?, ?)", ("7", "acme", "acme/beta", 1.0))
conn.commit()
finally:
conn.close()
s, body = _req(self.port, "GET", "/github/repos",
key=self.member_key)
self.assertEqual(s, 200, body)
names = [r["repo_full_name"] for r in body["repos"]]
self.assertEqual(names, ["acme/alpha"])
s, body = _req(self.port, "GET", "/github/repos",
key=self.outsider_key)
self.assertEqual(s, 200, body)
self.assertEqual(body["repos"], [])
# -- key 审计 -----------------------------------------------------
def test_api_key_audit_only_id_and_label(self):
s, body = _req(self.port, "POST", "/users/me/keys",
{"label": "p1-test"}, self.owner_key)
self.assertEqual(s, 201, body)
key_id = body["id"]
plain = body["api_key"]
s, _b = _req(self.port, "DELETE",
"/users/me/keys/" + key_id, key=self.owner_key)
self.assertEqual(s, 200, _b)
s, body = _req(self.port, "POST",
"/users/me/keys/revoke-others", {},
self.owner_key)
self.assertEqual(s, 200, body)
actions = {a for a, _d in self._audit_actions(self.owner_id)}
self.assertIn("api_key_created", actions)
self.assertIn("api_key_revoked", actions)
self.assertIn("api_key_revoke_others", actions)
blob = json.dumps(self._audit_actions(self.owner_id))
self.assertNotIn(plain, blob)
# -- 项目上限 ------------------------------------------------------
def test_project_limits(self):
s, body = _req(self.port, "POST", "/projects",
{"name": "x" * 101}, self.owner_key)
self.assertEqual(s, 400, body)
self.assertIn("100", body["error"])
s, body = _req(self.port, "POST", "/projects",
{"name": "ok", "id": "y" * 65}, self.owner_key)
self.assertEqual(s, 400, body)
self.assertIn("64", body["error"])
handler_project_mod.MAX_PROJECTS_PER_USER = 1
# setUp 已建 1 个 owner 项目，限额 1 时再建即超限。
s, body = _req(self.port, "POST", "/projects",
{"name": "second"}, self.owner_key)
self.assertEqual(s, 400, body)
self.assertIn("max 1", body["error"])
# -- 协调员预留 ----------------------------------------------------
def test_coordinator_reserve_blocks_parent(self):
s, _b = self._set_quota(self.owner_id, 1000000)
self.assertEqual(s, 200, _b)
parts = [{"name": "a", "prompt": "做 a"},
{"name": "b", "prompt": "做 b"},
{"name": "c", "prompt": "做 c"}]
s, body = _req(self.port, "POST", "/tasks",
{"project_id": self.project_id, "prompt": "协调",
"coordinator_parts": parts}, self.owner_key)
self.assertEqual(s, 402, body)
self.assertEqual(body["error"],
"coordinator_quota_reserve_exceeded")
self.assertEqual(body["coordinator_parts"], 3)
s, _b = self._set_quota(self.owner_id, 10000000)
self.assertEqual(s, 200, _b)
s, body = _req(self.port, "POST", "/tasks",
{"project_id": self.project_id, "prompt": "协调",
"coordinator_parts": parts}, self.owner_key)
self.assertEqual(s, 201, body)
self.assertIn("parent task creator",
body.get("quota_attribution", ""))
self.assertEqual(body.get("quota_reserved_usd_micros"), 3000000)
# -- 自动化配额门 ----------------------------------------------------
def test_automation_create_quota_gate(self):
s, _b = self._set_quota(self.owner_id, 0)
self.assertEqual(s, 200, _b)
s, body = _req(self.port, "POST",
"/projects/%s/automations" % self.project_id,
{"name": "nightly", "prompt": "巡检"},
self.owner_key)
self.assertEqual(s, 402, body)
s, _b = self._set_quota(self.owner_id, 10000000)
self.assertEqual(s, 200, _b)
s, body = _req(self.port, "POST",
"/projects/%s/automations" % self.project_id,
{"name": "nightly", "prompt": "巡检"},
self.owner_key)
self.assertEqual(s, 201, body)
# -- sub2api 懒开通 / 轮换端点 --------------------------------------
def test_relay_models_lazy_provision_not_just_pending(self):
os.environ["CLOUD_GATEWAY"] = "sub2api"
orig_validate = handler_identity_mod._validate_sub2api_key
orig_list = handler_identity_mod._list_sub2api_models
try:
handler_identity_mod._validate_sub2api_key = (
lambda conn, uid: ("provisioned", "fake-key", 7, 9))
handler_identity_mod._list_sub2api_models = (
lambda origin, key: ["model-a"])
s, body = _req(self.port, "GET", "/relay/models",
key=self.owner_key)
self.assertEqual(s, 200, body)
self.assertEqual(body["models"], ["model-a"])
self.assertEqual(body["provisioning"], "provisioned")
handler_identity_mod._validate_sub2api_key = (
lambda conn, uid: (_ for _ in ()).throw(
Sub2apiError("网关暂不可用")))
s, body = _req(self.port, "GET", "/relay/models",
key=self.owner_key)
self.assertEqual(s, 502, body)
finally:
handler_identity_mod._validate_sub2api_key = orig_validate
handler_identity_mod._list_sub2api_models = orig_list
def test_gateway_rotate_endpoint(self):
s, body = _req(self.port, "POST", "/users/me/gateway/rotate",
{}, self.owner_key)
self.assertEqual(s, 400, body)
os.environ["CLOUD_GATEWAY"] = "sub2api"
orig_rotate = server_mod._rotate_sub2api_key
try:
server_mod._rotate_sub2api_key = (
lambda conn, uid: ("rotated", "fake-key", 7, 9))
s, body = _req(self.port, "POST", "/users/me/gateway/rotate",
{}, self.owner_key)
self.assertEqual(s, 200, body)
self.assertTrue(body["rotated"])
self.assertNotIn("api_key", body)
actions = {a for a, _d in self._audit_actions(self.owner_id)}
self.assertIn("gateway_key_rotated", actions)
finally:
server_mod._rotate_sub2api_key = orig_rotate
# -- 推送审计与分发 SSRF ----------------------------------------------
def test_push_audit_and_dispatch_ssrf(self):
s, sub = _req(self.port, "POST",
"/projects/%s/push/subscriptions" % self.project_id,
{"channel": "webhook",
"target": "https://example.com/hook",
"preference": "instant"}, self.owner_key)
self.assertEqual(s, 201, sub)
actions = {a for a, _d in self._audit_actions(self.owner_id)}
self.assertIn("push_subscription_created", actions)
s, _b = _req(self.port, "DELETE",
"/projects/%s/push/subscriptions/%s"
% (self.project_id, sub["id"]), key=self.owner_key)
self.assertEqual(s, 200, _b)
actions = {a for a, _d in self._audit_actions(self.owner_id)}
self.assertIn("push_subscription_deleted", actions)
# 内网目标：分发不 POST、sent 为空、错误含 delivery failed。
s, sub = _req(self.port, "POST",
"/projects/%s/push/subscriptions" % self.project_id,
{"channel": "webhook",
"target": "http://127.0.0.1:9/hook",
"preference": "instant"}, self.owner_key)
self.assertEqual(s, 201, sub)
s, body = _req(self.port, "POST", "/admin/push/dispatch",
{"event_type": "failed",
"project_id": self.project_id,
"payload": {"task_id": "t1"}}, self.owner_key)
self.assertEqual(s, 200, body)
self.assertEqual(body["sent"], [])
self.assertEqual(len(body["errors"]), 1)
self.assertIn("delivery failed", body["errors"][0]["reason"])
actions = {a for a, _d in self._audit_actions(self.owner_id)}
self.assertIn("admin_push_dispatch", actions)
# -- 配额双字段与控制台文案 --------------------------------------------
def test_quota_payload_has_unpriced_fields(self):
s, body = _req(self.port, "GET", "/users/me/quota",
key=self.owner_key)
self.assertEqual(s, 200, body)
self.assertIn("unpriced_tokens_used", body)
self.assertIn("unpriced_token_cap", body)
def test_console_p1_markers(self):
conn = http.client.HTTPConnection(
"127.0.0.1", self.port, timeout=10)
try:
conn.request("GET", "/console")
resp = conn.getresponse()
text = resp.read().decode("utf-8")
finally:
conn.close()
self.assertEqual(resp.status, 200)
for marker in (
"登记评审需求",
"记录允许",
"记录拒绝",
"当前版本只留回执，不拦截实时行为",
"模型回合结束、改动在 worktree 里尚未合并",
"控制面已标记 merged，不代表 git 主分支已合入",
"协调员与 goal/计划互斥",
"仅 failed/interrupted/merge-failed 可重试",
"仅 maintainer 可合并"):
self.assertIn(marker, text)
if __name__ == "__main__":
unittest.main()
Reference in new issue View Git Blame Copy Permalink