coverage85-step4a: worker run_one/_run_attempt 缺口测试（+34 项）
This commit is contained in:
host committed
1 parent
4cd9b5eaa4
commit
0e665d616f
1 file changed
+566
@@ -0,0 +1,566 @@
|
||||
"""覆盖率 78%->85% 第 4 步：cloud/worker.py run_one/_run_attempt
|
||||
剩余缺口的真实行为测试。
|
||||
|
||||
只用临时 SQLite（server.init_db 建表）与同步假执行器：
|
||||
假执行器是执行边界（真实 Local/Docker 执行器的替身），
|
||||
绝不碰真实 sub2api/生产网关、不跑模型；超时与重试成本
|
||||
用 max_retries=0、retry_backoff_seconds=0 控制。
|
||||
"""
|
||||
|
||||
import os
|
||||
import sqlite3
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest import mock
|
||||
|
||||
from cloud.sandbox import DockerExecutor
|
||||
from cloud.server import init_db
|
||||
from cloud.worker import Worker
|
||||
|
||||
|
||||
def _insert_task(db_path, task_id, status="queued", **kw):
|
||||
conn = sqlite3.connect(db_path)
|
||||
try:
|
||||
cols = {"id": task_id, "project_id": kw.pop("project_id", "proj"),
|
||||
"prompt": kw.pop("prompt", "do it"), "status": status,
|
||||
"created_by": kw.pop("created_by", "owner1"),
|
||||
"created_at": kw.pop("created_at", 1000.0),
|
||||
"updated_at": kw.pop("updated_at", 1000.0)}
|
||||
cols.update(kw)
|
||||
names = ", ".join(cols)
|
||||
placeholders = ", ".join("?" * len(cols))
|
||||
conn.execute("INSERT INTO tasks(%s) VALUES(%s)"
|
||||
% (names, placeholders), tuple(cols.values()))
|
||||
conn.commit()
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def _row(db_path, sql, params=()):
|
||||
conn = sqlite3.connect(db_path)
|
||||
try:
|
||||
return conn.execute(sql, params).fetchall()
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def _set(db_path, sql, params=()):
|
||||
conn = sqlite3.connect(db_path)
|
||||
try:
|
||||
conn.execute(sql, params)
|
||||
conn.commit()
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def _result(status="completed", stdout="done", **kw):
|
||||
r = {"executor": "local", "status": status, "stdout": stdout,
|
||||
"stderr": "", "returncode": 0, "host_status": None}
|
||||
r.update(kw)
|
||||
return r
|
||||
|
||||
|
||||
class FakeSync:
|
||||
"""同步假执行器：非 Local/Docker/Cloud，走 _run_attempt 的
|
||||
else 分支（dispatch_task(task) / goal_task(task) 单参形）。
|
||||
"""
|
||||
|
||||
def __init__(self, dispatch=None, run=None, goal=None):
|
||||
self.dispatch_result = dispatch or _result()
|
||||
self.run_result = run or _result()
|
||||
self.goal_result = goal or _result()
|
||||
self.calls = []
|
||||
|
||||
def dispatch_task(self, task):
|
||||
self.calls.append(("dispatch", task["id"]))
|
||||
return dict(self.dispatch_result)
|
||||
|
||||
def run_task(self, task):
|
||||
self.calls.append(("run", task["id"]))
|
||||
return dict(self.run_result)
|
||||
|
||||
def goal_task(self, task):
|
||||
self.calls.append(("goal", task["id"]))
|
||||
return dict(self.goal_result)
|
||||
|
||||
|
||||
class Worker3Base(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.tmp.cleanup)
|
||||
self.db_path = os.path.join(self.tmp.name, "control.db")
|
||||
init_db(self.db_path)
|
||||
self.hb = os.path.join(self.tmp.name, "hb.json")
|
||||
|
||||
def _worker(self, executor=None, **kw):
|
||||
kw.setdefault("env", {"XCM_API_KEY": "test-placeholder"})
|
||||
kw.setdefault("heartbeat_file", self.hb)
|
||||
kw.setdefault("worker_id", "cov85-worker3")
|
||||
kw.setdefault("max_retries", 0)
|
||||
kw.setdefault("retry_backoff_seconds", 0)
|
||||
return Worker(self.db_path, "/fake/host.py",
|
||||
executor=executor, **kw)
|
||||
|
||||
|
||||
class RunOneGatesTest(Worker3Base):
|
||||
def test_missing_key_gate(self):
|
||||
_insert_task(self.db_path, "t-key", status="queued")
|
||||
fake = FakeSync()
|
||||
w = self._worker(executor=fake, env={})
|
||||
result = w.run_one()
|
||||
self.assertTrue(result["dispatched"])
|
||||
self.assertEqual(result["status"], "failed")
|
||||
self.assertIn("Missing environment variable", result["error"])
|
||||
self.assertEqual(_row(self.db_path,
|
||||
"SELECT status FROM tasks WHERE id = 't-key'"),
|
||||
[("failed",)])
|
||||
self.assertEqual(fake.calls, [])
|
||||
|
||||
def test_already_claimed(self):
|
||||
_insert_task(self.db_path, "t-claim", status="running")
|
||||
w = self._worker(executor=FakeSync())
|
||||
task = w.get_task("t-claim")
|
||||
result = w.run_one(task)
|
||||
self.assertFalse(result["dispatched"])
|
||||
self.assertEqual(result["reason"], "already_claimed")
|
||||
|
||||
def test_quota_gate(self):
|
||||
_insert_task(self.db_path, "t-quota", status="queued")
|
||||
_set(self.db_path,
|
||||
"INSERT INTO user_quotas(user_id, quota_usd_micros,"
|
||||
" updated_at) VALUES('owner1', 0, 0)")
|
||||
w = self._worker(executor=FakeSync())
|
||||
result = w.run_one()
|
||||
self.assertTrue(result["dispatched"])
|
||||
self.assertEqual(result["status"], "interrupted")
|
||||
self.assertIn("quota_exceeded", result["error"])
|
||||
self.assertEqual(_row(
|
||||
self.db_path,
|
||||
"SELECT interrupted_reason FROM tasks WHERE id = 't-quota'"),
|
||||
[("quota",)])
|
||||
|
||||
def test_goal_cap_over_and_zero_old_form(self):
|
||||
for rounds, want in [(101, "超过上限"), (0, "已禁用")]:
|
||||
with self.subTest(rounds=rounds):
|
||||
tid = "t-cap-%s" % rounds
|
||||
_insert_task(self.db_path, tid, status="queued",
|
||||
verify_command="make test",
|
||||
goal_max_rounds=rounds)
|
||||
w = self._worker(executor=FakeSync())
|
||||
result = w.run_one()
|
||||
self.assertEqual(result["status"], "failed")
|
||||
self.assertIn(want, result["error"])
|
||||
self.assertEqual(_row(
|
||||
self.db_path,
|
||||
"SELECT status FROM tasks WHERE id = ?", (tid,)),
|
||||
[("failed",)])
|
||||
|
||||
def test_goal_cap_bad_type_passes_gate(self):
|
||||
# goal_max_rounds 非整数：int() 抛错后 _goal_rounds_int
|
||||
# 为 None，门不拦，照常跑完（门只拦可解析的超限值）。
|
||||
_insert_task(self.db_path, "t-cap-bad", status="queued",
|
||||
verify_command="make test", goal_max_rounds="abc")
|
||||
w = self._worker(executor=FakeSync())
|
||||
result = w.run_one()
|
||||
self.assertEqual(result["status"], "completed")
|
||||
|
||||
def test_gateway_failure_gate(self):
|
||||
_insert_task(self.db_path, "t-gw", status="queued")
|
||||
w = self._worker(executor=FakeSync())
|
||||
with mock.patch.dict(os.environ, {"CLOUD_GATEWAY": "sub2api",
|
||||
"SUB2API_BASE_URL": ""}):
|
||||
result = w.run_one()
|
||||
self.assertEqual(result["status"], "failed")
|
||||
self.assertIn("网关准备失败", result["error"])
|
||||
|
||||
def test_plain_resume_message_swaps_prompt(self):
|
||||
_insert_task(self.db_path, "t-resume", status="queued",
|
||||
resume_message="继续说下去")
|
||||
fake = FakeSync()
|
||||
w = self._worker(executor=fake)
|
||||
result = w.run_one()
|
||||
self.assertEqual(result["status"], "completed")
|
||||
self.assertEqual(fake.calls, [("dispatch", "t-resume")])
|
||||
# 续跑消息已消费：库里 resume_message 清掉。
|
||||
self.assertEqual(_row(
|
||||
self.db_path,
|
||||
"SELECT resume_message FROM tasks WHERE id = 't-resume'"),
|
||||
[(None,)])
|
||||
|
||||
def test_verify_resume_message_is_breakpoint_kind(self):
|
||||
# 旧验收形态 + 续跑消息：prompt 不换、_resume_kind 为
|
||||
# breakpoint，假执行器走 goal_task 一次即完成。
|
||||
_insert_task(self.db_path, "t-resume-v", status="queued",
|
||||
verify_command="make test",
|
||||
resume_message="继续原目标")
|
||||
fake = FakeSync()
|
||||
w = self._worker(executor=fake)
|
||||
result = w.run_one()
|
||||
self.assertEqual(result["status"], "completed")
|
||||
self.assertEqual(fake.calls, [("goal", "t-resume-v")])
|
||||
|
||||
def test_goal_resume_pending_fresh_restart(self):
|
||||
_insert_task(self.db_path, "t-grp", status="queued",
|
||||
goal_mode=1, goal_resume_pending=1)
|
||||
fake = FakeSync()
|
||||
w = self._worker(executor=fake)
|
||||
result = w.run_one()
|
||||
self.assertEqual(result["status"], "completed")
|
||||
self.assertEqual(fake.calls, [("goal", "t-grp")])
|
||||
self.assertEqual(_row(
|
||||
self.db_path,
|
||||
"SELECT goal_resume_pending FROM tasks WHERE id = 't-grp'"),
|
||||
[(0,)])
|
||||
|
||||
|
||||
class RunOneAttemptFlowTest(Worker3Base):
|
||||
def test_completed_simple(self):
|
||||
_insert_task(self.db_path, "t-ok", status="queued")
|
||||
w = self._worker(executor=FakeSync())
|
||||
result = w.run_one()
|
||||
self.assertEqual(result["status"], "completed")
|
||||
self.assertEqual(result["stdout"], "done")
|
||||
self.assertIsNotNone(result["cost_record_id"])
|
||||
|
||||
def test_failed_simple_error_text(self):
|
||||
_insert_task(self.db_path, "t-fail", status="queued")
|
||||
fake = FakeSync(dispatch=_result(status="failed", stdout="",
|
||||
stderr="boom-err"))
|
||||
w = self._worker(executor=fake)
|
||||
result = w.run_one()
|
||||
self.assertEqual(result["status"], "failed")
|
||||
self.assertIn("boom-err", result["error"])
|
||||
|
||||
def test_timeout_converts_to_failed(self):
|
||||
_insert_task(self.db_path, "t-timeout", status="queued")
|
||||
fake = FakeSync(dispatch=_result(status="timeout", stdout=""))
|
||||
w = self._worker(executor=fake)
|
||||
result = w.run_one()
|
||||
self.assertEqual(result["status"], "failed")
|
||||
self.assertIn("task timeout", result["error"])
|
||||
|
||||
def test_retry_same_model(self):
|
||||
_insert_task(self.db_path, "t-retry", status="queued")
|
||||
fake = FakeSync(dispatch=_result(status="failed", stdout="",
|
||||
stderr="first-fail"))
|
||||
w = self._worker(executor=fake, max_retries=1,
|
||||
retry_backoff_seconds=0)
|
||||
result = w.run_one()
|
||||
self.assertEqual(result["status"], "failed")
|
||||
self.assertIn("[retries=1]", result["error"])
|
||||
self.assertEqual([c[0] for c in fake.calls],
|
||||
["dispatch", "dispatch"])
|
||||
|
||||
def test_fallback_model_second_failure_combines_errors(self):
|
||||
_insert_task(self.db_path, "t-fb", status="queued")
|
||||
calls = {"n": 0}
|
||||
|
||||
class _Fb(FakeSync):
|
||||
def dispatch_task(self, task):
|
||||
calls["n"] += 1
|
||||
if calls["n"] == 1:
|
||||
return _result(status="failed", stdout="",
|
||||
stderr="primary-bad 429",
|
||||
host_status="failed")
|
||||
return _result(status="failed", stdout="",
|
||||
stderr="fallback-bad",
|
||||
host_status="failed")
|
||||
fake = _Fb()
|
||||
w = self._worker(executor=fake, fallback_models=["fb-model"])
|
||||
result = w.run_one()
|
||||
self.assertEqual(result["status"], "failed")
|
||||
# 真 fallback 换模型再跑一次：两次原因都写进错误文本。
|
||||
self.assertEqual(calls["n"], 2)
|
||||
self.assertIn("primary model", result["error"])
|
||||
self.assertIn("fallback model", result["error"])
|
||||
self.assertIn("fallback-bad", result["error"])
|
||||
|
||||
def test_fallback_via_token(self):
|
||||
_insert_task(self.db_path, "t-fb2", status="queued")
|
||||
seq = [("failed", "model overloaded 429"),
|
||||
("failed", "second bad")]
|
||||
|
||||
class _Fb2(FakeSync):
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.n = 0
|
||||
|
||||
def dispatch_task(self, task):
|
||||
self.n += 1
|
||||
status, err = seq[min(self.n - 1, len(seq) - 1)]
|
||||
return _result(status=status, stdout="",
|
||||
stderr=err, host_status="failed")
|
||||
fake = _Fb2()
|
||||
w = self._worker(executor=fake, fallback_models=["fb-model"])
|
||||
result = w.run_one()
|
||||
self.assertEqual(result["status"], "failed")
|
||||
# 真 fallback 走过时错误文本含两次原因（或至少第二次）。
|
||||
self.assertEqual(fake.n, 2)
|
||||
self.assertIn("fallback model", result["error"])
|
||||
|
||||
def test_cancelled_via_result(self):
|
||||
_insert_task(self.db_path, "t-cancel", status="queued")
|
||||
fake = FakeSync(dispatch=_result(status="cancelled", stdout=""))
|
||||
w = self._worker(executor=fake)
|
||||
result = w.run_one()
|
||||
self.assertEqual(result["status"], "interrupted")
|
||||
self.assertEqual(result["error"], "cancelled")
|
||||
|
||||
def test_goal_already_exists_falls_back_to_resume(self):
|
||||
# goal 首轮 stdout 含“已存在”：run_one 走一次断点续跑，
|
||||
# 续跑完成即最终完成（不直接 failed）。
|
||||
_insert_task(self.db_path, "t-goal-exists", status="queued",
|
||||
goal_mode=1)
|
||||
calls = []
|
||||
|
||||
class _GoalEx(FakeSync):
|
||||
def goal_task(self, task):
|
||||
calls.append(task["id"])
|
||||
if len(calls) == 1:
|
||||
return _result(status="failed", stdout="任务已存在")
|
||||
return _result(status="completed", stdout="resumed-ok")
|
||||
w = self._worker(executor=_GoalEx())
|
||||
result = w.run_one()
|
||||
self.assertEqual(result["status"], "completed")
|
||||
self.assertEqual(result["stdout"], "resumed-ok")
|
||||
self.assertEqual(len(calls), 2)
|
||||
|
||||
def test_goal_met_false_notifies_goal_unmet(self):
|
||||
_insert_task(self.db_path, "t-goal-unmet", status="queued",
|
||||
goal_mode=1)
|
||||
fake = FakeSync(goal=_result(status="completed", stdout="",
|
||||
goal_met=False))
|
||||
w = self._worker(executor=fake)
|
||||
result = w.run_one()
|
||||
self.assertEqual(result["status"], "completed")
|
||||
actions = _row(self.db_path,
|
||||
"SELECT action FROM audit_log ORDER BY id")
|
||||
self.assertIn(("worker_completed",), actions)
|
||||
|
||||
def test_followup_goal_convert_queues_task(self):
|
||||
_insert_task(self.db_path, "t-fu", status="queued")
|
||||
_set(self.db_path,
|
||||
"INSERT INTO task_followups(id, task_id, project_id, body,"
|
||||
" status, created_at) VALUES('fu1', 't-fu', 'proj',"
|
||||
" '/goal 收齐周报', 'pending', 1000)")
|
||||
w = self._worker(executor=FakeSync())
|
||||
result = w.run_one()
|
||||
self.assertEqual(result["status"], "queued")
|
||||
self.assertEqual(_row(
|
||||
self.db_path,
|
||||
"SELECT status FROM tasks WHERE id = 't-fu'"), [("queued",)])
|
||||
self.assertEqual(_row(
|
||||
self.db_path,
|
||||
"SELECT status FROM task_followups WHERE id = 'fu1'"),
|
||||
[("done",)])
|
||||
|
||||
def test_followup_plain_message_runs_second_round(self):
|
||||
_insert_task(self.db_path, "t-fu2", status="queued")
|
||||
_set(self.db_path,
|
||||
"INSERT INTO task_followups(id, task_id, project_id, body,"
|
||||
" status, created_at) VALUES('fu2', 't-fu2', 'proj',"
|
||||
" '再补充一点', 'pending', 1000)")
|
||||
fake = FakeSync()
|
||||
w = self._worker(executor=fake)
|
||||
result = w.run_one()
|
||||
self.assertEqual(result["status"], "completed")
|
||||
self.assertEqual(len(fake.calls), 2)
|
||||
self.assertEqual(_row(
|
||||
self.db_path,
|
||||
"SELECT status FROM task_followups WHERE id = 'fu2'"),
|
||||
[("done",)])
|
||||
|
||||
def test_followup_failure_stops_drain(self):
|
||||
_insert_task(self.db_path, "t-fu3", status="queued")
|
||||
_set(self.db_path,
|
||||
"INSERT INTO task_followups(id, task_id, project_id, body,"
|
||||
" status, created_at) VALUES('fu3', 't-fu3', 'proj',"
|
||||
" '再补充一点', 'pending', 1000)")
|
||||
|
||||
class _FuFail(FakeSync):
|
||||
def dispatch_task(self, task):
|
||||
self.calls.append(("dispatch", task["id"]))
|
||||
if len(self.calls) == 1:
|
||||
return _result()
|
||||
return _result(status="failed", stdout="",
|
||||
stderr="fu-bad")
|
||||
w = self._worker(executor=_FuFail())
|
||||
result = w.run_one()
|
||||
self.assertEqual(result["status"], "failed")
|
||||
self.assertIn("fu-bad", result["error"])
|
||||
self.assertEqual(_row(
|
||||
self.db_path,
|
||||
"SELECT status FROM task_followups WHERE id = 'fu3'"),
|
||||
[("failed",)])
|
||||
|
||||
def test_quota_stop_after_attempt(self):
|
||||
# 首轮返回 quota_exceeded：run_one 走配额中断收尾，
|
||||
# 状态 interrupted、续跑消息保留（本例无消息，只断状态）。
|
||||
_insert_task(self.db_path, "t-qstop", status="queued")
|
||||
fake = FakeSync(dispatch=_result(status="quota_exceeded",
|
||||
stdout=""))
|
||||
w = self._worker(executor=fake)
|
||||
result = w.run_one()
|
||||
self.assertEqual(result["status"], "interrupted")
|
||||
self.assertIn("quota", result["error"])
|
||||
self.assertEqual(_row(
|
||||
self.db_path,
|
||||
"SELECT interrupted_reason FROM tasks WHERE id = 't-qstop'"),
|
||||
[("quota",)])
|
||||
|
||||
|
||||
class RunAttemptDirectTest(Worker3Base):
|
||||
def _task_dict(self, tid, **kw):
|
||||
d = {"id": tid, "project_id": "proj", "prompt": "do it",
|
||||
"created_by": "owner1"}
|
||||
d.update(kw)
|
||||
return d
|
||||
|
||||
def test_dispatch_already_exists_runs_again(self):
|
||||
w = self._worker(executor=FakeSync(
|
||||
dispatch=_result(status="failed", stdout="任务已存在"),
|
||||
run=_result(stdout="run-ok")))
|
||||
result, error = w._run_attempt(self._task_dict("t-ae"))
|
||||
self.assertEqual(result["status"], "completed")
|
||||
self.assertEqual(result["stdout"], "run-ok")
|
||||
self.assertIsNone(error)
|
||||
|
||||
def test_executor_raises_becomes_failed(self):
|
||||
class _Boom(FakeSync):
|
||||
def dispatch_task(self, task):
|
||||
raise RuntimeError("runner 炸了")
|
||||
w = self._worker(executor=_Boom())
|
||||
result, error = w._run_attempt(self._task_dict("t-boom"))
|
||||
self.assertEqual(result["status"], "failed")
|
||||
self.assertEqual(error, "runner 炸了")
|
||||
|
||||
def test_failed_stdout_tail_and_docker_timeout_marker(self):
|
||||
w = self._worker(executor=FakeSync(
|
||||
dispatch=_result(status="failed", stdout="docker run timeout")))
|
||||
result, _ = w._run_attempt(self._task_dict("t-dto"))
|
||||
self.assertEqual(result["status"], "timeout")
|
||||
|
||||
def test_quota_hit_rewrites_to_quota_exceeded(self):
|
||||
w = self._worker(executor=FakeSync())
|
||||
w._mark_quota_hit("t-qh")
|
||||
result, error = w._run_attempt(self._task_dict("t-qh"))
|
||||
self.assertEqual(result["status"], "quota_exceeded")
|
||||
self.assertIn("配额看门狗", error)
|
||||
|
||||
def test_goal_breakpoint_unsupported_for_fake_executor(self):
|
||||
w = self._worker(executor=FakeSync())
|
||||
task = self._task_dict("t-bp", goal_mode=1)
|
||||
task["_goal_resume_breakpoint"] = True
|
||||
result, error = w._run_attempt(task)
|
||||
self.assertEqual(result["status"], "failed")
|
||||
self.assertIn("不支持 goal 断点续跑", error)
|
||||
|
||||
def test_goal_task_via_fake_executor(self):
|
||||
w = self._worker(executor=FakeSync(
|
||||
goal=_result(stdout="goal-done")))
|
||||
result, _ = w._run_attempt(self._task_dict("t-goal",
|
||||
goal_mode=1))
|
||||
self.assertEqual(result["status"], "completed")
|
||||
self.assertEqual(result["stdout"], "goal-done")
|
||||
|
||||
def test_no_time_limit_task(self):
|
||||
w = self._worker(executor=FakeSync())
|
||||
result, _ = w._run_attempt(self._task_dict(
|
||||
"t-ntl", no_time_limit=1))
|
||||
self.assertEqual(result["status"], "completed")
|
||||
|
||||
|
||||
class DockerAttemptTest(Worker3Base):
|
||||
def _docker_worker(self, **kw):
|
||||
class _Docker(DockerExecutor):
|
||||
def __init__(self):
|
||||
super().__init__(runner=lambda cmd, timeout=None: None,
|
||||
env={"XCM_API_KEY": "test"})
|
||||
self.timeout = kw.pop("timeout", None)
|
||||
|
||||
def container_name(self, task):
|
||||
return "codex-t-" + task["id"]
|
||||
|
||||
def goal_task(self, task, **_kw):
|
||||
return _result(stdout="docker-goal-ok")
|
||||
|
||||
def goal_resume_task(self, task, text, **_kw):
|
||||
return _result(stdout="docker-resume-ok")
|
||||
|
||||
def goal_resume_breakpoint_task(self, task, **_kw):
|
||||
return _result(stdout="docker-bp-ok")
|
||||
|
||||
def dispatch_task(self, task, **_kw):
|
||||
return _result(stdout="docker-dispatch-ok")
|
||||
|
||||
def run_task(self, task, **_kw):
|
||||
return _result(stdout="docker-run-ok")
|
||||
|
||||
def ensure_network(self, name):
|
||||
return True
|
||||
|
||||
def network_gateway(self, network):
|
||||
return "10.0.0.1"
|
||||
w = self._worker(executor=_Docker(), **kw)
|
||||
return w
|
||||
|
||||
def _task_dict(self, tid, **kw):
|
||||
d = {"id": tid, "project_id": "proj", "prompt": "do it",
|
||||
"created_by": "owner1"}
|
||||
d.update(kw)
|
||||
return d
|
||||
|
||||
def test_docker_goal_and_resume_variants(self):
|
||||
w = self._docker_worker()
|
||||
result, _ = w._run_attempt(self._task_dict("t-dg",
|
||||
goal_mode=1))
|
||||
self.assertEqual(result["stdout"], "docker-goal-ok")
|
||||
task = self._task_dict("t-dr", goal_mode=1)
|
||||
task["_resume"] = True
|
||||
task["_resume_kind"] = "new-goal"
|
||||
result, _ = w._run_attempt(task)
|
||||
self.assertEqual(result["stdout"], "docker-resume-ok")
|
||||
task = self._task_dict("t-dbp", goal_mode=1)
|
||||
task["_goal_resume_breakpoint"] = True
|
||||
result, _ = w._run_attempt(task)
|
||||
self.assertEqual(result["stdout"], "docker-bp-ok")
|
||||
result, _ = w._run_attempt(self._task_dict("t-dd"))
|
||||
self.assertEqual(result["stdout"], "docker-dispatch-ok")
|
||||
|
||||
def test_docker_context_egress_gateway_failure(self):
|
||||
class _NoGw(DockerExecutor):
|
||||
def __init__(self):
|
||||
super().__init__(runner=lambda cmd, timeout=None: None,
|
||||
env={"XCM_API_KEY": "test"})
|
||||
|
||||
def network_gateway(self, network):
|
||||
return None
|
||||
w = self._worker(executor=_NoGw(), egress_proxy_port=8888)
|
||||
result, error = w._run_attempt(self._task_dict("t-dgw"))
|
||||
self.assertEqual(result["status"], "failed")
|
||||
self.assertIn("egress 代理网关解析失败", error)
|
||||
|
||||
def test_docker_timeout_bad_type_falls_back(self):
|
||||
# executor.timeout 非数字：min() 抛错后回落 task_timeout，
|
||||
# 任务仍正常跑完（不炸 _run_attempt）。
|
||||
w = self._docker_worker(timeout="bogus")
|
||||
result, _ = w._run_attempt(self._task_dict("t-dbt"))
|
||||
self.assertEqual(result["status"], "completed")
|
||||
|
||||
|
||||
class RunLoopPrJobsTest(Worker3Base):
|
||||
def test_run_loop_zero_and_run_all_empty(self):
|
||||
w = self._worker(executor=FakeSync())
|
||||
self.assertEqual(w.run_loop(max_seconds=0), [])
|
||||
self.assertEqual(w.run_all(), [])
|
||||
|
||||
def test_run_one_with_explicit_task_dict(self):
|
||||
_insert_task(self.db_path, "t-exp", status="queued")
|
||||
w = self._worker(executor=FakeSync())
|
||||
task = w.get_task("t-exp")
|
||||
result = w.run_one(task)
|
||||
self.assertEqual(result["status"], "completed")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user