feat(session-mechanism/graph): B 方案——租约超时、失败策略与触发规则、动态展开
- 租约与超时：节点 start 记租约时刻，ready 时把租约超时的 running 节点识别出来并按策略处置 （默认回 pending ＋ attempts+1），卡死的节点不再永远占着并发名额 - 失败策略与下游触发规则：节点级重试（次数／间隔／退避／超限转 blocked）； 依赖触发支持 all_success（默认）／all_done／one_success 三档 - 动态展开（map）：支持建图时给列表、也支持运行时从上游产物读 N 项；展开出的子节点同样受 max_parallel 约束，子节点全部完成父节点才算完成 - 测试 52 → 92 条，实跑 Ran 92 tests OK（68 s），现有用例零回归 - README 补三节：字段怎么写、默认值是什么 - 改动只在 graph/ 目录内；collabd.py／goalctl.py／看板／钩子 diff 为空
This commit is contained in:
admin committed
1 parent
42a130e4f2
commit
3b5bcc535a
3 files changed
+1265
-82
No files matched your search
@@ -2,11 +2,13 @@
|
||||
# -*- 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 独立性（只写临时目录，不碰现有机制文件）
|
||||
## 覆盖十二类
|
||||
1 建图（--kpi / --nodes-json） 7 边界（单节点 / 空节点 / gid 含中文 / 图不存在）
|
||||
2 校验硬错（环 / 悬空依赖 / 状态非法） 8 独立性（只写临时目录，不碰现有机制文件）
|
||||
3 校验警告（无域 / 无判据 / 域重叠） 9 并发写（两进程同时写不丢状态）
|
||||
4 就绪集（依赖 / 域冲突 / 并发上限） 10 租约（超时回收三种策略 / 名额释放 / 节点级覆盖）
|
||||
5 状态流转（pending→running→done） 11 重试与触发规则（三种退避 / 三档 deps_rule）
|
||||
6 非法操作（重复开工 / 非法 state） 12 动态展开（over / from / N=0 / N 大 / child_domain）
|
||||
|
||||
## 跑法
|
||||
python graph/test_graphctl.py -v
|
||||
@@ -17,7 +19,6 @@
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import atexit
|
||||
import atexit
|
||||
import json
|
||||
import os
|
||||
@@ -574,6 +575,423 @@ class T9_Concurrency(Base):
|
||||
self.assertGreater(self.load().get("_rev"), r0)
|
||||
|
||||
|
||||
# ── 10 租约与超时 ───────────────────────────────────────────────────────
|
||||
class T10_Lease(Base):
|
||||
"""卡死的节点不能永远占着并发名额（2026-10-19 补）。"""
|
||||
|
||||
def backdate(self, node, sec=10000, gid=None):
|
||||
"""把租约时刻往前拨（等价于"这节点已经跑了 sec 秒没回报"）。"""
|
||||
g = self.load(gid)
|
||||
n = g["nodes"][node]
|
||||
n["lease_at_ts"] = time.time() - sec
|
||||
n["lease_at"] = time.strftime("%Y-%m-%dT%H:%M:%S",
|
||||
time.localtime(time.time() - sec))
|
||||
self.dump(g, gid)
|
||||
|
||||
def test_start_records_lease(self):
|
||||
self.init()
|
||||
ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "N1")
|
||||
n = self.load()["nodes"]["N1"]
|
||||
self.assertIn("lease_at", n)
|
||||
self.assertGreater(n["lease_at_ts"], 0)
|
||||
self.assertEqual(n["starts"], 1)
|
||||
|
||||
def test_timeout_requeues_by_default(self):
|
||||
self.init({"nodes": [{"id": "A", "domain": ["a/"], "accepts": ["x"]}]})
|
||||
ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "A")
|
||||
self.backdate("A")
|
||||
r = ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||||
self.assertEqual([x["node"] for x in r["reclaimed"]], ["A"])
|
||||
n = self.load()["nodes"]["A"]
|
||||
self.assertEqual(n["status"], "pending")
|
||||
self.assertEqual(n["attempts"], 1)
|
||||
self.assertNotIn("lease_at_ts", n)
|
||||
|
||||
def test_timeout_frees_capacity(self):
|
||||
"""cap=1，A 卡死占着名额 ⇒ 回收后名额必须还给别人。"""
|
||||
self.init({"max_parallel": 1, "nodes": [
|
||||
{"id": "A", "domain": ["a/"], "accepts": ["x"]},
|
||||
{"id": "B", "domain": ["b/"], "accepts": ["y"]}]})
|
||||
ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "A")
|
||||
r = ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||||
self.assertEqual(r["slots"], 0) # 先确认确实被占着
|
||||
self.backdate("A")
|
||||
r = ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||||
self.assertEqual(r["running"], [])
|
||||
self.assertEqual(r["slots"], 1)
|
||||
self.assertEqual(len(r["ready"]), 1)
|
||||
|
||||
def test_policy_block(self):
|
||||
self.init({"nodes": [{"id": "A", "domain": ["a/"], "accepts": ["x"]}]})
|
||||
ok("init", "--ws", self.ws, "--title", "阻塞", "--gid", "b1",
|
||||
"--on-timeout", "block", "--nodes-json", "-",
|
||||
stdin=json.dumps({"nodes": [{"id": "A", "domain": ["a/"], "accepts": ["x"]}]}))
|
||||
ok("start", "--ws", self.ws, "--gid", "b1", "--node", "A")
|
||||
self.backdate("A", gid="b1")
|
||||
ok("ready", "--ws", self.ws, "--gid", "b1")
|
||||
n = self.load("b1")["nodes"]["A"]
|
||||
self.assertEqual(n["status"], "blocked")
|
||||
self.assertEqual(n["blocked_by"], "lease_timeout")
|
||||
self.assertEqual(n["attempts"], 1)
|
||||
|
||||
def test_policy_keep_only_notices(self):
|
||||
ok("init", "--ws", self.ws, "--title", "只提示", "--gid", "k1",
|
||||
"--on-timeout", "keep", "--nodes-json", "-",
|
||||
stdin=json.dumps({"nodes": [{"id": "A", "domain": ["a/"], "accepts": ["x"]}]}))
|
||||
ok("start", "--ws", self.ws, "--gid", "k1", "--node", "A")
|
||||
self.backdate("A", gid="k1")
|
||||
r = ok("ready", "--ws", self.ws, "--gid", "k1")
|
||||
self.assertEqual(r["reclaimed"], [])
|
||||
self.assertEqual([x["node"] for x in r["lease_notices"]], ["A"])
|
||||
n = self.load("k1")["nodes"]["A"]
|
||||
self.assertEqual(n["status"], "running")
|
||||
self.assertEqual(n["attempts"], 0)
|
||||
|
||||
def test_node_level_lease_override(self):
|
||||
"""全局 3600 秒，节点自己声明 60 秒 ⇒ 按节点的算。"""
|
||||
self.init({"nodes": [{"id": "A", "domain": ["a/"], "accepts": ["x"],
|
||||
"lease_sec": 60},
|
||||
{"id": "B", "domain": ["b/"], "accepts": ["y"]}]})
|
||||
ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "A")
|
||||
ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "B")
|
||||
self.backdate("A", 120)
|
||||
self.backdate("B", 120) # B 用全局 3600，不该被回收
|
||||
r = ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||||
self.assertEqual([x["node"] for x in r["reclaimed"]], ["A"])
|
||||
|
||||
def test_no_lease_field_is_not_reclaimed(self):
|
||||
"""手改出来的 running（没有租约时刻）不判超时 —— 兼容旧图。"""
|
||||
self.init({"nodes": [{"id": "A", "domain": ["a/"], "accepts": ["x"]}]})
|
||||
self.state("A", "running")
|
||||
r = ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||||
self.assertEqual(r["reclaimed"], [])
|
||||
self.assertEqual(self.load()["nodes"]["A"]["status"], "running")
|
||||
|
||||
def test_lease_zero_disables(self):
|
||||
ok("init", "--ws", self.ws, "--title", "关闭", "--gid", "z1", "--lease-sec", "0",
|
||||
"--nodes-json", "-",
|
||||
stdin=json.dumps({"nodes": [{"id": "A", "domain": ["a/"], "accepts": ["x"]}]}))
|
||||
ok("start", "--ws", self.ws, "--gid", "z1", "--node", "A")
|
||||
self.backdate("A", 99999, gid="z1")
|
||||
r = ok("ready", "--ws", self.ws, "--gid", "z1")
|
||||
self.assertEqual(r["reclaimed"], [])
|
||||
|
||||
def test_reclaim_writes_event(self):
|
||||
self.init({"nodes": [{"id": "A", "domain": ["a/"], "accepts": ["x"]}]})
|
||||
ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "A")
|
||||
self.backdate("A")
|
||||
ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||||
evs = [e for e in self.events() if e.get("event") == "lease_timeout"]
|
||||
self.assertEqual(len(evs), 1)
|
||||
self.assertEqual(evs[0]["node"], "A")
|
||||
|
||||
|
||||
# ── 11 重试策略与依赖触发规则 ───────────────────────────────────────────
|
||||
class T11_Retry(Base):
|
||||
def _one(self, retry=None, rule=None):
|
||||
n = {"id": "A", "domain": ["a/"], "accepts": ["x"]}
|
||||
if retry:
|
||||
n["retry"] = retry
|
||||
if rule:
|
||||
n["deps_rule"] = rule
|
||||
self.init({"nodes": [n]})
|
||||
|
||||
def test_retry_requeues_with_delay(self):
|
||||
self._one({"max": 2, "delay_sec": 30})
|
||||
r = ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A",
|
||||
"--state", "blocked")
|
||||
self.assertEqual(r["status"], "pending")
|
||||
self.assertTrue(r["retry"]["requeued"])
|
||||
self.assertEqual(r["retry"]["delay_sec"], 30)
|
||||
n = self.load()["nodes"]["A"]
|
||||
self.assertEqual(n["status"], "pending")
|
||||
self.assertEqual(n["attempts"], 1)
|
||||
self.assertGreater(n["not_before"], time.time() + 20)
|
||||
|
||||
def test_retry_waits_until_not_before(self):
|
||||
self._one({"max": 2, "delay_sec": 600})
|
||||
ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", "--state", "blocked")
|
||||
r = ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||||
self.assertNotIn("A", r["ready"])
|
||||
self.assertTrue(any(w["node"] == "A" and "重试等待中" in w["why"]
|
||||
for w in r["waiting"]))
|
||||
|
||||
def test_retry_ready_after_delay(self):
|
||||
self._one({"max": 2, "delay_sec": 600})
|
||||
ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", "--state", "blocked")
|
||||
g = self.load()
|
||||
g["nodes"]["A"]["not_before"] = time.time() - 1
|
||||
self.dump(g)
|
||||
r = ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||||
self.assertIn("A", r["ready"])
|
||||
|
||||
def test_retry_exhausted_blocks(self):
|
||||
self._one({"max": 1, "delay_sec": 0})
|
||||
r1 = ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A",
|
||||
"--state", "blocked")
|
||||
self.assertEqual(r1["status"], "pending") # 第 1 次失败 ⇒ 还给 1 次
|
||||
r2 = ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A",
|
||||
"--state", "blocked")
|
||||
self.assertEqual(r2["status"], "blocked")
|
||||
n = self.load()["nodes"]["A"]
|
||||
self.assertEqual(n["attempts"], 2)
|
||||
self.assertTrue(n["retry_exhausted"])
|
||||
|
||||
def test_backoff_exponential_grows(self):
|
||||
self._one({"max": 3, "delay_sec": 10, "backoff": "exponential", "factor": 2})
|
||||
r1 = ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A",
|
||||
"--state", "blocked")
|
||||
self.assertEqual(r1["retry"]["delay_sec"], 10)
|
||||
g = self.load()
|
||||
g["nodes"]["A"]["not_before"] = time.time() - 1
|
||||
self.dump(g)
|
||||
r2 = ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A",
|
||||
"--state", "blocked")
|
||||
self.assertEqual(r2["retry"]["delay_sec"], 20)
|
||||
|
||||
def test_backoff_linear(self):
|
||||
self._one({"max": 3, "delay_sec": 10, "backoff": "linear"})
|
||||
r = ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A",
|
||||
"--state", "blocked")
|
||||
self.assertEqual(r["retry"]["delay_sec"], 10)
|
||||
g = self.load()
|
||||
g["nodes"]["A"]["not_before"] = time.time() - 1
|
||||
self.dump(g)
|
||||
r2 = ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A",
|
||||
"--state", "blocked")
|
||||
self.assertEqual(r2["retry"]["delay_sec"], 20) # 10 × 2
|
||||
|
||||
def test_default_no_auto_retry(self):
|
||||
"""没写 retry ⇒ 与旧行为一致：blocked 立刻 blocked（⛔ 不许偷偷重排）。"""
|
||||
self._one()
|
||||
r = ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A",
|
||||
"--state", "blocked")
|
||||
self.assertEqual(r["status"], "blocked")
|
||||
self.assertEqual(self.load()["nodes"]["A"]["attempts"], 1)
|
||||
|
||||
def _dep_case(self, rule):
|
||||
child = {"id": "C", "domain": ["c/"], "accepts": ["x"], "deps": ["A", "B"]}
|
||||
if rule:
|
||||
child["deps_rule"] = rule
|
||||
self.init({"nodes": [
|
||||
{"id": "A", "domain": ["a/"], "accepts": ["x"]},
|
||||
{"id": "B", "domain": ["b/"], "accepts": ["y"]},
|
||||
child]})
|
||||
|
||||
def test_deps_rule_default_all_success(self):
|
||||
self._dep_case(None)
|
||||
ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", "--state", "done")
|
||||
ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "B", "--state", "blocked")
|
||||
r = ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||||
self.assertNotIn("C", r["ready"])
|
||||
|
||||
def test_deps_rule_all_done(self):
|
||||
"""all_done：上游跑完就算（成败不论）。"""
|
||||
self._dep_case("all_done")
|
||||
ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", "--state", "done")
|
||||
ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "B", "--state", "blocked")
|
||||
r = ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||||
self.assertIn("C", r["ready"])
|
||||
|
||||
def test_deps_rule_all_done_waits_pending(self):
|
||||
self._dep_case("all_done")
|
||||
ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", "--state", "done")
|
||||
r = ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||||
self.assertNotIn("C", r["ready"])
|
||||
self.assertTrue(any(w["node"] == "C" and "依赖未跑完" in w["why"]
|
||||
for w in r["waiting"]))
|
||||
|
||||
def test_deps_rule_one_success(self):
|
||||
self._dep_case("one_success")
|
||||
ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", "--state", "done")
|
||||
ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "B", "--state", "blocked")
|
||||
r = ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||||
self.assertIn("C", r["ready"])
|
||||
|
||||
def test_deps_rule_one_success_none_yet(self):
|
||||
self._dep_case("one_success")
|
||||
r = ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||||
self.assertNotIn("C", r["ready"])
|
||||
self.assertTrue(any(w["node"] == "C" and "任一上游成功" in w["why"]
|
||||
for w in r["waiting"]))
|
||||
|
||||
def test_illegal_deps_rule_rejected(self):
|
||||
err = bad("init", "--ws", self.ws, "--title", "坏规则", "--gid", "bad",
|
||||
"--nodes-json", "-",
|
||||
stdin=json.dumps({"nodes": [{"id": "A", "deps_rule": "any"}]}))
|
||||
self.assertIn("deps_rule", err)
|
||||
|
||||
def test_illegal_deps_rule_caught_by_check(self):
|
||||
self.init({"nodes": [{"id": "A", "accepts": ["x"]}]})
|
||||
g = self.load()
|
||||
g["nodes"]["A"]["deps_rule"] = "any"
|
||||
self.dump(g)
|
||||
r = json.loads(_run("check", "--ws", self.ws, "--gid", self.gid)[1])
|
||||
self.assertFalse(r["ok"])
|
||||
self.assertTrue(any("deps_rule" in e for e in r["errors"]))
|
||||
|
||||
|
||||
# ── 12 动态展开（map / foreach）────────────────────────────────────────
|
||||
class T12_Map(Base):
|
||||
def test_expand_from_list(self):
|
||||
self.init({"max_parallel": 5, "nodes": [
|
||||
{"id": "P", "title": "逐页改", "accepts": ["都改完"],
|
||||
"map": {"over": ["a.md", "b.md", "c.md"]}}]})
|
||||
r = ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||||
self.assertEqual(r["expanded"][0]["node"], "P")
|
||||
self.assertEqual(r["expanded"][0]["count"], 3)
|
||||
g = self.load()
|
||||
self.assertEqual(g["nodes"]["P"]["status"], "gating")
|
||||
self.assertEqual(g["nodes"]["P"]["children"], ["P#1", "P#2", "P#3"])
|
||||
self.assertEqual(g["nodes"]["P#2"]["item"], "b.md")
|
||||
self.assertEqual(sorted(r["ready"]), ["P#1", "P#2", "P#3"])
|
||||
|
||||
def test_children_done_completes_parent(self):
|
||||
self.init({"max_parallel": 5, "nodes": [
|
||||
{"id": "P", "accepts": ["齐"], "map": {"over": ["x", "y"]}},
|
||||
{"id": "Z", "domain": ["z/"], "accepts": ["收尾"], "deps": ["P"]}]})
|
||||
ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||||
ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "P#1", "--state", "done")
|
||||
r = ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "P#2", "--state", "done")
|
||||
g = self.load()
|
||||
self.assertEqual(g["nodes"]["P"]["status"], "done")
|
||||
self.assertEqual(r["completed_parents"], ["P"])
|
||||
self.assertIn("Z", r["next_ready"])
|
||||
|
||||
def test_parent_not_done_while_children_left(self):
|
||||
self.init({"max_parallel": 5, "nodes": [
|
||||
{"id": "P", "accepts": ["齐"], "map": {"over": ["x", "y"]}}]})
|
||||
ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||||
ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "P#1", "--state", "done")
|
||||
g = self.load()
|
||||
self.assertEqual(g["nodes"]["P"]["status"], "gating")
|
||||
r = ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||||
self.assertTrue(any(w["node"] == "P" and "子节点" in w["why"] for w in r["waiting"]))
|
||||
|
||||
def test_zero_items_completes_immediately(self):
|
||||
self.init({"nodes": [{"id": "P", "accepts": ["齐"],
|
||||
"map": {"over": []}}]})
|
||||
r = ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||||
g = self.load()
|
||||
self.assertEqual(g["nodes"]["P"]["status"], "done")
|
||||
self.assertEqual(g["nodes"]["P"]["children"], [])
|
||||
self.assertEqual([x for x in g["nodes"]], ["P"])
|
||||
self.assertEqual(r["ready"], [])
|
||||
|
||||
def test_expand_from_evidence_lines(self):
|
||||
self.init({"max_parallel": 4, "nodes": [
|
||||
{"id": "U", "domain": ["u/"], "accepts": ["清单在"]},
|
||||
{"id": "P", "accepts": ["都处理"],
|
||||
"deps": ["U"], "map": {"from": {"node": "U", "pick": "evidence"}}}]})
|
||||
ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "U", "--state", "done",
|
||||
"--evidence", "a.md\nb.md；c.md")
|
||||
ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||||
self.assertEqual(self.load()["nodes"]["P"]["children"], ["P#1", "P#2", "P#3"])
|
||||
|
||||
def test_expand_from_artifacts(self):
|
||||
self.init({"max_parallel": 4, "nodes": [
|
||||
{"id": "U", "domain": ["u/"], "accepts": ["清单在"]},
|
||||
{"id": "P", "accepts": ["都处理"],
|
||||
"deps": ["U"], "map": {"from": {"node": "U", "pick": "artifacts"}}}]})
|
||||
ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "U", "--state", "done",
|
||||
"--artifact", "a.md", "--artifact", "b.md")
|
||||
ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||||
self.assertEqual(self.load()["nodes"]["P"]["children"], ["P#1", "P#2"])
|
||||
|
||||
def test_expand_source_not_ready_waits(self):
|
||||
self.init({"nodes": [
|
||||
{"id": "U", "domain": ["u/"], "accepts": ["清单在"]},
|
||||
{"id": "P", "accepts": ["都处理"],
|
||||
"deps": ["U"], "map": {"from": {"node": "U", "pick": "evidence"}}}]})
|
||||
r = ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||||
self.assertNotIn("P", r["ready"])
|
||||
self.assertEqual(self.load()["nodes"]["P"]["status"], "pending")
|
||||
|
||||
def test_large_n_blocked_by_max_parallel(self):
|
||||
"""N=10 但 cap=2 ⇒ 这一批只放 2 个，其余进 backlog（⛔ 不许一次性全放出去）。"""
|
||||
self.init({"max_parallel": 2, "nodes": [
|
||||
{"id": "P", "accepts": ["齐"], "map": {"over": ["i%d" % i for i in range(1, 11)]}}]})
|
||||
r = ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||||
self.assertEqual(len(self.load()["nodes"]["P"]["children"]), 10)
|
||||
self.assertEqual(len(r["ready"]), 2)
|
||||
self.assertEqual(len(r["backlog"]), 8)
|
||||
|
||||
def test_child_domain_template_allows_parallel(self):
|
||||
self.init({"max_parallel": 3, "nodes": [
|
||||
{"id": "P", "accepts": ["齐"],
|
||||
"map": {"over": ["a", "b"], "child_domain": "out/{i}/"}}]})
|
||||
r = ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||||
g = self.load()
|
||||
self.assertEqual(g["nodes"]["P#1"]["domain"], ["out/1/"])
|
||||
self.assertEqual(g["nodes"]["P#2"]["domain"], ["out/2/"])
|
||||
self.assertEqual(sorted(r["ready"]), ["P#1", "P#2"])
|
||||
|
||||
def test_children_default_have_no_domain(self):
|
||||
"""默认不继承父域：同一批展开通常写同一批文件，继承会把并发挡死。"""
|
||||
self.init({"max_parallel": 3, "nodes": [
|
||||
{"id": "P", "domain": ["docs/"], "accepts": ["齐"],
|
||||
"map": {"over": ["a", "b"]}}]})
|
||||
ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||||
g = self.load()
|
||||
self.assertEqual(g["nodes"]["P#1"]["domain"], [])
|
||||
self.assertEqual(len(ok("ready", "--ws", self.ws, "--gid", self.gid)["ready"]), 2)
|
||||
|
||||
def test_start_map_node_rejected(self):
|
||||
self.init({"nodes": [{"id": "P", "accepts": ["齐"],
|
||||
"map": {"over": ["a"]}}]})
|
||||
err = bad("start", "--ws", self.ws, "--gid", self.gid, "--node", "P")
|
||||
self.assertIn("展开节点", err)
|
||||
|
||||
def test_report_on_map_node_rejected(self):
|
||||
self.init({"nodes": [{"id": "P", "accepts": ["齐"],
|
||||
"map": {"over": ["a"]}}]})
|
||||
ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||||
err = bad("report", "--ws", self.ws, "--gid", self.gid, "--node", "P",
|
||||
"--state", "done")
|
||||
self.assertIn("展开父节点", err)
|
||||
|
||||
def test_check_invalid_map(self):
|
||||
self.init({"nodes": [{"id": "P", "accepts": ["齐"]}]})
|
||||
g = self.load()
|
||||
g["nodes"]["P"]["map"] = {"as": "item"}
|
||||
self.dump(g)
|
||||
r = json.loads(_run("check", "--ws", self.ws, "--gid", self.gid)[1])
|
||||
self.assertFalse(r["ok"])
|
||||
self.assertTrue(any("map" in e for e in r["errors"]))
|
||||
|
||||
def test_init_invalid_map_rejected(self):
|
||||
err = bad("init", "--ws", self.ws, "--title", "坏展开", "--gid", "bad",
|
||||
"--nodes-json", "-",
|
||||
stdin=json.dumps({"nodes": [{"id": "P", "map": {"over": "不是列表"}}]}))
|
||||
self.assertIn("over", err)
|
||||
|
||||
def test_expand_command_manual(self):
|
||||
self.init({"nodes": [{"id": "P", "accepts": ["齐"],
|
||||
"map": {"over": ["a", "b"]}}]})
|
||||
r = ok("expand", "--ws", self.ws, "--gid", self.gid, "--node", "P")
|
||||
self.assertEqual(r["count"], 2)
|
||||
err = bad("expand", "--ws", self.ws, "--gid", self.gid, "--node", "P")
|
||||
self.assertIn("已展开", err)
|
||||
|
||||
def test_expand_waits_for_deps(self):
|
||||
"""有依赖的展开节点：上游没完成时手动 expand 要拒绝（除非 --force）。"""
|
||||
self.init({"nodes": [{"id": "U", "domain": ["u/"], "accepts": ["x"]},
|
||||
{"id": "P", "accepts": ["齐"], "deps": ["U"],
|
||||
"map": {"over": ["a"]}}]})
|
||||
err = bad("expand", "--ws", self.ws, "--gid", self.gid, "--node", "P")
|
||||
self.assertIn("未就绪", err)
|
||||
|
||||
def test_child_attempts_independent(self):
|
||||
self.init({"max_parallel": 3, "nodes": [
|
||||
{"id": "P", "accepts": ["齐"], "map": {"over": ["a", "b"]}}]})
|
||||
ok("ready", "--ws", self.ws, "--gid", self.gid)
|
||||
ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "P#1", "--state", "blocked")
|
||||
g = self.load()
|
||||
self.assertEqual(g["nodes"]["P#1"]["attempts"], 1)
|
||||
self.assertEqual(g["nodes"]["P#2"]["attempts"], 0)
|
||||
self.assertEqual(g["nodes"]["P"]["attempts"], 0)
|
||||
|
||||
|
||||
def tearDownModule():
|
||||
_drop(TMP_ROOT)
|
||||
|
||||
|
||||
Reference in new issue
Block a user