"""T4 交流文本 Workspace 单测（封闭，纯逻辑）。""" import json import pytest from router_system.workspace import ( LIMITS, STATUS_FLOW, Workspace, build_anchor, estimate_tokens, parse_anchor, validate, ) ALLOWED_BRIEF = { "goal": "用 Python 实现快速排序并解释复杂度", "constraints": ["必须用标准库", "时间复杂度 O(n log n)"], "tags": ["code"], "acceptance": [{"id": "a1", "check": "排序结果正确", "machine_checkable": True}], "plan": [ {"id": "s1", "task": "实现快排主函数", "deps": [], "done_criteria": "可运行"}, {"id": "s2", "task": "补充单元测试", "deps": ["s1"], "done_criteria": "3/3 通过"}, ], } def _ws(**kw): return Workspace.new(request_id=kw.get("request_id", "abc123def456"), query=kw.get("query", "写个快排"), api_token_cap=kw.get("api_token_cap", 8000), rounds_cap=kw.get("rounds_cap", 6)) # ---------- 结构/校验 ---------- def test_new_has_valid_defaults(): ws = _ws() assert ws.status == "draft" assert validate(ws.data) == [] b = ws.budget() assert b["api_token_cap"] == 8000 and b["rounds_cap"] == 6 def test_brief_write_once(): ws = _ws() ws.apply_brief(ALLOWED_BRIEF) assert ws.status == "in_progress" with pytest.raises(ValueError): ws.apply_brief(ALLOWED_BRIEF) # 二次写入被拒 def test_apply_brief_missing_field_rejected(): ws = _ws() bad = dict(ALLOWED_BRIEF) del bad["plan"] with pytest.raises(ValueError): ws.apply_brief(bad) def test_apply_brief_invalid_tag_rejected(): ws = _ws() bad = dict(ALLOWED_BRIEF, tags=["hacker"]) with pytest.raises(ValueError): ws.apply_brief(bad) def test_apply_brief_overlong_goal_rejected(): ws = _ws() bad = dict(ALLOWED_BRIEF, goal="长" * (LIMITS["goal"] + 1)) with pytest.raises(ValueError): ws.apply_brief(bad) def test_apply_brief_too_many_plan_steps_rejected(): ws = _ws() plan = [{"id": f"s{i}", "task": "t", "deps": [], "done_criteria": "c"} for i in range(LIMITS["plan_steps"] + 1)] bad = dict(ALLOWED_BRIEF, plan=plan) with pytest.raises(ValueError): ws.apply_brief(bad) # ---------- 状态机 ---------- def test_status_transitions(): ws = _ws() ws.apply_brief(ALLOWED_BRIEF) ws.transition("reviewing") ws.transition("done") assert ws.status == "done" with pytest.raises(ValueError): ws.transition("in_progress") # done 之后不允许再回 def test_illegal_transition(): ws = _ws() with pytest.raises(ValueError): ws.transition("done") # draft 不能直接到 done # ---------- 写入 + 预算 ---------- def test_progress_and_issues_and_decisions(): ws = _ws() ws.apply_brief(ALLOWED_BRIEF) ws.add_progress("s1", "done", "实现完成", artifact="a://s1_main.py") iid = ws.add_issue("s2", "a://s2_tests.py#L12-18", "测试失败", "3/3 通过", "修了两次", "请裁决") assert iid == "i1" ws.add_decision(iid, "改用排序后断言", [{"id": "s2", "task": "修测试"}]) assert len(ws["progress"]) == 1 assert len(ws["issues"]) == 1 assert len(ws["decisions"]) == 1 def test_budget_accumulation_and_exhaustion(): ws = _ws(api_token_cap=10) ws.apply_brief(ALLOWED_BRIEF) ws.add_budget(input_tokens=6, output_tokens=2) assert ws.exhausted() is False ws.add_budget(input_tokens=3) # 累计 11 > 10 assert ws.exhausted() is True def test_round_exhaustion(): ws = _ws(rounds_cap=2) ws.apply_brief(ALLOWED_BRIEF) ws.mark_round() ws.mark_round() assert ws.exhausted() is True def test_issue_text_overlong_rejected(): ws = _ws() ws.apply_brief(ALLOWED_BRIEF) with pytest.raises(ValueError): ws.add_issue("s1", "a://f.py", "观" * (LIMITS["issue_text"] + 1), "e", "t", "q") # ---------- rollup ---------- def test_rollup_folds_done_progress_and_resolved_issues(): ws = _ws() ws.apply_brief(ALLOWED_BRIEF) ws.add_progress("s1", "done", "快排实现完成", "a://s1_main.py") ws.add_progress("s2", "failed", "测试未过") iid = ws.add_issue("s2", "a://s2_tests.py#L1", "失败", "通过", "试过", "问") ws.add_decision(iid, "已修复") n = ws.rollup() assert n == 1 # 折叠 1 条 done progress assert "s1" in "".join(ws["archive"]) # s2 failed 保留；已 resolve 的 issue 从 issues 移除 steps = [p["step"] for p in ws["progress"]] assert steps == ["s2"] assert len(ws["issues"]) == 0 def test_rollup_keeps_unresolved_issue(): ws = _ws() ws.apply_brief(ALLOWED_BRIEF) ws.add_progress("s1", "done", "ok", "a://s1.py") ws.add_issue("s2", "a://s2.py#L1", "obs", "exp", "try", "ask") # 无 decision ws.rollup() assert len(ws["issues"]) == 1 # ---------- 渲染 ---------- def test_render_for_architect_within_budget(): ws = _ws() ws.apply_brief(ALLOWED_BRIEF) for i in range(6): ws.add_progress(f"s{i}", "done", f"步骤{i}完成" * 3) ws.add_issue(f"s{i}", f"a://f{i}.py#L1", "obs", "exp", "try", "ask" * 10) ws.add_decision(f"i{i+1}", "裁决" * 20) out = ws.render_for_architect() assert estimate_tokens(out) <= 1200 assert "== meta ==" in out def test_render_for_worker_has_step_and_criteria(): ws = _ws() ws.apply_brief(ALLOWED_BRIEF) out = ws.render_for_worker("s2", artifact_text="def test(): pass") assert "当前步" in out assert "s2" in out assert "def test(): pass" in out assert "验收标准" in out and "a1" in out def test_render_for_worker_marks_current_step(): ws = _ws() ws.apply_brief(ALLOWED_BRIEF) out = ws.render_for_worker("s1") assert "<-- 当前步" in out # ---------- 锚点 ---------- def test_anchor_build_and_parse(): a = build_anchor("s1_main.py", 12, 18) assert a == "a://s1_main.py#L12-18" p = parse_anchor(a) assert p == {"file": "s1_main.py", "start": 12, "end": 18} # 无行号 assert parse_anchor("a://f.py") == {"file": "f.py", "start": 1, "end": 1} assert parse_anchor("not-an-anchor") is None # ---------- 持久化 ---------- def test_save_load_roundtrip(tmp_path): ws = _ws() ws.apply_brief(ALLOWED_BRIEF) ws.add_progress("s1", "done", "ok") path = tmp_path / "ws.json" ws.save(path) loaded = Workspace.load(path) assert loaded.to_dict() == ws.to_dict() assert loaded.request_id == ws.request_id def test_json_serializable(tmp_path): ws = _ws() ws.apply_brief(ALLOWED_BRIEF) ws.add_issue("s1", "a://f.py#L1", "obs", "exp", "try", "ask") # 应能被 json 直接序列化（中文 ensure_ascii=False） s = json.dumps(ws.to_dict(), ensure_ascii=False) assert json.loads(s)["request_id"] == ws.request_id def test_validate_rejects_non_object(): assert validate("nope") != [] def test_status_flow_enumeration(): assert set(STATUS_FLOW.keys()) == {"draft", "in_progress", "reviewing", "escalated", "done", "failed"}