admin
4866ad8d37
- graph/test_graphctl.py：新增 49 个用例，覆盖建图 / 校验硬错与警告 / 就绪集 / 状态流转 / 非法操作 / 边界 / 写入范围隔离；实跑 Ran 49 tests OK（27.5s） - graph/graphctl.py：修测试暴露的问题（+50/-10） - graph/README.md：补「怎么跑测试」一节（+27 行） - SKILL.md：目录树与入口指针登记 graph/（新增 4 行文档，机制行为零改动） - 隔离性核对：collabd.py / goalctl.py / 两个看板 / 全部钩子 diff 为空； state.py 闸门读数绿、常驻心跳 7 秒、看板 20100 在听
530 lines
24 KiB
Python
530 lines
24 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""test_graphctl.py —— 「图模式」自动化测试（2026-10-18）。
|
||
|
||
## 覆盖八类
|
||
1 建图（--kpi / --nodes-json） 5 状态流转（pending→running→done、blocked 累计 attempts）
|
||
2 校验硬错（环 / 悬空依赖 / 状态非法） 6 非法操作（重复开工 / 非法 state / 不存在的节点）
|
||
3 校验警告（无域 / 无判据 / 域重叠） 7 边界（单节点 / 空节点 / gid 含中文 / 图不存在）
|
||
4 就绪集（依赖 / 域冲突 / 并发上限） 8 独立性（只写临时目录，不碰现有机制文件）
|
||
|
||
## 跑法
|
||
python graph/test_graphctl.py -v
|
||
|
||
## 落点
|
||
临时工作区 = `$GRAPH_TEST_TMP`（默认 `AIProject/ai1net-dsh-server/tmp`）/`graph-test-<pid>`/<用例名>，
|
||
整体删除；每个用例一个干净工作区，互不影响。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import atexit
|
||
import atexit
|
||
import json
|
||
import os
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
HERE = Path(__file__).resolve().parent
|
||
GRAPHCTL = HERE / "graphctl.py"
|
||
PY = sys.executable
|
||
|
||
_DEFAULT_TMP = "E:/ProgramData/AIProject/ai1net-dsh-server/tmp"
|
||
TMP_ROOT = Path(os.environ.get("GRAPH_TEST_TMP") or _DEFAULT_TMP) / ("graph-test-%d" % os.getpid())
|
||
|
||
_ENV = dict(os.environ)
|
||
_ENV["PYTHONIOENCODING"] = "utf-8"
|
||
|
||
|
||
# ── 基础设施 ────────────────────────────────────────────────────────────
|
||
def _run(*args, stdin: str | None = None):
|
||
"""跑一次 CLI，返回 (rc, stdout 文本, stderr 文本)。"""
|
||
p = subprocess.run([PY, str(GRAPHCTL), *[str(x) for x in args]],
|
||
input=(stdin.encode("utf-8") if stdin is not None else b""),
|
||
stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=_ENV)
|
||
return (p.returncode,
|
||
p.stdout.decode("utf-8", "replace"),
|
||
p.stderr.decode("utf-8", "replace"))
|
||
|
||
|
||
def ok(*args, stdin: str | None = None) -> dict:
|
||
"""要求 rc=0 并把 stdout 解析成 dict。"""
|
||
rc, out, err = _run(*args, stdin=stdin)
|
||
if rc != 0:
|
||
raise AssertionError("期望 rc=0，实际 rc=%d\nstderr=%s\nstdout=%s" % (rc, err, out))
|
||
return json.loads(out)
|
||
|
||
|
||
def bad(*args, stdin: str | None = None):
|
||
"""要求 rc!=0，返回 stderr（供断错误原因）。"""
|
||
rc, out, err = _run(*args, stdin=stdin)
|
||
if rc == 0:
|
||
raise AssertionError("期望失败，实际成功：stdout=%s" % out)
|
||
return err
|
||
|
||
|
||
def _drop(p: Path, tries: int = 5) -> None:
|
||
"""删临时目录：Windows 上刚写完的文件可能被杀软/索引器短暂占用，重试几次。"""
|
||
last = None
|
||
for _ in range(tries):
|
||
try:
|
||
shutil.rmtree(p)
|
||
return
|
||
except FileNotFoundError:
|
||
return
|
||
except Exception as e: # noqa: BLE001 —— 重试后仍失败才报，带上原因
|
||
last = e
|
||
time.sleep(0.1)
|
||
sys.stderr.write("[cleanup] 删不掉（已重试 %d 次）：%s ⇒ %r\n" % (tries, p, last))
|
||
|
||
|
||
class Base(unittest.TestCase):
|
||
def setUp(self):
|
||
self.ws = TMP_ROOT / self._testMethodName
|
||
shutil.rmtree(self.ws, ignore_errors=True)
|
||
self.ws.mkdir(parents=True, exist_ok=True)
|
||
self.gid = "g1"
|
||
|
||
def tearDown(self):
|
||
global _TD_COUNT
|
||
_TD_COUNT += 1
|
||
_drop(self.ws)
|
||
|
||
# -- 便捷动作 --
|
||
def gpath(self, gid=None) -> Path:
|
||
return self.ws / ".workbuddy" / "graph" / (gid or self.gid) / "graph.json"
|
||
|
||
def load(self, gid=None) -> dict:
|
||
return json.loads(self.gpath(gid).read_text(encoding="utf-8"))
|
||
|
||
def dump(self, g: dict, gid=None) -> None:
|
||
self.gpath(gid).write_text(json.dumps(g, ensure_ascii=False, indent=2),
|
||
encoding="utf-8")
|
||
|
||
def init(self, spec: dict | None = None, gid=None, title="测试目标"):
|
||
gid = gid or self.gid
|
||
args = ["init", "--ws", self.ws, "--title", title, "--gid", gid]
|
||
if spec is None:
|
||
args += ["--kpi", "判据一"]
|
||
return ok(*args)
|
||
return ok(*args, "--nodes-json", "-",
|
||
stdin=json.dumps(spec, ensure_ascii=False))
|
||
|
||
def state(self, node, status):
|
||
g = self.load()
|
||
g["nodes"][node]["status"] = status
|
||
self.dump(g)
|
||
|
||
def events(self, gid=None) -> list:
|
||
p = self.ws / ".workbuddy" / "graph" / (gid or self.gid) / "events.jsonl"
|
||
return [json.loads(x) for x in p.read_text(encoding="utf-8").splitlines() if x.strip()]
|
||
|
||
|
||
# ── 1 建图 ──────────────────────────────────────────────────────────────
|
||
class T1_Init(Base):
|
||
def test_kpi_one_node_per_criterion(self):
|
||
ok("init", "--ws", self.ws, "--title", "三件事", "--gid", "k1",
|
||
"--kpi", "判据A;判据B；判据C")
|
||
g = self.load("k1")
|
||
self.assertEqual(sorted(g["nodes"]), ["N1", "N2", "N3"])
|
||
self.assertEqual(g["nodes"]["N1"]["accepts"], ["判据A"])
|
||
self.assertEqual(g["nodes"]["N3"]["accepts"], ["判据C"]) # 全角分号也认
|
||
|
||
def test_kpi_default_fields(self):
|
||
self.init()
|
||
n = self.load()["nodes"]["N1"]
|
||
self.assertEqual(n["status"], "pending")
|
||
self.assertEqual(n["attempts"], 0)
|
||
self.assertEqual(n["deps"], [])
|
||
self.assertEqual(n["evidence"], "")
|
||
|
||
def test_nodes_json_deps_and_domain(self):
|
||
self.init({"nodes": [
|
||
{"id": "A", "title": "改界面", "domain": ["ui/"], "accepts": ["能打开"], "deps": []},
|
||
{"id": "B", "title": "汇总", "domain": ["docs"], "accepts": ["齐了"], "deps": ["A"]},
|
||
]})
|
||
g = self.load()
|
||
self.assertEqual(g["nodes"]["B"]["deps"], ["A"])
|
||
self.assertEqual(g["nodes"]["B"]["domain"], ["docs/"]) # 自动补尾斜杠
|
||
|
||
def test_nodes_json_max_parallel(self):
|
||
self.init({"max_parallel": 2, "nodes": [{"id": "A", "accepts": ["x"]}]})
|
||
self.assertEqual(self.load()["meta"]["max_parallel"], 2)
|
||
|
||
def test_init_needs_kpi_or_nodes_json(self):
|
||
err = bad("init", "--ws", self.ws, "--title", "空手")
|
||
self.assertIn("--kpi", err)
|
||
|
||
def test_duplicate_id_rejected(self):
|
||
"""显式 id 重复必须报错，不能静默覆盖掉前一个节点。"""
|
||
err = bad("init", "--ws", self.ws, "--title", "重名", "--gid", self.gid,
|
||
"--nodes-json", "-",
|
||
stdin=json.dumps({"nodes": [
|
||
{"id": "A", "accepts": ["一"]},
|
||
{"id": "A", "accepts": ["二"]}]}, ensure_ascii=False))
|
||
self.assertIn("重复", err)
|
||
|
||
def test_missing_id_does_not_collide(self):
|
||
"""省略 id 时自动编号要避开已有 id。"""
|
||
self.init({"nodes": [{"id": "N1", "accepts": ["一"]}, {"accepts": ["二"]}]})
|
||
self.assertEqual(len(self.load()["nodes"]), 2)
|
||
|
||
def test_init_writes_event(self):
|
||
self.init()
|
||
self.assertEqual(self.events()[0]["event"], "init")
|
||
|
||
|
||
# ── 2 校验硬错 ──────────────────────────────────────────────────────────
|
||
class T2_CheckErrors(Base):
|
||
def test_cycle(self):
|
||
self.init({"nodes": [{"id": "A", "deps": ["B"], "accepts": ["x"]},
|
||
{"id": "B", "deps": ["A"], "accepts": ["y"]}]})
|
||
rc, out, _ = _run("check", "--ws", self.ws, "--gid", self.gid)
|
||
r = json.loads(out)
|
||
self.assertEqual(rc, 1)
|
||
self.assertFalse(r["ok"])
|
||
self.assertTrue(any("环" in e for e in r["errors"]))
|
||
|
||
def test_self_cycle(self):
|
||
self.init({"nodes": [{"id": "A", "deps": ["A"], "accepts": ["x"]}]})
|
||
r = ok("check", "--ws", self.ws, "--gid", self.gid) if False else \
|
||
json.loads(_run("check", "--ws", self.ws, "--gid", self.gid)[1])
|
||
self.assertTrue(any("环" in e for e in r["errors"]))
|
||
|
||
def test_dangling_dep(self):
|
||
self.init({"nodes": [{"id": "A", "deps": ["Z"], "accepts": ["x"]}]})
|
||
r = json.loads(_run("check", "--ws", self.ws, "--gid", self.gid)[1])
|
||
self.assertFalse(r["ok"])
|
||
self.assertTrue(any("Z" in e and "不存在" in e for e in r["errors"]))
|
||
|
||
def test_illegal_status(self):
|
||
self.init()
|
||
self.state("N1", "weird")
|
||
r = json.loads(_run("check", "--ws", self.ws, "--gid", self.gid)[1])
|
||
self.assertFalse(r["ok"])
|
||
self.assertTrue(any("状态非法" in e for e in r["errors"]))
|
||
|
||
def test_clean_graph_ok(self):
|
||
self.init({"nodes": [{"id": "A", "domain": ["ui/"], "accepts": ["x"]},
|
||
{"id": "B", "domain": ["docs/"], "accepts": ["y"], "deps": ["A"]}]})
|
||
r = ok("check", "--ws", self.ws, "--gid", self.gid)
|
||
self.assertTrue(r["ok"])
|
||
self.assertEqual(r["errors"], [])
|
||
self.assertEqual(r["nodes"], 2)
|
||
|
||
|
||
# ── 3 校验警告 ──────────────────────────────────────────────────────────
|
||
class T3_CheckWarnings(Base):
|
||
def test_no_domain(self):
|
||
self.init({"nodes": [{"id": "A", "accepts": ["x"]}]})
|
||
r = ok("check", "--ws", self.ws, "--gid", self.gid)
|
||
self.assertTrue(r["ok"])
|
||
self.assertTrue(any("没有声明域" in w for w in r["warnings"]))
|
||
|
||
def test_no_accepts(self):
|
||
self.init({"nodes": [{"id": "A", "domain": ["ui/"]}]})
|
||
r = ok("check", "--ws", self.ws, "--gid", self.gid)
|
||
self.assertTrue(any("没有验收判据" in w for w in r["warnings"]))
|
||
|
||
def test_overlap_nested(self):
|
||
self.init({"nodes": [{"id": "A", "domain": ["docs/"], "accepts": ["x"]},
|
||
{"id": "B", "domain": ["docs/ui/"], "accepts": ["y"]}]})
|
||
r = ok("check", "--ws", self.ws, "--gid", self.gid)
|
||
self.assertTrue(any("重叠" in w for w in r["warnings"]))
|
||
|
||
def test_overlap_reverse_nested(self):
|
||
self.init({"nodes": [{"id": "A", "domain": ["docs/ui/"], "accepts": ["x"]},
|
||
{"id": "B", "domain": ["docs/"], "accepts": ["y"]}]})
|
||
r = ok("check", "--ws", self.ws, "--gid", self.gid)
|
||
self.assertTrue(any("重叠" in w for w in r["warnings"]))
|
||
|
||
def test_disjoint_no_warning(self):
|
||
self.init({"nodes": [{"id": "A", "domain": ["ui/"], "accepts": ["x"]},
|
||
{"id": "B", "domain": ["docs-ui/"], "accepts": ["y"]}]})
|
||
r = ok("check", "--ws", self.ws, "--gid", self.gid)
|
||
self.assertFalse(any("重叠" in w for w in r["warnings"]))
|
||
|
||
|
||
# ── 4 就绪集 ────────────────────────────────────────────────────────────
|
||
class T4_Ready(Base):
|
||
def test_all_pending_no_deps(self):
|
||
self.init({"nodes": [{"id": "A", "domain": ["ui/"], "accepts": ["x"]},
|
||
{"id": "B", "domain": ["docs/"], "accepts": ["y"]}]})
|
||
r = ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||
self.assertEqual(sorted(r["ready"]), ["A", "B"])
|
||
self.assertEqual(r["backlog"], [])
|
||
self.assertEqual(r["running"], [])
|
||
|
||
def test_dep_unmet(self):
|
||
self.init({"nodes": [{"id": "A", "domain": ["ui/"], "accepts": ["x"]},
|
||
{"id": "B", "domain": ["docs/"], "accepts": ["y"], "deps": ["A"]}]})
|
||
r = ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||
self.assertEqual(r["ready"], ["A"])
|
||
self.assertTrue(any(w["node"] == "B" and "依赖未完成" in w["why"]
|
||
for w in r["waiting"]))
|
||
|
||
def test_dep_done_releases(self):
|
||
self.init({"nodes": [{"id": "A", "domain": ["ui/"], "accepts": ["x"]},
|
||
{"id": "B", "domain": ["docs/"], "accepts": ["y"], "deps": ["A"]}]})
|
||
ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "A")
|
||
ok("report", "--ws", self.ws, "--gid", self.gid,
|
||
"--node", "A", "--state", "done", "--evidence", "截图在 ui/shot.png")
|
||
r = ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||
self.assertEqual(r["ready"], ["B"])
|
||
|
||
def test_domain_clash_with_running(self):
|
||
self.init({"nodes": [{"id": "A", "domain": ["docs/"], "accepts": ["x"]},
|
||
{"id": "B", "domain": ["docs/ui/"], "accepts": ["y"]}]})
|
||
ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "A")
|
||
r = ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||
self.assertNotIn("B", r["ready"])
|
||
self.assertTrue(any(w["node"] == "B" and "域与在跑的冲突" in w["why"]
|
||
for w in r["waiting"]))
|
||
|
||
def test_max_parallel_overflow(self):
|
||
self.init({"max_parallel": 2, "nodes": [
|
||
{"id": x, "domain": ["%s/" % x.lower()], "accepts": ["ok"]}
|
||
for x in ("A", "B", "C", "D")]})
|
||
r = ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||
self.assertEqual(len(r["ready"]), 2)
|
||
self.assertEqual(len(r["backlog"]), 2)
|
||
self.assertEqual(r["cap"], 2)
|
||
|
||
def test_max_parallel_counts_running(self):
|
||
self.init({"max_parallel": 2, "nodes": [
|
||
{"id": x, "domain": ["%s/" % x.lower()], "accepts": ["ok"]}
|
||
for x in ("A", "B", "C")]})
|
||
ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "A")
|
||
r = ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||
self.assertEqual(r["slots"], 1)
|
||
self.assertEqual(len(r["ready"]), 1)
|
||
self.assertEqual(r["running"], ["A"])
|
||
|
||
def test_overlapping_peers_not_in_same_batch(self):
|
||
"""两个域重叠的节点⛔ 不能出现在同一个 ready 批次里（check 说「永不同批」，ready 要兑现）。"""
|
||
self.init({"nodes": [{"id": "A", "domain": ["docs/"], "accepts": ["x"]},
|
||
{"id": "B", "domain": ["docs/ui/"], "accepts": ["y"]}]})
|
||
r = ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||
self.assertEqual(len(r["ready"]), 1)
|
||
left = [w for w in r["waiting"] if "重叠" in w["why"]]
|
||
self.assertEqual(len(left), 1, r)
|
||
|
||
def test_domainless_parallelises(self):
|
||
"""没有域的节点按现状不参与域比较 ⇒ 可与别的节点同批（check 的警告文案须与此一致）。"""
|
||
self.init({"nodes": [{"id": "A", "accepts": ["x"]},
|
||
{"id": "B", "domain": ["docs/"], "accepts": ["y"]}]})
|
||
r = ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||
self.assertEqual(sorted(r["ready"]), ["A", "B"])
|
||
|
||
|
||
# ── 5 状态流转 ──────────────────────────────────────────────────────────
|
||
class T5_Flow(Base):
|
||
def test_pending_running_done(self):
|
||
self.init()
|
||
r = ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "N1", "--by", "执行会话-x")
|
||
self.assertEqual(r["status"], "running")
|
||
self.assertEqual(self.load()["nodes"]["N1"]["status"], "running")
|
||
r = ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "N1",
|
||
"--state", "done", "--evidence", "产物在位")
|
||
self.assertEqual(r["status"], "done")
|
||
g = self.load()
|
||
self.assertEqual(g["nodes"]["N1"]["status"], "done")
|
||
self.assertEqual(g["nodes"]["N1"]["evidence"], "产物在位")
|
||
|
||
def test_events_append_only(self):
|
||
self.init()
|
||
ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "N1")
|
||
ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "N1", "--state", "done")
|
||
ev = self.events()
|
||
self.assertEqual(ev[0]["event"], "init")
|
||
self.assertEqual([e.get("to") for e in ev[1:]], ["running", "done"])
|
||
|
||
def test_blocked_bumps_attempts(self):
|
||
self.init()
|
||
ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "N1", "--state", "blocked")
|
||
n = self.load()["nodes"]["N1"]
|
||
self.assertEqual(n["status"], "blocked")
|
||
self.assertEqual(n["attempts"], 1)
|
||
|
||
def test_blocked_twice(self):
|
||
self.init()
|
||
ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "N1", "--state", "blocked")
|
||
ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "N1")
|
||
ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "N1", "--state", "blocked")
|
||
self.assertEqual(self.load()["nodes"]["N1"]["attempts"], 2)
|
||
|
||
def test_done_does_not_touch_attempts(self):
|
||
self.init()
|
||
ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "N1")
|
||
ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "N1", "--state", "done")
|
||
self.assertEqual(self.load()["nodes"]["N1"]["attempts"], 0)
|
||
|
||
def test_report_returns_next_ready(self):
|
||
self.init({"nodes": [{"id": "A", "domain": ["ui/"], "accepts": ["x"]},
|
||
{"id": "B", "domain": ["docs/"], "accepts": ["y"], "deps": ["A"]}]})
|
||
ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "A")
|
||
r = ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", "--state", "done")
|
||
self.assertEqual(r["next_ready"], ["B"])
|
||
|
||
|
||
# ── 6 非法操作 ──────────────────────────────────────────────────────────
|
||
class T6_Illegal(Base):
|
||
def test_start_running_rejected(self):
|
||
self.init()
|
||
ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "N1")
|
||
err = bad("start", "--ws", self.ws, "--gid", self.gid, "--node", "N1")
|
||
self.assertIn("不能", err)
|
||
|
||
def test_start_done_rejected(self):
|
||
self.init()
|
||
ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "N1")
|
||
ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "N1", "--state", "done")
|
||
bad("start", "--ws", self.ws, "--gid", self.gid, "--node", "N1")
|
||
|
||
def test_start_blocked_allowed_for_retry(self):
|
||
"""blocked 要能重开（否则 attempts 白记，且只能手改 JSON 才能重试）。"""
|
||
self.init()
|
||
ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "N1", "--state", "blocked")
|
||
r = ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "N1")
|
||
self.assertEqual(r["status"], "running")
|
||
self.assertEqual(self.load()["nodes"]["N1"]["attempts"], 1)
|
||
|
||
def test_state_only_done_or_blocked(self):
|
||
self.init()
|
||
err = bad("report", "--ws", self.ws, "--gid", self.gid, "--node", "N1", "--state", "pending")
|
||
self.assertIn("done", err)
|
||
|
||
def test_unknown_node_start(self):
|
||
self.init()
|
||
err = bad("start", "--ws", self.ws, "--gid", self.gid, "--node", "ZZ")
|
||
self.assertIn("没有这个节点", err)
|
||
|
||
def test_unknown_node_report(self):
|
||
self.init()
|
||
err = bad("report", "--ws", self.ws, "--gid", self.gid, "--node", "ZZ", "--state", "done")
|
||
self.assertIn("没有这个节点", err)
|
||
|
||
def test_unknown_gid(self):
|
||
err = bad("show", "--ws", self.ws, "--gid", "不存在")
|
||
self.assertIn("找不到图", err)
|
||
|
||
|
||
# ── 7 边界 ──────────────────────────────────────────────────────────────
|
||
class T7_Edge(Base):
|
||
def test_single_node(self):
|
||
self.init()
|
||
r = ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||
self.assertEqual(r["ready"], ["N1"])
|
||
ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "N1")
|
||
ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "N1", "--state", "done")
|
||
r = ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||
self.assertEqual(r["ready"], [])
|
||
self.assertEqual(r["waiting"], [])
|
||
|
||
def test_empty_graph(self):
|
||
self.init({"nodes": []})
|
||
self.assertEqual(self.load()["nodes"], {})
|
||
self.assertEqual(ok("check", "--ws", self.ws, "--gid", self.gid)["nodes"], 0)
|
||
r = ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||
self.assertEqual((r["ready"], r["backlog"], r["waiting"]), ([], [], []))
|
||
self.assertEqual(len(ok("list", "--ws", self.ws)["graphs"]), 1)
|
||
|
||
def test_chinese_gid_explicit(self):
|
||
ok("init", "--ws", self.ws, "--title", "中文", "--gid", "20261018-2235-界面改造",
|
||
"--kpi", "看上去对")
|
||
g = ok("show", "--ws", self.ws, "--gid", "20261018-2235-界面改造")
|
||
self.assertEqual(g["gid"], "20261018-2235-界面改造")
|
||
self.assertEqual(len(ok("list", "--ws", self.ws)["graphs"]), 1)
|
||
|
||
def test_chinese_title_auto_gid(self):
|
||
r = ok("init", "--ws", self.ws, "--title", "给三块功能各加一份文档", "--kpi", "一;二")
|
||
self.assertIn("给三块功能", r["gid"])
|
||
self.assertEqual(r["nodes"], 2)
|
||
g = ok("show", "--ws", self.ws, "--gid", r["gid"])
|
||
self.assertEqual(len(g["nodes"]), 2)
|
||
|
||
def test_list_counts_by_status(self):
|
||
self.init({"nodes": [{"id": "A", "domain": ["a/"], "accepts": ["x"]},
|
||
{"id": "B", "domain": ["b/"], "accepts": ["y"]}]})
|
||
ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "A")
|
||
row = ok("list", "--ws", self.ws)["graphs"][0]
|
||
self.assertEqual(row["nodes"], 2)
|
||
self.assertEqual(row["by_status"], {"running": 1, "pending": 1})
|
||
|
||
def test_list_empty_ws(self):
|
||
self.assertEqual(ok("list", "--ws", self.ws)["graphs"], [])
|
||
|
||
|
||
# ── 8 独立性 ────────────────────────────────────────────────────────────
|
||
class T8_Isolation(Base):
|
||
def _full_lifecycle(self):
|
||
self.init({"nodes": [{"id": "A", "domain": ["ui/"], "accepts": ["x"]},
|
||
{"id": "B", "domain": ["docs/"], "accepts": ["y"], "deps": ["A"]}]})
|
||
ok("check", "--ws", self.ws, "--gid", self.gid)
|
||
ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||
ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "A")
|
||
ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", "--state", "done")
|
||
ok("show", "--ws", self.ws, "--gid", self.gid)
|
||
ok("list", "--ws", self.ws)
|
||
|
||
def test_sentinel_mechanism_files_untouched(self):
|
||
"""工作区里已有的 goal.json / tasks.json / collabd-state.json 一个字节都不许变。"""
|
||
sentinels = {
|
||
"goal.json": '{"goal": "sentinel", "version": 7}\n',
|
||
"tasks.json": '{"tasks": [{"id": "t1"}]}\n',
|
||
".workbuddy/collab/collabd-state.json": '{"mode": "sentinel"}\n',
|
||
}
|
||
before = {}
|
||
for rel, body in sentinels.items():
|
||
p = self.ws / rel
|
||
p.parent.mkdir(parents=True, exist_ok=True)
|
||
p.write_text(body, encoding="utf-8")
|
||
before[rel] = p.read_bytes()
|
||
|
||
self._full_lifecycle()
|
||
|
||
for rel in sentinels:
|
||
self.assertEqual((self.ws / rel).read_bytes(), before[rel], "%s 被改动了" % rel)
|
||
|
||
def test_no_mechanism_files_created(self):
|
||
self._full_lifecycle()
|
||
for name in ("goal.json", "tasks.json", "collabd-state.json"):
|
||
hits = [p for p in self.ws.rglob(name)]
|
||
self.assertEqual(hits, [], "不该出现的文件：%s" % hits)
|
||
|
||
def test_writes_only_under_mark_dir(self):
|
||
"""图模式的写入范围只有 <ws>/.workbuddy/graph/。"""
|
||
self._full_lifecycle()
|
||
top = sorted(p.name for p in self.ws.iterdir())
|
||
self.assertEqual(top, [".workbuddy"])
|
||
self.assertEqual(sorted(p.name for p in (self.ws / ".workbuddy").iterdir()), ["graph"])
|
||
self.assertEqual(sorted(p.name for p in (self.ws / ".workbuddy" / "graph").iterdir()), [self.gid])
|
||
|
||
def test_temp_root_is_under_project_tmp(self):
|
||
"""临时工作区落在项目 tmp 下（跑完由 tearDownModule 删）。"""
|
||
self.assertIn("tmp", [p.lower() for p in TMP_ROOT.parts])
|
||
self.assertTrue(str(TMP_ROOT).startswith(str(Path(_DEFAULT_TMP).parent.parent)) or
|
||
os.environ.get("GRAPH_TEST_TMP"), TMP_ROOT)
|
||
|
||
|
||
def tearDownModule():
|
||
_drop(TMP_ROOT)
|
||
|
||
|
||
_TD_COUNT = 0
|
||
|
||
|
||
def _final_cleanup():
|
||
"""兜底：不管 unittest 的模块夹具跑没跑，进程退出前把临时根删干净。"""
|
||
_drop(TMP_ROOT)
|
||
|
||
|
||
atexit.register(_final_cleanup)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main(verbosity=2)
|