537 lines
22 KiB
Python
537 lines
22 KiB
Python
"""审计修复批 6 测试：B1、I1、I2、M1、M2、M5、M6、M7。
|
||
|
||
只用注入的假 runner 与 init_db 的本地库，不跑真实模型回合、
|
||
不联网。按任务书：M3（报告措辞）、M4（观察项）、M8/M9
|
||
（已知边界）不需要代码，本批只在报告里写明不做；N-12 的
|
||
goal 包前言续跑缺口不在本批范围。
|
||
|
||
行为变更锁定（本批有意改语义，测试即契约）：
|
||
- B1：run_one 入口初始化 _gates_checked=[]，零门任务两轮后
|
||
passed_gates 不再含硬编码三门；政策任务标记不受影响。
|
||
- I1：_GATE_ERROR_MARKERS 补全长措辞
|
||
"known to lack a required capability"，无结构化行时显式拦
|
||
与复核拒绝仍归 gate_rejected 且有回落审计行。
|
||
- I2：接线 error 时返回行实际状态（不再是输入终局），
|
||
run_one 终局块在行已不是终态时抑制终态审计与通知；
|
||
批 5 的 test_escalate_cas_miss_audits_skipped 断言意图
|
||
随之改写（final == 行状态 queued，不再是输入终局 failed）。
|
||
- M1：轮次耗尽 + 无 finding 时 reason 为 max_rounds_exceeded
|
||
（不再被 gate_failed_without_finding 掩盖），决策不变。
|
||
- M2：两处 worker_routing 写点均以 chosen 结尾（行尾约定）。
|
||
- M5：升级结果写回有界重试一次（仍带 only_if_status 守卫）。
|
||
- M6：_coord_clear_packets 在清包前即失败时包列原样不动。
|
||
- M7：pass 清包 CAS 未命中按 I2 新行为断言。
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
import sqlite3
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
import unittest
|
||
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
from unittest import mock
|
||
|
||
try:
|
||
from cloud.audit_loop import LoopState, evaluate
|
||
from cloud.auth import AuthStore
|
||
from cloud.routing_shadow import parse_worker_routing_detail
|
||
from cloud.server import init_db
|
||
from cloud.worker import Worker, _GATE_ERROR_MARKERS
|
||
import cloud.worker as worker_module
|
||
except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
||
from cloud.audit_loop import LoopState, evaluate
|
||
from cloud.auth import AuthStore
|
||
from cloud.routing_shadow import parse_worker_routing_detail
|
||
from cloud.server import init_db
|
||
from cloud.worker import Worker, _GATE_ERROR_MARKERS
|
||
import cloud.worker as worker_module
|
||
|
||
|
||
def _insert_task(db_path, task_id, model=None, status="queued",
|
||
project_id="p1", audit_loop=0, parent=None, part=None,
|
||
error=None, result=None, packet=None):
|
||
now = time.time()
|
||
conn = sqlite3.connect(db_path)
|
||
try:
|
||
conn.execute(
|
||
"INSERT INTO tasks(id, project_id, prompt, status,"
|
||
" idempotency_key, depends_on, priority, created_by,"
|
||
" created_at, updated_at, last_heartbeat, audit_loop,"
|
||
" parent_task_id, coordinator_part, result_text, error_text,"
|
||
" model, audit_loop_packet)"
|
||
" VALUES(?, ?, 'do work', ?, NULL, '[]', 0, 'tester',"
|
||
" ?, ?, NULL, ?, ?, ?, ?, ?, ?, ?)",
|
||
(task_id, project_id, status, now, now, audit_loop,
|
||
parent, part, result, error, model, packet))
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def _set_started_at(db_path, task_id, started_at):
|
||
conn = sqlite3.connect(db_path)
|
||
try:
|
||
conn.execute(
|
||
"UPDATE tasks SET started_at = ? WHERE id = ?",
|
||
(started_at, task_id))
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def _insert_audit(db_path, project_id, action, detail, created_at=None):
|
||
conn = sqlite3.connect(db_path)
|
||
try:
|
||
conn.execute(
|
||
"INSERT INTO audit_log(user_id, project_id, action, detail,"
|
||
" created_at) VALUES(NULL, ?, ?, ?, ?)",
|
||
(project_id, action, detail,
|
||
created_at if created_at is not None else time.time()))
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def _task_row(db_path, task_id):
|
||
conn = sqlite3.connect(db_path)
|
||
try:
|
||
return conn.execute(
|
||
"SELECT status, audit_loop_packet FROM tasks WHERE id = ?",
|
||
(task_id,)).fetchone()
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def _state_row(db_path, task_id):
|
||
conn = sqlite3.connect(db_path)
|
||
try:
|
||
return conn.execute(
|
||
"SELECT round, passed_gates, last_decision"
|
||
" FROM task_audit_loop_state WHERE task_id = ?",
|
||
(task_id,)).fetchone()
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def _seed_state(db_path, task_id, round_no, last_decision,
|
||
passed='["execution"]'):
|
||
conn = sqlite3.connect(db_path)
|
||
try:
|
||
conn.execute(
|
||
"CREATE TABLE IF NOT EXISTS task_audit_loop_state("
|
||
"task_id TEXT PRIMARY KEY, round INTEGER,"
|
||
" passed_gates TEXT, criteria TEXT, last_decision TEXT,"
|
||
" updated_at DOUBLE PRECISION)")
|
||
conn.execute(
|
||
"INSERT OR REPLACE INTO task_audit_loop_state"
|
||
"(task_id, round, passed_gates, criteria, last_decision,"
|
||
" updated_at) VALUES(?, ?, ?, ?, ?, ?)",
|
||
(task_id, round_no, passed, '{}', last_decision,
|
||
time.time()))
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def _routing_details(db_path):
|
||
conn = sqlite3.connect(db_path)
|
||
try:
|
||
return [r[0] for r in conn.execute(
|
||
"SELECT detail FROM audit_log WHERE action = 'worker_routing'"
|
||
" ORDER BY id").fetchall()]
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def _completed_runner(cmd):
|
||
name = cmd[cmd.index("--name") + 1] if "dispatch" in cmd else "t"
|
||
return SimpleNamespace(
|
||
returncode=0,
|
||
stdout=json.dumps({"task": name, "status": "completed",
|
||
"text": "done"}),
|
||
stderr="")
|
||
|
||
|
||
def _failing_runner(cmd):
|
||
name = cmd[cmd.index("--name") + 1] if "dispatch" in cmd else "t"
|
||
return SimpleNamespace(
|
||
returncode=0,
|
||
stdout=json.dumps({"task": name, "status": "failed",
|
||
"text": "boom"}),
|
||
stderr="")
|
||
|
||
|
||
def _worker(db_path, policy_doc=None, env=None, gateway=None,
|
||
runner=None):
|
||
calls = []
|
||
|
||
def recording(cmd):
|
||
calls.append(list(cmd))
|
||
return (runner or _completed_runner)(cmd)
|
||
|
||
worker = Worker(db_path, "/fake/host.py", runner=recording,
|
||
env=env or {"XCM_API_KEY": "test-placeholder"},
|
||
routing_policy=policy_doc)
|
||
if gateway is not None:
|
||
worker._ensure_gateway = lambda task: gateway
|
||
worker._sub2api_default_model = lambda e: "gateway-model"
|
||
worker._calls = calls
|
||
return worker
|
||
|
||
|
||
class _RunOneCase(unittest.TestCase):
|
||
"""run_one 级夹具（批 4 GateResultsCheckedTestCase 同款口径）。"""
|
||
|
||
def setUp(self):
|
||
self._old = os.environ.get("CODEX_PUSH_ALLOW_PRIVATE_TARGETS")
|
||
os.environ["CODEX_PUSH_ALLOW_PRIVATE_TARGETS"] = "1"
|
||
self.tmp = tempfile.TemporaryDirectory()
|
||
self.db_path = os.path.join(self.tmp.name, "control.db")
|
||
init_db(self.db_path)
|
||
|
||
def tearDown(self):
|
||
if self._old is None:
|
||
os.environ.pop("CODEX_PUSH_ALLOW_PRIVATE_TARGETS", None)
|
||
else:
|
||
os.environ[CODEX_PUSH_ALLOW_PRIVATE_TARGETS] = self._old
|
||
self.tmp.cleanup()
|
||
|
||
|
||
class B1ZeroGateTestCase(_RunOneCase):
|
||
"""B1：run_one 入口初始化 _gates_checked，零门不再硬编码三门。"""
|
||
|
||
def test_explicit_zero_gate_two_rounds_passed_gates_empty(self):
|
||
# 显式任务、无网关、无策略、无工具、开审计循环：
|
||
# 旧代码两轮后 passed_gates 是三门硬编码（漏网的直接
|
||
# 原因正是本例缺 passed_gates 断言）；新代码应为空集。
|
||
_insert_task(self.db_path, "T3", model="explicit-model",
|
||
audit_loop=1)
|
||
worker = _worker(self.db_path, gateway=(None, {}),
|
||
runner=_failing_runner)
|
||
for _ in range(2):
|
||
out = worker.run_one()
|
||
self.assertEqual(out["status"], "queued")
|
||
row = _state_row(self.db_path, "T3")
|
||
self.assertIsNotNone(row)
|
||
self.assertEqual(row[0], 2)
|
||
self.assertEqual(json.loads(row[1]), [])
|
||
self.assertEqual(row[2], "pushback")
|
||
|
||
def test_policy_two_rounds_marks_survive_init(self):
|
||
# 对照：政策任务本轮实际跑过的 capability 标记不能被
|
||
# 入口初始化冲掉，两轮后 passed_gates 仍为 ["capability"]。
|
||
_insert_task(self.db_path, "T2", audit_loop=1)
|
||
worker = _worker(self.db_path, {
|
||
"mode": "active",
|
||
"tiers": {"default": {"model": "policy-model",
|
||
"requires": {"tool_calling": True}}},
|
||
"model_capabilities": {
|
||
"policy-model": {"tool_calling": True},
|
||
},
|
||
}, gateway=(None, {}), runner=_failing_runner)
|
||
for _ in range(2):
|
||
out = worker.run_one()
|
||
self.assertEqual(out["status"], "queued")
|
||
row = _state_row(self.db_path, "T2")
|
||
self.assertIsNotNone(row)
|
||
self.assertEqual(row[0], 2)
|
||
self.assertEqual(json.loads(row[1]), ["capability"])
|
||
self.assertEqual(row[2], "pushback")
|
||
|
||
|
||
class I1MarkerTestCase(unittest.TestCase):
|
||
"""I1：无结构化行时显式拦与复核拒绝仍归 gate_rejected。"""
|
||
|
||
def setUp(self):
|
||
self._old = os.environ.get("CODEX_PUSH_ALLOW_PRIVATE_TARGETS")
|
||
os.environ["CODEX_PUSH_ALLOW_PRIVATE_TARGETS"] = "1"
|
||
self.tmp = tempfile.TemporaryDirectory()
|
||
self.db_path = os.path.join(self.tmp.name, "control.db")
|
||
init_db(self.db_path)
|
||
store = AuthStore(self.db_path)
|
||
store.conn.close()
|
||
self.audits = []
|
||
self.worker = Worker(self.db_path, "/fake/host.py",
|
||
runner=_completed_runner,
|
||
env={"XCM_API_KEY": "test-placeholder"},
|
||
max_retries=0, retry_backoff_seconds=0)
|
||
self.worker._audit = lambda p, a, d="": self.audits.append(
|
||
(p, a, d))
|
||
conn = sqlite3.connect(self.db_path)
|
||
try:
|
||
self.worker._ensure_result_columns(conn)
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
|
||
def tearDown(self):
|
||
if self._old is None:
|
||
os.environ.pop("CODEX_PUSH_ALLOW_PRIVATE_TARGETS", None)
|
||
else:
|
||
os.environ[CODEX_PUSH_ALLOW_PRIVATE_TARGETS] = self._old
|
||
self.tmp.cleanup()
|
||
|
||
def _merge(self, child_error, audit_rows=()):
|
||
now = time.time()
|
||
_insert_task(self.db_path, "P1", status="running", audit_loop=1,
|
||
parent=None)
|
||
_insert_task(self.db_path, "C1", status="failed", parent="P1",
|
||
part="a", error=child_error)
|
||
_set_started_at(self.db_path, "C1", now - 100)
|
||
for action, detail in audit_rows:
|
||
_insert_audit(self.db_path, "p1", action, detail,
|
||
created_at=now)
|
||
return self.worker._coord_audit_loop_merge(
|
||
{"id": "P1", "project_id": "p1", "audit_loop": 1},
|
||
[{"name": "a", "prompt": "do a"}],
|
||
[{"id": "C1", "name": "a"}])
|
||
|
||
def test_explicit_block_error_only_is_gate_rejected(self):
|
||
# 显式网关拦的生产方 error 文案（worker.py 网关块），
|
||
# 无 gate_rejected 行、无 gateway_explicit 行时仍须
|
||
# 走子串回落归 gate_rejected（旧代码误归 execution
|
||
# 而推回）。回落审计行须点名新补的 marker。
|
||
out = self._merge(
|
||
"default tier explicit model m lacks capability:"
|
||
" tool_calling; refusing to dispatch a model known to"
|
||
" lack a required capability")
|
||
self.assertEqual(out["decision"], "escalate")
|
||
self.assertIn("a", out.get("gate_rejected", []))
|
||
self.assertEqual(_task_row(self.db_path, "C1")[0], "failed")
|
||
marker_rows = [d for _p, a, d in self.audits
|
||
if a == "worker_coord_gate_classified_by_marker"]
|
||
self.assertTrue(marker_rows)
|
||
self.assertIn("C1", marker_rows[0])
|
||
self.assertIn("marker=known to lack a required capability",
|
||
marker_rows[0])
|
||
|
||
def test_recheck_reject_error_only_is_gate_rejected(self):
|
||
# 写回复核拒绝的生产方 error 文案（worker.py 复核块），
|
||
# 同款无结构化行情形。
|
||
out = self._merge(
|
||
"default tier persisted model m lacks capability:"
|
||
" tool_calling; refusing to dispatch a model known to"
|
||
" lack a required capability")
|
||
self.assertEqual(out["decision"], "escalate")
|
||
self.assertIn("a", out.get("gate_rejected", []))
|
||
self.assertEqual(_task_row(self.db_path, "C1")[0], "failed")
|
||
marker_rows = [d for _p, a, d in self.audits
|
||
if a == "worker_coord_gate_classified_by_marker"]
|
||
self.assertTrue(marker_rows)
|
||
self.assertIn("C1", marker_rows[0])
|
||
self.assertIn("marker=known to lack a required capability",
|
||
marker_rows[0])
|
||
|
||
|
||
class _WireCase(unittest.TestCase):
|
||
"""接线级夹具（批 5 _WorkerCase 同款口径，另加 run_one 级
|
||
测试所需的 CODEX_PUSH_ALLOW_PRIVATE_TARGETS 环境）。"""
|
||
|
||
def setUp(self):
|
||
self._old = os.environ.get("CODEX_PUSH_ALLOW_PRIVATE_TARGETS")
|
||
os.environ["CODEX_PUSH_ALLOW_PRIVATE_TARGETS"] = "1"
|
||
self.tmp = tempfile.TemporaryDirectory()
|
||
self.db_path = os.path.join(self.tmp.name, "control.db")
|
||
init_db(self.db_path)
|
||
self.audits = []
|
||
|
||
def tearDown(self):
|
||
if self._old is None:
|
||
os.environ.pop("CODEX_PUSH_ALLOW_PRIVATE_TARGETS", None)
|
||
else:
|
||
os.environ[CODEX_PUSH_ALLOW_PRIVATE_TARGETS] = self._old
|
||
self.tmp.cleanup()
|
||
|
||
def _worker(self, **kwargs):
|
||
kwargs.setdefault("env", {"XCM_API_KEY": "test-placeholder"})
|
||
kwargs.setdefault("max_retries", 0)
|
||
kwargs.setdefault("retry_backoff_seconds", 0)
|
||
kwargs.setdefault("runner", _completed_runner)
|
||
worker = Worker(self.db_path, "/fake/host.py", **kwargs)
|
||
worker._audit = lambda p, a, d="": self.audits.append((p, a, d))
|
||
return worker
|
||
|
||
def _actions(self):
|
||
return [a for _, a, _ in self.audits]
|
||
|
||
|
||
class I2WireTestCase(_WireCase):
|
||
"""I2 + M7：接线 error 返回行实际状态；pass CAS 未命中同款。"""
|
||
|
||
def test_pass_cas_miss_returns_row_status_and_keeps_packet(self):
|
||
# M7：pass 清包 CAS 未命中——行已被并发重排回 queued，
|
||
# 接线返回 error + 行状态 queued（不再是输入终局
|
||
# completed），旧包原样不动（事务回滚），状态行不动。
|
||
_insert_task(self.db_path, "TP2", status="queued", audit_loop=1,
|
||
packet='{"round": 1}')
|
||
_seed_state(self.db_path, "TP2", 1, "pushback")
|
||
worker = self._worker()
|
||
task = worker.get_task("TP2")
|
||
final, _, decision = worker._audit_loop_wire(
|
||
task, "completed", {"stdout": ""}, "boom")
|
||
self.assertEqual(decision, "error")
|
||
self.assertEqual(final, "queued")
|
||
row = _task_row(self.db_path, "TP2")
|
||
self.assertEqual(row[0], "queued")
|
||
self.assertEqual(row[1], '{"round": 1}')
|
||
state = _state_row(self.db_path, "TP2")
|
||
self.assertEqual(state[0], 1)
|
||
self.assertEqual(state[2], "pushback")
|
||
self.assertNotIn("worker_audit_loop_passed", self._actions())
|
||
|
||
def test_run_one_suppresses_terminal_on_error_nonterminal(self):
|
||
# I2 终局块：接线 error 且行已不是终态时，抑制终态审计
|
||
# 与 failed 通知，返回体状态与行一致（queued）。
|
||
_insert_task(self.db_path, "TS", status="queued", audit_loop=1)
|
||
worker = self._worker(runner=_failing_runner)
|
||
notifies = []
|
||
worker._notify_task_finished = lambda *a, **k: notifies.append(
|
||
(a, k))
|
||
worker._audit_loop_wire = lambda t, f, r, e: (
|
||
"queued", "boom", "error")
|
||
out = worker.run_one()
|
||
self.assertEqual(out["status"], "queued")
|
||
self.assertNotIn("worker_failed", self._actions())
|
||
self.assertNotIn("worker_completed", self._actions())
|
||
self.assertFalse([n for n in notifies if n[0][2] == "failed"])
|
||
|
||
def test_run_one_terminal_still_runs_on_error_terminal(self):
|
||
# 对照：接线 error 但行仍是终态 failed 时，终态审计与
|
||
# failed 通知照常跑——抑制只针对“行已不是终态”情形，
|
||
# 不是见 error 就全抑。
|
||
_insert_task(self.db_path, "TF", status="queued", audit_loop=1)
|
||
worker = self._worker(runner=_failing_runner)
|
||
notifies = []
|
||
worker._notify_task_finished = lambda *a, **k: notifies.append(
|
||
(a, k))
|
||
worker._audit_loop_wire = lambda t, f, r, e: (
|
||
"failed", "boom", "error")
|
||
out = worker.run_one()
|
||
self.assertEqual(out["status"], "failed")
|
||
self.assertIn("worker_failed", self._actions())
|
||
self.assertTrue([n for n in notifies if n[0][2] == "failed"])
|
||
|
||
|
||
class M1MaxRoundsLabelTestCase(unittest.TestCase):
|
||
"""M1：轮次耗尽 + 无 finding 时 reason 为 max_rounds_exceeded。"""
|
||
|
||
def test_max_rounds_label_wins_over_no_finding(self):
|
||
# 上一轮未通过过的门本轮失败、无 finding、无回归、无缺门：
|
||
# 旧代码 reason 是 gate_failed_without_finding；新代码
|
||
# max_rounds 分支在前，reason 为 max_rounds_exceeded，
|
||
# 决策同为 escalate、轮次不变。
|
||
state = LoopState("T1", round=3, passed_gates=[],
|
||
criteria={}, last_decision="pushback")
|
||
out = evaluate(state, "T1", {"execution": False}, findings=[])
|
||
self.assertEqual(out["decision"], "escalate")
|
||
self.assertEqual(out["reason"], "max_rounds_exceeded")
|
||
self.assertEqual(out["new_state"].round, 3)
|
||
self.assertEqual(out["new_state"].last_decision, "escalate")
|
||
|
||
|
||
class M2ChosenLastTestCase(_RunOneCase):
|
||
"""M2：两处 worker_routing 写点均以 chosen 结尾（行尾约定）。"""
|
||
|
||
def _assert_chosen_last(self, details):
|
||
self.assertTrue(details)
|
||
for detail in details:
|
||
parsed = parse_worker_routing_detail(detail)
|
||
self.assertIsNotNone(parsed, detail)
|
||
last_token = detail.strip().rsplit(" ", 1)[-1]
|
||
self.assertTrue(last_token.startswith("chosen="), detail)
|
||
|
||
def test_policy_routing_row_ends_with_chosen(self):
|
||
# 策略/shadow 决策路径的 worker_routing 行。
|
||
_insert_task(self.db_path, "M1", audit_loop=0)
|
||
worker = _worker(self.db_path, {
|
||
"mode": "shadow",
|
||
"tiers": {"default": "policy-model"},
|
||
}, gateway=(None, {}), runner=_failing_runner)
|
||
worker.run_one()
|
||
details = _routing_details(self.db_path)
|
||
self.assertTrue([d for d in details if "source=shadow" in d])
|
||
self._assert_chosen_last(
|
||
[d for d in details if "source=shadow" in d])
|
||
|
||
def test_explicit_routing_row_ends_with_chosen(self):
|
||
# 显式任务路径的 worker_routing 行（source=explicit）。
|
||
_insert_task(self.db_path, "M2", model="explicit-model",
|
||
audit_loop=0)
|
||
worker = _worker(self.db_path, gateway=(None, {}),
|
||
runner=_failing_runner)
|
||
worker.run_one()
|
||
details = _routing_details(self.db_path)
|
||
self.assertTrue([d for d in details if "source=explicit" in d])
|
||
self._assert_chosen_last(
|
||
[d for d in details if "source=explicit" in d])
|
||
|
||
|
||
class M5EscalateRetryTestCase(_WireCase):
|
||
"""M5：升级结果写回有界重试一次（仍带 failed 守卫）。"""
|
||
|
||
def _escalating_worker(self):
|
||
_insert_task(self.db_path, "TE", status="failed", audit_loop=1)
|
||
_seed_state(self.db_path, "TE", 3, "pushback")
|
||
return self._worker()
|
||
|
||
def test_escalate_result_write_retried_once(self):
|
||
# 第一次写失败、第二次成功：接线仍返回 escalate，
|
||
# 写回恰好调两次（重试生效，且不是无界重试）。
|
||
worker = self._escalating_worker()
|
||
calls = []
|
||
real = worker._write_execution_result
|
||
|
||
def flaky(*a, **k):
|
||
calls.append(1)
|
||
if len(calls) == 1:
|
||
raise RuntimeError("transient write failure")
|
||
return real(*a, **k)
|
||
|
||
worker._write_execution_result = flaky
|
||
task = worker.get_task("TE")
|
||
final, _, decision = worker._audit_loop_wire(
|
||
task, "failed", {"stdout": ""}, "boom")
|
||
self.assertEqual(decision, "escalate")
|
||
self.assertEqual(final, "failed")
|
||
self.assertEqual(len(calls), 2)
|
||
|
||
def test_escalate_result_write_bounded_on_repeated_failure(self):
|
||
# 两次都失败：仍返回 escalate（不谎报成 error），
|
||
# 写回恰好调两次，不会无界重试。
|
||
worker = self._escalating_worker()
|
||
worker._write_execution_result = mock.Mock(
|
||
side_effect=RuntimeError("persistent write failure"))
|
||
task = worker.get_task("TE")
|
||
final, _, decision = worker._audit_loop_wire(
|
||
task, "failed", {"stdout": ""}, "boom")
|
||
self.assertEqual(decision, "escalate")
|
||
self.assertEqual(final, "failed")
|
||
self.assertEqual(worker._write_execution_result.call_count, 2)
|
||
|
||
|
||
class M6ClearPacketsEarlyFailureTestCase(_WireCase):
|
||
"""M6：清包前即失败时包列原样不动（旧测试只在清干净后注错）。"""
|
||
|
||
def test_clear_packets_failure_before_clear_keeps_packets(self):
|
||
_insert_task(self.db_path, "PP", status="failed", audit_loop=1,
|
||
packet='{"round": 1}')
|
||
_insert_task(self.db_path, "CC", status="failed", audit_loop=0,
|
||
packet='{"round": 1}')
|
||
worker = self._worker()
|
||
|
||
def _boom(path):
|
||
raise RuntimeError("injected connect failure")
|
||
|
||
with mock.patch.object(worker_module, "_db_connect", _boom):
|
||
# 不抛错，且包列原样不动——失败不改已定结局。
|
||
worker._coord_clear_packets(
|
||
{"id": "PP", "project_id": "p1"}, ["CC"])
|
||
self.assertEqual(_task_row(self.db_path, "PP")[1],
|
||
'{"round": 1}')
|
||
self.assertEqual(_task_row(self.db_path, "CC")[1],
|
||
'{"round": 1}')
|