host
1141d0d0f8
仓库版 ExecStart 只带 --websearch-mcp，生产主机现行 unit 在此之外 还带 --seed-skills --imagegen-mcp --playwright-mcp；按仓库重装会静默 丢掉技能播种与两路 MCP。同步后仓库 ExecStart 与主机现行 ExecStart 逐 token 一致，并补 deploy/tests/test_autoscaler_unit.py（5 项）拦 后续漂移：四项齐全、无重复、均在 cloud/worker.py 有定义、不在 scaler 侧、--docker-runtime 仍为 runsc。 (cherry picked from commit bc3a23fc068cac189c0173b8e1c2befa7b832ed2)
95 lines
3.8 KiB
Python
95 lines
3.8 KiB
Python
"""autoscaler unit 与生产主机现行 unit 的同步回归。
|
||
|
||
背景：仓库版 deploy/autoscaler/codex-autoscaler.service 的 ExecStart
|
||
长期只带 --websearch-mcp，而生产主机现行 unit 在此之外还带
|
||
--seed-skills --imagegen-mcp --playwright-mcp（2026-11-01 主机只读核对：
|
||
主机 ExecStart = 仓库版 ExecStart + 这三项，别无差异）。
|
||
仓库版滞后意味着按仓库重装会静默丢掉技能播种与两路 MCP。
|
||
|
||
本文件只读仓库内两个文件，不碰主机、不联网、不跑 worker：
|
||
- unit 的 worker 侧（`--` 之后）必须含全部四项 MCP/技能播种 flag；
|
||
- worker 侧所有 --flag 必须在 cloud/worker.py 里有 add_argument 定义，
|
||
防止拼写漂移导致 worker 启动即报未知参数；
|
||
- 四项 flag 绝不能出现在 scaler 侧（`--` 之前），它们是 worker 参数。
|
||
"""
|
||
|
||
import os
|
||
import unittest
|
||
|
||
REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(
|
||
os.path.abspath(__file__))))
|
||
UNIT_PATH = os.path.join(REPO_ROOT, "deploy", "autoscaler",
|
||
"codex-autoscaler.service")
|
||
WORKER_PATH = os.path.join(REPO_ROOT, "cloud", "worker.py")
|
||
|
||
# 生产主机现行 unit 带的 worker 侧 MCP/技能播种四项。
|
||
MCP_SEED_FLAGS = ("--websearch-mcp", "--seed-skills",
|
||
"--imagegen-mcp", "--playwright-mcp")
|
||
|
||
|
||
def _execstart():
|
||
with open(UNIT_PATH, encoding="utf-8") as f:
|
||
lines = [ln.rstrip("\n") for ln in f]
|
||
starts = [ln for ln in lines if ln.startswith("ExecStart=")]
|
||
assert len(starts) == 1, "unit 应恰好有一行 ExecStart，实得 %d 行" % len(starts)
|
||
return starts[0][len("ExecStart="):]
|
||
|
||
|
||
def _split_sides():
|
||
cmd = _execstart()
|
||
assert cmd.startswith("/usr/bin/python3 -m cloud.autoscaler"), \
|
||
"unit 应以 cloud.autoscaler 启动，实得：%s" % cmd[:60]
|
||
parts = cmd.split(" -- ", 1)
|
||
assert len(parts) == 2, "ExecStart 缺少 scaler/worker 分隔符 ' -- '"
|
||
return parts[0], parts[1]
|
||
|
||
|
||
def _flag_tokens(side):
|
||
"""side 里以 -- 开头的 token（去重前顺序保留，含 =value 形式剥离）。"""
|
||
flags = []
|
||
for tok in side.split():
|
||
if tok.startswith("--"):
|
||
flags.append(tok.split("=", 1)[0])
|
||
return flags
|
||
|
||
|
||
class TestAutoscalerUnitSync(unittest.TestCase):
|
||
def test_worker_side_includes_all_four_mcp_seed_flags(self):
|
||
_, worker_side = _split_sides()
|
||
flags = _flag_tokens(worker_side)
|
||
for flag in MCP_SEED_FLAGS:
|
||
self.assertIn(flag, flags,
|
||
"worker 侧缺 %s，仓库 unit 已滞后于生产主机" % flag)
|
||
|
||
def test_worker_side_flags_have_no_duplicates(self):
|
||
_, worker_side = _split_sides()
|
||
flags = _flag_tokens(worker_side)
|
||
self.assertEqual(len(flags), len(set(flags)),
|
||
"worker 侧 flag 重复：%s" % (flags,))
|
||
|
||
def test_worker_side_flags_are_defined_in_worker_parser(self):
|
||
_, worker_side = _split_sides()
|
||
with open(WORKER_PATH, encoding="utf-8") as f:
|
||
worker_src = f.read()
|
||
for flag in _flag_tokens(worker_side):
|
||
self.assertIn('add_argument("%s"' % flag, worker_src,
|
||
"%s 未在 cloud/worker.py 定义，worker 会报未知参数"
|
||
% flag)
|
||
|
||
def test_mcp_seed_flags_not_on_scaler_side(self):
|
||
scaler_side, _ = _split_sides()
|
||
flags = _flag_tokens(scaler_side)
|
||
for flag in MCP_SEED_FLAGS:
|
||
self.assertNotIn(flag, flags,
|
||
"%s 不应出现在 scaler 侧（`--` 之前）" % flag)
|
||
|
||
def test_worker_side_keeps_docker_runtime_runsc(self):
|
||
_, worker_side = _split_sides()
|
||
tokens = worker_side.split()
|
||
self.assertIn("--docker-runtime", tokens)
|
||
self.assertEqual(tokens[tokens.index("--docker-runtime") + 1], "runsc")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|