148 lines
7.8 KiB
Python
148 lines
7.8 KiB
Python
"""cheap_service_app 单测:import 惰性红线 + 会话注册表读兜底/映射解析/I1 fail-closed + 端口池派生 + 工厂结构/soft/三闸=150 + collector REPLY_END flush。
|
|
真起 Service(Redis)、真跑门(chrome)排 mini-desktop 窗口;本测只验装配零件与惰性,零网络/LLM/chrome。
|
|
跑:cheap-worker/.venv/bin/python -m pytest cheap-worker/tests/test_cheap_service_app.py -v
|
|
"""
|
|
import asyncio
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1])) # → cheap-worker/
|
|
import _bootstrap # noqa: E402,F401
|
|
import cheap_run # noqa: E402
|
|
import cheap_service_app as A # noqa: E402
|
|
|
|
|
|
def test_read_session_cfg_missing_returns_empty(tmp_path, monkeypatch):
|
|
# 注册表不存在 → {}(工厂据此回落 game_id=session_id;正常路 driver 恒在 /chat 前写、不会缺)。
|
|
monkeypatch.setattr(cheap_run, "session_cfg_path", lambda sid: tmp_path / "_cheap-sessions" / f"{sid}.json")
|
|
assert A._read_session_cfg("s1") == {}
|
|
|
|
|
|
def test_read_session_cfg_reads_mapping_and_whitelist(tmp_path, monkeypatch):
|
|
# 读会话注册表:external_game_id(后端 gameId)+ write_whitelist + scaffold_template + restricted。
|
|
import json
|
|
monkeypatch.setattr(cheap_run, "session_cfg_path", lambda sid: tmp_path / "_cheap-sessions" / f"{sid}.json")
|
|
p = tmp_path / "_cheap-sessions" / "s2.json"
|
|
p.parent.mkdir(parents=True)
|
|
p.write_text(json.dumps({"external_game_id": "70012", "write_whitelist": ["game-logic.js"],
|
|
"scaffold_template": "_template-shop", "restricted": True}), encoding="utf-8")
|
|
cfg = A._read_session_cfg("s2")
|
|
assert cfg["external_game_id"] == "70012"
|
|
assert cfg["write_whitelist"] == ["game-logic.js"]
|
|
assert cfg["scaffold_template"] == "_template-shop"
|
|
assert cfg["restricted"] is True
|
|
|
|
|
|
def test_resolve_external_game_id_falls_back_to_session(tmp_path, monkeypatch):
|
|
# sidecar 缺 external_game_id → 工厂回落 session_id(响亮失败,不静默用错目录)。
|
|
monkeypatch.setattr(cheap_run, "session_cfg_path", lambda sid: tmp_path / "_cheap-sessions" / f"{sid}.json")
|
|
assert A._resolve_external_game_id("sX", {}) == "sX"
|
|
assert A._resolve_external_game_id("sX", {"external_game_id": "70099"}) == "70099"
|
|
|
|
|
|
def test_port_pool_derives_distinct_pairs():
|
|
# 端口池派生不同端口对、释放后可再取(并发避撞的机制核)。
|
|
p1 = A.acquire_ports()
|
|
p2 = A.acquire_ports()
|
|
assert p1 != p2
|
|
assert p1[0] >= A._CHEAP_PORT_BASE and p1[1] >= A._CHEAP_CDP_BASE
|
|
A.release_ports(p1)
|
|
A.release_ports(p2)
|
|
p3 = A.acquire_ports() # 释放后能再取到(不泄漏)
|
|
assert p3 in (p1, p2)
|
|
A.release_ports(p3)
|
|
|
|
|
|
def test_tools_factory_returns_six_tools_bound_to_external_game_id(tmp_path, monkeypatch):
|
|
import json
|
|
monkeypatch.setattr(cheap_run, "session_cfg_path", lambda sid: tmp_path / "_cheap-sessions" / f"{sid}.json")
|
|
p = tmp_path / "_cheap-sessions" / "s1.json"
|
|
p.parent.mkdir(parents=True)
|
|
p.write_text(json.dumps({"external_game_id": "70012"}), encoding="utf-8") # create 路:仅映射
|
|
tools = asyncio.run(A._cheap_tools_factory("u1", "a1", "s1"))
|
|
assert len(tools) == 6, "便宜档六工具(read/list/write/check/build/finish)"
|
|
|
|
|
|
def test_resolve_write_whitelist_i1_fail_closed():
|
|
# I1 纯函数:create 路(非 restricted、无白名单)→ None(不收窄,全 L3 自由写);
|
|
# reskin 路(restricted:true + 白名单)→ 收窄成 set;restricted 但白名单缺失 → 空集(fail-closed 禁写、不回落 None)。
|
|
assert A._resolve_write_whitelist({}) is None
|
|
assert A._resolve_write_whitelist({"restricted": False}) is None
|
|
assert A._resolve_write_whitelist({"restricted": True, "write_whitelist": ["game-logic.js"]}) == {"game-logic.js"}
|
|
assert A._resolve_write_whitelist({"restricted": True}) == set(), "restricted 但缺白名单 → 空集(禁写)"
|
|
assert A._resolve_write_whitelist({"restricted": True, "write_whitelist": []}) == set()
|
|
|
|
|
|
def test_middlewares_factory_soft_budget_and_failsafe_caps(tmp_path, monkeypatch):
|
|
import json
|
|
monkeypatch.setattr(cheap_run, "session_cfg_path", lambda sid: tmp_path / "_cheap-sessions" / f"{sid}.json")
|
|
monkeypatch.setattr(cheap_run, "game_dir", lambda gid: tmp_path / f"amgen-{gid}")
|
|
p = tmp_path / "_cheap-sessions" / "s2.json"
|
|
p.parent.mkdir(parents=True)
|
|
p.write_text(json.dumps({"external_game_id": "70020"}), encoding="utf-8")
|
|
mws = asyncio.run(A._cheap_middlewares_factory("u1", "a1", "s2"))
|
|
names = [type(m).__name__ for m in mws]
|
|
assert names == ["_CheapRunCollector", "Tier2TraceMiddleware", "RepairMiddleware", "CircuitBreakerMiddleware"]
|
|
breaker = mws[3]
|
|
assert breaker.soft_budget is True, "决策①:cheap 软预算切 soft"
|
|
assert breaker.rmb_hard_limit == 10.0, "cheap ¥10(soft 档软目标,不再是硬地板)"
|
|
# 决策① + Codex C4:成本上界靠 max_repairs=6;三道次数/轮数闸抬到 150 当纯失控兜底(非成本尺)。
|
|
assert breaker.max_model_calls == 150, "失控兜底 150(非成本尺)"
|
|
assert breaker.max_tool_calls == 150, "C4:必须显式设 150,否则默认 60 会先撞"
|
|
assert breaker.step_timeout_s >= 390, "C3:单步静默阈值须 > 单次门跑最坏(~390s)"
|
|
assert breaker.enable_rmb_gate is True
|
|
collector = mws[0]
|
|
assert collector._game_id == "70020", "C2:collector 绑后端 gameId(经 sidecar 解析),非 session_id"
|
|
repair = mws[2]
|
|
assert repair._max_repairs == 6 # genconfig iteration.max_resumes(成本上界=优雅终止)
|
|
# budget_exhausted 读实测已花 ¥ 超上限(非预估):未超 False、超 True。
|
|
breaker._rmb_gate_active = True
|
|
breaker.spent_rmb = 0.0
|
|
assert repair._budget_exhausted() is False
|
|
breaker.spent_rmb = 10.5
|
|
assert repair._budget_exhausted() is True
|
|
|
|
|
|
def test_collector_flushes_on_reply_end(tmp_path, monkeypatch):
|
|
# C1:collector 在 REPLY_END 流经时**同步 flush**(不等 finally),关闭「driver 读 REPLY_END 时 sidecar 还空」竞态。
|
|
import json
|
|
monkeypatch.setattr(cheap_run, "game_dir", lambda gid: tmp_path / f"amgen-{gid}")
|
|
|
|
class _FakeBreaker:
|
|
spent_rmb, _rmb_gate_active, budget_soft_tripped = 0.42, True, False
|
|
|
|
class _FakeRepair:
|
|
repairs = 2
|
|
|
|
class _FakeTracer:
|
|
def summary(self):
|
|
return {"steps": 3}
|
|
|
|
class ReplyEndEvent: # 类名即判据:collector 认 type(evt).__name__ == "ReplyEndEvent"(镜像框架事件类名)
|
|
pass
|
|
|
|
Collector = A._get_collector_cls()
|
|
col = Collector(game_id="70021", breaker=_FakeBreaker(), repair=_FakeRepair(), tracer=_FakeTracer())
|
|
|
|
async def _fake_next(**_):
|
|
yield ReplyEndEvent() # REPLY_END 先于 finish Msg 流经(_agent.py:615)
|
|
|
|
async def _drive():
|
|
seen_after = None
|
|
async for evt in col.on_reply(None, {}, _fake_next):
|
|
# yield REPLY_END 之前 flush 必已发生(同步、无 await)→ 此刻 sidecar 已在盘上
|
|
seen_after = (tmp_path / "amgen-70021" / "evidence" / "service-run-summary.json").exists()
|
|
return seen_after
|
|
|
|
assert asyncio.run(_drive()) is True, "REPLY_END 流经时 sidecar 应已落盘(同步 flush)"
|
|
summary = json.loads((tmp_path / "amgen-70021" / "evidence" / "service-run-summary.json").read_text(encoding="utf-8"))
|
|
assert summary["costRmb"] == 0.42 and summary["repairs"] == 2
|
|
|
|
|
|
def test_reply_end_event_class_name_canary():
|
|
# collector 用 type(evt).__name__ == "ReplyEndEvent" 字符串判(cheap_service_app.py on_reply)锁 REPLY_END 落盘时机;
|
|
# 此 canary 钉住 AgentScope 事件类真名:2.0.x 若改名 ReplyEndEvent,字符串判会静默退化成 finally-only flush
|
|
# (成功局 costRmb=0),本断言让改名即红、把静默退化变永久绊线。生产码保留字符串判(不引硬 import 依赖、保 6c6g 惰性)。
|
|
from agentscope.event import ReplyEndEvent # 已实测可 import
|
|
assert ReplyEndEvent.__name__ == "ReplyEndEvent"
|