tzt
ce0f6170d3
- 入库历史遗漏源码/测试：router_system 9 模块（agent/executors/inference/knowledge/ memory/planner/skills/trace）、tests 11 个测试文件、config/knowledge 领域知识 - 入库根目录方案文档（v2/v3/可行性×2）、references 文献（arxiv 14-18/cnki_open/ 参考文献清单）、research 论文素材（routerarena/paper/中文文献 PDF） - 前端构建产物刷新（新 hash）；webapp 误写文档删除 - gitignore 增补：deepseek-harness、research/_refs、.mimosa/.zcode、网关日志/pid、 临时调试脚本、tests/e2e/node_modules、AI代理功能开发/prefix - 基线确认：318 passed
64 lines
2.1 KiB
Python
64 lines
2.1 KiB
Python
"""黑板 / DAG（WorkingMemory / TaskGraph）单元测试。"""
|
||
from router_system.memory import TaskGraph, TaskNode, WorkingMemory
|
||
|
||
|
||
def _graph():
|
||
g = TaskGraph()
|
||
g.add_node(TaskNode(id="a", kind="analyze", domain="code", query="q"))
|
||
g.add_node(TaskNode(id="b", kind="design", domain="code", query="q", deps=["a"]))
|
||
g.add_node(TaskNode(id="c", kind="implement", domain="code", query="q", deps=["b"]))
|
||
return g
|
||
|
||
|
||
def test_topo_order_respects_deps():
|
||
g = _graph()
|
||
order = [n.id for n in g.topo_order()]
|
||
assert order.index("a") < order.index("b") < order.index("c")
|
||
|
||
|
||
def test_cycle_fallback_no_crash():
|
||
g = TaskGraph()
|
||
g.add_node(TaskNode(id="x", kind="solve", domain="m", query="q", deps=["y"]))
|
||
g.add_node(TaskNode(id="y", kind="solve", domain="m", query="q", deps=["x"]))
|
||
order = [n.id for n in g.topo_order()]
|
||
assert set(order) == {"x", "y"}
|
||
|
||
|
||
def test_failed_and_all_done():
|
||
g = _graph()
|
||
assert g.all_done() is False
|
||
for n in g.nodes():
|
||
n.status = "done"
|
||
assert g.all_done() is True
|
||
g.get("a").status = "failed"
|
||
assert [n.id for n in g.failed()] == ["a"]
|
||
|
||
|
||
def test_merge_order():
|
||
mem = WorkingMemory()
|
||
mem.write_section("s1", "第一部分")
|
||
mem.write_section("s2", "第二部分")
|
||
assert mem.merge(["s1", "s2"]) == "第一部分\n\n第二部分"
|
||
# 指定顺序可调换
|
||
assert mem.merge(["s2", "s1"]) == "第二部分\n\n第一部分"
|
||
# 无顺序 → 写入顺序
|
||
assert mem.merge() == "第一部分\n\n第二部分"
|
||
|
||
|
||
def test_overwrite_fact_records_trace():
|
||
mem = WorkingMemory()
|
||
mem.write_fact("k", "v1", rule_id="r1")
|
||
mem.write_fact("k", "v2", rule_id="r2")
|
||
assert mem.get_fact("k") == "v2"
|
||
assert any(t.startswith("overwrite:k@r2") for t in mem.trace)
|
||
|
||
|
||
def test_duplicate_node_id_rejected():
|
||
g = TaskGraph()
|
||
g.add_node(TaskNode(id="a", kind="solve", domain="m", query="q"))
|
||
try:
|
||
g.add_node(TaskNode(id="a", kind="solve", domain="m", query="q"))
|
||
assert False, "重复节点 id 应报错"
|
||
except ValueError:
|
||
pass
|