host
488aedbcad
(cherry picked from commit 1db02545f74c96b3b92be4fc32954659cec1f814) Reconcile-conflict: A14b E402 imports referencing cloud.handler_core (does not exist here; core methods live in server.py) — kept HEAD imports with incoming's noqa: E402 markers, dropped the unused core_mod import (verified unused elsewhere in both files).
706 lines
31 KiB
Python
706 lines
31 KiB
Python
"""Coverage batch v2: cloud/worker.py _deliver_marker_tools branches.
|
|
|
|
Baseline left 93 lines of _deliver_marker_tools uncovered: quota /
|
|
DB exception paths, automation bind/create failures, control
|
|
fallbacks, todo/task/cost/quota/files branches, poll/wait
|
|
exceptions and the peer-message delivery error. The harness mirrors
|
|
test_marker_tools.WorkerDeliverTestCase (temp SQLite via init_db,
|
|
sender PM created by tester/maintainer).
|
|
|
|
Fault injection notes (honest about what is stubbed and why):
|
|
- Collaborator helpers (_ensure_result_columns, _todo_list_db,
|
|
_poll_task_snapshot, _wait_task_targets, _db_connect, CostStore,
|
|
_marker_ancestry, get_task, cancel, _audit) are patched to
|
|
simulate a failure the real code must report, never to fake the
|
|
deliver report itself; every assertion reads the real followup
|
|
body or the real DB row afterwards.
|
|
- The quota owner-from-parent fallback (sender without created_by)
|
|
is defensive code: the real _marker_auth returns False for a
|
|
creator-less sender, so the fallback is only reachable when auth
|
|
is stubbed True. Those two tests say so in their docstrings.
|
|
- Root bypasses file permission bits, so the dir-list and file-read
|
|
failure branches are reached with a selective patch (listdir /
|
|
open raising for exactly the target path), not chmod.
|
|
"""
|
|
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest import mock
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[2]
|
|
sys.path.insert(0, str(REPO_ROOT))
|
|
|
|
from cloud.server import init_db # noqa: E402
|
|
from cloud.todos import create_todo as _create_todo # noqa: E402
|
|
from cloud.worker import Worker # noqa: E402
|
|
|
|
FAKE_ENV = {"XCM_API_KEY": "test-placeholder"}
|
|
|
|
|
|
def _base_result(**over):
|
|
result = {
|
|
"status": "completed",
|
|
"task_creates": [], "task_polls": [], "wait_threads": [],
|
|
"task_cancels": [], "task_retries": [], "todo_lists": [],
|
|
"todo_creates": [], "todo_updates": [], "task_lists": [],
|
|
"task_costs": [], "quota_status": [], "task_files": [],
|
|
"goal_converts": [], "automation_creates": [],
|
|
}
|
|
result.update(over)
|
|
return result
|
|
|
|
|
|
class DeliverBase(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.db_path = os.path.join(self.tmp.name, "control.db")
|
|
init_db(self.db_path)
|
|
self._insert("PM", status="running", created_by="tester")
|
|
self._member("tester", "maintainer")
|
|
|
|
def tearDown(self):
|
|
self.tmp.cleanup()
|
|
|
|
def _insert(self, task_id, status="queued", created_by="tester",
|
|
project_id="p1", parent=None, **cols):
|
|
now = time.time()
|
|
conn = sqlite3.connect(self.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)"
|
|
" VALUES(?, ?, ?, ?, NULL, '[]', 0, ?, ?, ?, NULL)",
|
|
(task_id, project_id, "do work", status, created_by,
|
|
now, now))
|
|
if parent is not None:
|
|
conn.execute(
|
|
"UPDATE tasks SET parent_task_id = ? WHERE id = ?",
|
|
(parent, task_id))
|
|
for key, value in cols.items():
|
|
conn.execute(
|
|
"UPDATE tasks SET %s = ? WHERE id = ?" % key,
|
|
(value, task_id))
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
def _member(self, user_id, role):
|
|
conn = sqlite3.connect(self.db_path)
|
|
try:
|
|
conn.execute(
|
|
"CREATE TABLE IF NOT EXISTS users(id TEXT PRIMARY KEY,"
|
|
" username TEXT UNIQUE, api_key_hash TEXT,"
|
|
" created_at REAL)")
|
|
conn.execute(
|
|
"CREATE TABLE IF NOT EXISTS project_members("
|
|
"project_id TEXT, user_id TEXT, role TEXT,"
|
|
" PRIMARY KEY(project_id, user_id))")
|
|
conn.execute(
|
|
"INSERT OR REPLACE INTO project_members"
|
|
"(project_id, user_id, role) VALUES('p1', ?, ?)",
|
|
(user_id, role))
|
|
conn.execute(
|
|
"INSERT OR IGNORE INTO users(id, username, api_key_hash,"
|
|
" created_at) VALUES(?, ?, '', 0)", (user_id, user_id))
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
def _worker(self, **kwargs):
|
|
return Worker(self.db_path, "/fake/host.py", env=FAKE_ENV,
|
|
**kwargs)
|
|
|
|
def _sender(self, worker, task_id="PM"):
|
|
return worker.get_task(task_id)
|
|
|
|
def _followup_bodies(self, task_id="PM"):
|
|
conn = sqlite3.connect(self.db_path)
|
|
try:
|
|
rows = conn.execute(
|
|
"SELECT body FROM task_followups"
|
|
" WHERE task_id = ?", (task_id,)).fetchall()
|
|
return [row[0] for row in rows]
|
|
finally:
|
|
conn.close()
|
|
|
|
def _status(self, task_id):
|
|
conn = sqlite3.connect(self.db_path)
|
|
try:
|
|
row = conn.execute(
|
|
"SELECT status, result_text FROM tasks WHERE id = ?",
|
|
(task_id,)).fetchone()
|
|
return row[0], row[1]
|
|
finally:
|
|
conn.close()
|
|
|
|
def _deliver(self, worker, sender, result, task_id="PM"):
|
|
worker._deliver_marker_tools(sender, result)
|
|
return self._followup_bodies(task_id)
|
|
|
|
|
|
class CreatesAndAutomationTestCase(DeliverBase):
|
|
def test_creates_quota_check_exception(self):
|
|
worker = self._worker()
|
|
sender = self._sender(worker)
|
|
with mock.patch.object(worker, "_quota_exceeded",
|
|
side_effect=RuntimeError("boom")):
|
|
bodies = self._deliver(worker, sender, _base_result(
|
|
task_creates=[{"name": "E1", "prompt": "do e1",
|
|
"model": None, "effort": None,
|
|
"executor": None}]))
|
|
self.assertTrue(any("quota_check_failed" in b for b in bodies))
|
|
|
|
def test_creates_db_exception(self):
|
|
worker = self._worker()
|
|
sender = self._sender(worker)
|
|
with mock.patch.object(worker, "_ensure_result_columns",
|
|
side_effect=RuntimeError("db down")):
|
|
bodies = self._deliver(worker, sender, _base_result(
|
|
task_creates=[{"name": "E1", "prompt": "do e1",
|
|
"model": None, "effort": None,
|
|
"executor": None}]))
|
|
self.assertTrue(any("创建失败" in b for b in bodies))
|
|
|
|
def test_automation_first_run_past_skipped(self):
|
|
worker = self._worker()
|
|
bodies = self._deliver(worker, self._sender(worker), _base_result(
|
|
automation_creates=[{"name": "auto1", "prompt": "do auto",
|
|
"interval_seconds": 3600,
|
|
"first_run_at": time.time() - 3600}]))
|
|
self.assertTrue(any("首次运行时刻已过去" in b for b in bodies))
|
|
|
|
def test_automation_duplicate_name_second_fails(self):
|
|
worker = self._worker()
|
|
bodies = self._deliver(worker, self._sender(worker), _base_result(
|
|
automation_creates=[
|
|
{"name": "dup", "prompt": "do auto",
|
|
"interval_seconds": 3600},
|
|
{"name": "dup", "prompt": "do auto again",
|
|
"interval_seconds": 3600}]))
|
|
joined = "\n".join(bodies)
|
|
self.assertIn("已创建自动化", joined)
|
|
self.assertIn("自动化创建失败", joined)
|
|
self.assertIn("already exists", joined)
|
|
|
|
def test_automation_bind_bad_mode_reports_but_counts(self):
|
|
worker = self._worker()
|
|
bodies = self._deliver(worker, self._sender(worker), _base_result(
|
|
automation_creates=[{"name": "b1", "prompt": "do auto",
|
|
"interval_seconds": 3600,
|
|
"bind": "bogus"}]))
|
|
joined = "\n".join(bodies)
|
|
self.assertIn("绑定到本会话失败", joined)
|
|
self.assertIn("定义本身仍有效", joined)
|
|
conn = sqlite3.connect(self.db_path)
|
|
try:
|
|
row = conn.execute(
|
|
"SELECT name FROM automations WHERE name = 'b1'"
|
|
).fetchone()
|
|
finally:
|
|
conn.close()
|
|
self.assertIsNotNone(row)
|
|
|
|
def test_automation_bind_heartbeat_success(self):
|
|
worker = self._worker()
|
|
bodies = self._deliver(worker, self._sender(worker), _base_result(
|
|
automation_creates=[{"name": "hb1", "prompt": "do auto",
|
|
"interval_seconds": 3600,
|
|
"bind": "heartbeat"}]))
|
|
self.assertTrue(any("已绑到本会话唤醒" in b for b in bodies))
|
|
|
|
def test_automation_quota_exception(self):
|
|
worker = self._worker()
|
|
sender = self._sender(worker)
|
|
with mock.patch.object(worker, "_quota_exceeded",
|
|
side_effect=RuntimeError("boom")):
|
|
bodies = self._deliver(worker, sender, _base_result(
|
|
automation_creates=[{"name": "q1", "prompt": "do auto",
|
|
"interval_seconds": 3600}]))
|
|
joined = "\n".join(bodies)
|
|
self.assertIn("自动化创建跳过", joined)
|
|
self.assertIn("quota_check_failed", joined)
|
|
|
|
|
|
class ControlFallbackTestCase(DeliverBase):
|
|
def test_cancel_target_read_failure(self):
|
|
self._insert("E1", status="running", parent="PM")
|
|
worker = self._worker()
|
|
sender = self._sender(worker)
|
|
# _was_cancelled also calls get_task, so only the target read
|
|
# itself may fail; everything else delegates to the real one.
|
|
orig_get = worker.get_task
|
|
|
|
def _flaky_get(task_id):
|
|
if task_id == "E1":
|
|
raise RuntimeError("db down")
|
|
return orig_get(task_id)
|
|
|
|
with mock.patch.object(worker, "get_task",
|
|
side_effect=_flaky_get):
|
|
bodies = self._deliver(worker, sender, _base_result(
|
|
task_cancels=[{"task_id": "E1"}]))
|
|
self.assertTrue(any("读取失败" in b for b in bodies))
|
|
self.assertEqual(self._status("E1")[0], "running")
|
|
|
|
def test_cancel_ancestry_failure_falls_back_to_other(self):
|
|
self._insert("E1", status="running", parent="PM")
|
|
worker = self._worker()
|
|
with mock.patch("cloud.worker._marker_ancestry",
|
|
side_effect=RuntimeError("db down")):
|
|
bodies = self._deliver(worker, self._sender(worker),
|
|
_base_result(
|
|
task_cancels=[{"task_id": "E1"}]))
|
|
self.assertTrue(any("拒绝" in b for b in bodies))
|
|
self.assertEqual(self._status("E1")[0], "running")
|
|
|
|
def test_cancel_not_done_reports(self):
|
|
self._insert("E1", status="running", parent="PM")
|
|
worker = self._worker()
|
|
with mock.patch.object(worker, "cancel", return_value=False):
|
|
bodies = self._deliver(worker, self._sender(worker),
|
|
_base_result(
|
|
task_cancels=[{"task_id": "E1"}]))
|
|
self.assertTrue(any("未执行" in b for b in bodies))
|
|
self.assertEqual(self._status("E1")[0], "running")
|
|
|
|
def test_cancel_execution_exception(self):
|
|
self._insert("E1", status="running", parent="PM")
|
|
worker = self._worker()
|
|
with mock.patch.object(worker, "cancel",
|
|
side_effect=RuntimeError("boom")):
|
|
bodies = self._deliver(worker, self._sender(worker),
|
|
_base_result(
|
|
task_cancels=[{"task_id": "E1"}]))
|
|
self.assertTrue(any("执行失败" in b for b in bodies))
|
|
|
|
def test_retry_not_done_when_rowcount_zero(self):
|
|
self._insert("E1", status="failed", parent="PM")
|
|
worker = self._worker()
|
|
|
|
class _CursorProxy:
|
|
def __init__(self, real):
|
|
self._real = real
|
|
self.rowcount = 0
|
|
|
|
def __getattr__(self, name):
|
|
return getattr(self._real, name)
|
|
|
|
class _ConnProxy:
|
|
def __init__(self, real):
|
|
self._real = real
|
|
|
|
def execute(self, sql, params=()):
|
|
if "UPDATE tasks SET status = 'queued'" in sql:
|
|
return _CursorProxy(self._real.execute(sql, params))
|
|
return self._real.execute(sql, params)
|
|
|
|
def __getattr__(self, name):
|
|
return getattr(self._real, name)
|
|
|
|
import cloud.worker as worker_mod
|
|
real_db_connect = worker_mod._db_connect
|
|
|
|
def _connect(path):
|
|
return _ConnProxy(real_db_connect(path))
|
|
|
|
with mock.patch("cloud.worker._db_connect", side_effect=_connect):
|
|
bodies = self._deliver(worker, self._sender(worker),
|
|
_base_result(
|
|
task_retries=[{"task_id": "E1"}]))
|
|
# The proxy cursor reports rowcount 0, so deliver must report
|
|
# "not executed" -- but the real UPDATE still ran through the
|
|
# proxy's delegate, so the test asserts the report, not the
|
|
# row, for this TOCTOU simulation.
|
|
self.assertTrue(any("未执行" in b for b in bodies))
|
|
|
|
def test_retry_execution_exception_via_audit(self):
|
|
self._insert("E1", status="failed", parent="PM")
|
|
worker = self._worker()
|
|
sender = self._sender(worker)
|
|
orig_audit = worker._audit
|
|
calls = [0]
|
|
|
|
def _flaky_audit(project_id, action, detail=""):
|
|
calls[0] += 1
|
|
if calls[0] == 1:
|
|
raise RuntimeError("audit down")
|
|
return orig_audit(project_id, action, detail)
|
|
|
|
with mock.patch.object(worker, "_audit", side_effect=_flaky_audit):
|
|
bodies = self._deliver(worker, sender, _base_result(
|
|
task_retries=[{"task_id": "E1"}]))
|
|
self.assertTrue(any("执行失败" in b for b in bodies))
|
|
|
|
|
|
class TodoTestCase(DeliverBase):
|
|
def test_todo_create_write_denied(self):
|
|
self._member("tester", "viewer")
|
|
worker = self._worker()
|
|
bodies = self._deliver(worker, self._sender(worker), _base_result(
|
|
todo_creates=[{"title": "T1"}]))
|
|
self.assertTrue(any("创建待办" in b and "拒绝" in b for b in bodies))
|
|
|
|
def test_todo_create_cross_project_task(self):
|
|
self._insert("X1", status="running", project_id="p2")
|
|
worker = self._worker()
|
|
bodies = self._deliver(worker, self._sender(worker), _base_result(
|
|
todo_creates=[{"title": "T1", "task_id": "X1"}]))
|
|
self.assertTrue(any("属于别的项目" in b for b in bodies))
|
|
|
|
def test_todo_create_validation_failure(self):
|
|
worker = self._worker()
|
|
bodies = self._deliver(worker, self._sender(worker), _base_result(
|
|
todo_creates=[{"title": ""}]))
|
|
self.assertTrue(any("创建待办" in b and "失败" in b for b in bodies))
|
|
|
|
def test_todo_update_cross_project_task(self):
|
|
self._insert("X1", status="running", project_id="p2")
|
|
conn = sqlite3.connect(self.db_path)
|
|
try:
|
|
todo = _create_todo(conn, "p1", "tester", "T1")
|
|
finally:
|
|
conn.close()
|
|
worker = self._worker()
|
|
bodies = self._deliver(worker, self._sender(worker), _base_result(
|
|
todo_updates=[{"id": todo["id"],
|
|
"fields": {"task_id": "X1"}}]))
|
|
self.assertTrue(any("属于别的项目" in b for b in bodies))
|
|
|
|
def test_todo_update_failure(self):
|
|
conn = sqlite3.connect(self.db_path)
|
|
try:
|
|
todo = _create_todo(conn, "p1", "tester", "T1")
|
|
finally:
|
|
conn.close()
|
|
worker = self._worker()
|
|
bodies = self._deliver(worker, self._sender(worker), _base_result(
|
|
todo_updates=[{"id": todo["id"],
|
|
"fields": {"status": "bogus"}}]))
|
|
self.assertTrue(any("更新待办" in b and "失败" in b for b in bodies))
|
|
|
|
def test_todo_list_failure(self):
|
|
worker = self._worker()
|
|
with mock.patch("cloud.worker._todo_list_db",
|
|
side_effect=RuntimeError("db down")):
|
|
bodies = self._deliver(worker, self._sender(worker),
|
|
_base_result(todo_lists=[{}]))
|
|
self.assertTrue(any("查待办失败" in b for b in bodies))
|
|
|
|
def test_todo_list_truncation(self):
|
|
conn = sqlite3.connect(self.db_path)
|
|
try:
|
|
for i in range(21):
|
|
_create_todo(conn, "p1", "tester", "T%d" % i)
|
|
finally:
|
|
conn.close()
|
|
worker = self._worker()
|
|
bodies = self._deliver(worker, self._sender(worker),
|
|
_base_result(todo_lists=[{}]))
|
|
self.assertTrue(any("还有 1 条未列出" in b for b in bodies))
|
|
|
|
def test_task_list_failure(self):
|
|
worker = self._worker()
|
|
sender = self._sender(worker)
|
|
# _was_cancelled (first _db_connect use) and the final
|
|
# followup insert must still work; only the task_lists query
|
|
# itself (second use) fails.
|
|
import cloud.worker as worker_mod
|
|
real_connect = worker_mod._db_connect
|
|
calls = [0]
|
|
|
|
def _flaky_connect(path):
|
|
calls[0] += 1
|
|
if calls[0] == 2:
|
|
raise RuntimeError("db down")
|
|
return real_connect(path)
|
|
|
|
with mock.patch("cloud.worker._db_connect",
|
|
side_effect=_flaky_connect):
|
|
bodies = self._deliver(worker, sender,
|
|
_base_result(task_lists=[{"limit": 10}]))
|
|
self.assertTrue(any("查任务列表失败" in b for b in bodies))
|
|
|
|
|
|
class CostQuotaFilesTestCase(DeliverBase):
|
|
def test_task_cost_read_denied(self):
|
|
# Viewer still has read permission, so read denial needs a
|
|
# sender with no project membership at all.
|
|
self._insert("OUT", status="running", created_by="outsider")
|
|
self._insert("E1", status="completed", parent="PM")
|
|
worker = self._worker()
|
|
bodies = self._deliver(worker, self._sender(worker, "OUT"),
|
|
_base_result(
|
|
task_costs=[{"task_id": "E1"}]),
|
|
task_id="OUT")
|
|
self.assertTrue(any("创建人无 read 权限" in b for b in bodies))
|
|
|
|
def test_task_cost_relation_reject(self):
|
|
self._insert("S1", status="running", created_by="other")
|
|
worker = self._worker()
|
|
bodies = self._deliver(worker, self._sender(worker), _base_result(
|
|
task_costs=[{"task_id": "S1"}]))
|
|
self.assertTrue(any("不在派生链内且创建人不同" in b for b in bodies))
|
|
|
|
def test_task_cost_ancestry_failure(self):
|
|
self._insert("E1", status="running", parent="PM")
|
|
worker = self._worker()
|
|
with mock.patch("cloud.worker._marker_ancestry",
|
|
side_effect=RuntimeError("db down")):
|
|
bodies = self._deliver(worker, self._sender(worker),
|
|
_base_result(
|
|
task_costs=[{"task_id": "E1"}]))
|
|
self.assertTrue(any("不在派生链内且创建人不同" in b for b in bodies))
|
|
|
|
def test_task_cost_store_close_failure_swallowed(self):
|
|
self._insert("E1", status="completed", parent="PM",
|
|
result_text="done")
|
|
worker = self._worker()
|
|
|
|
class _FakeConn:
|
|
def close(self):
|
|
raise RuntimeError("close down")
|
|
|
|
class _FakeStore:
|
|
def __init__(self, db_path):
|
|
self._conn = _FakeConn()
|
|
|
|
def report_by_task(self, task_id):
|
|
return {"usd_micros": 1234567, "total_tokens": 100,
|
|
"actual_tokens": 80, "estimated_tokens": 20,
|
|
"turns": 2, "unpriced_actual_tokens": 5}
|
|
|
|
with mock.patch("cloud.worker.CostStore", _FakeStore):
|
|
bodies = self._deliver(worker, self._sender(worker),
|
|
_base_result(
|
|
task_costs=[{"task_id": "E1"}]))
|
|
joined = "\n".join(bodies)
|
|
self.assertIn("成本：$1.2346", joined)
|
|
self.assertIn("无价 actual token 5", joined)
|
|
|
|
def test_task_cost_report_failure(self):
|
|
self._insert("E1", status="completed", parent="PM")
|
|
worker = self._worker()
|
|
|
|
class _FailingStore:
|
|
def __init__(self, db_path):
|
|
self._conn = SimpleNamespace(
|
|
close=lambda: None)
|
|
|
|
def report_by_task(self, task_id):
|
|
raise RuntimeError("cost down")
|
|
|
|
with mock.patch("cloud.worker.CostStore", _FailingStore):
|
|
bodies = self._deliver(worker, self._sender(worker),
|
|
_base_result(
|
|
task_costs=[{"task_id": "E1"}]))
|
|
self.assertTrue(any("成本失败" in b for b in bodies))
|
|
|
|
def test_quota_owner_from_parent_via_stubbed_auth(self):
|
|
# Defensive fallback only: the real _marker_auth returns
|
|
# False for a creator-less sender, so this branch is only
|
|
# reachable when auth is stubbed True. The test pins the
|
|
# fallback logic itself (owner resolved from the parent).
|
|
self._insert("ORPH", status="running", created_by=None,
|
|
parent="PM")
|
|
worker = self._worker()
|
|
worker._marker_auth = lambda task, action: (True, "")
|
|
sender = self._sender(worker, "ORPH")
|
|
bodies = self._deliver(worker, sender,
|
|
_base_result(quota_status=[{}]),
|
|
task_id="ORPH")
|
|
# Owner resolved from the parent (tester, no explicit quota
|
|
# row) so the report shows the default quota, not "unlimited".
|
|
self.assertTrue(any("配额" in b and "总额 $10.0000" in b
|
|
for b in bodies), bodies)
|
|
|
|
def test_quota_no_owner_via_stubbed_auth(self):
|
|
self._insert("ORPH2", status="running", created_by=None)
|
|
worker = self._worker()
|
|
worker._marker_auth = lambda task, action: (True, "")
|
|
sender = self._sender(worker, "ORPH2")
|
|
bodies = self._deliver(worker, sender,
|
|
_base_result(quota_status=[{}]),
|
|
task_id="ORPH2")
|
|
self.assertTrue(any("发送方无创建人" in b for b in bodies))
|
|
|
|
def test_quota_unlimited_explicit_null_row(self):
|
|
conn = sqlite3.connect(self.db_path)
|
|
try:
|
|
conn.execute(
|
|
"CREATE TABLE IF NOT EXISTS user_quotas("
|
|
"user_id TEXT PRIMARY KEY, quota_usd_micros BIGINT,"
|
|
" updated_at DOUBLE PRECISION)")
|
|
conn.execute(
|
|
"INSERT OR REPLACE INTO user_quotas"
|
|
"(user_id, quota_usd_micros, updated_at)"
|
|
" VALUES('tester', NULL, 0)")
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
worker = self._worker()
|
|
bodies = self._deliver(worker, self._sender(worker),
|
|
_base_result(quota_status=[{}]))
|
|
self.assertTrue(any("不限额" in b for b in bodies))
|
|
|
|
def test_quota_limit_failure(self):
|
|
worker = self._worker()
|
|
with mock.patch.object(worker, "_quota_limit_and_used",
|
|
side_effect=RuntimeError("quota down")):
|
|
bodies = self._deliver(worker, self._sender(worker),
|
|
_base_result(quota_status=[{}]))
|
|
self.assertTrue(any("查配额失败" in b for b in bodies))
|
|
|
|
def test_task_files_target_missing(self):
|
|
worker = self._worker()
|
|
bodies = self._deliver(worker, self._sender(worker), _base_result(
|
|
task_files=[{"task_id": "NOPE", "path": ""}]))
|
|
self.assertTrue(any("不存在（或属于别的项目）" in b for b in bodies))
|
|
|
|
def test_task_files_ancestry_failure(self):
|
|
self._insert("E1", status="running", parent="PM")
|
|
worker = self._worker()
|
|
with mock.patch("cloud.worker._marker_ancestry",
|
|
side_effect=RuntimeError("db down")):
|
|
bodies = self._deliver(worker, self._sender(worker),
|
|
_base_result(
|
|
task_files=[{"task_id": "E1",
|
|
"path": ""}]))
|
|
self.assertTrue(any("不在派生链内且创建人不同" in b for b in bodies))
|
|
|
|
|
|
class TaskFilesWorktreeTestCase(DeliverBase):
|
|
def _worktree_worker(self):
|
|
worker = self._worker(tenants_root=self.tmp.name)
|
|
worktree = os.path.join(self.tmp.name, "p1", "work",
|
|
"host-worktrees", "E1")
|
|
os.makedirs(worktree)
|
|
return worker, worktree
|
|
|
|
def test_task_files_dir_and_file_happy(self):
|
|
worker, worktree = self._worktree_worker()
|
|
Path(worktree, "a.txt").write_text("hello-file", encoding="utf-8")
|
|
self._insert("E1", status="running", parent="PM")
|
|
bodies = self._deliver(worker, self._sender(worker), _base_result(
|
|
task_files=[{"task_id": "E1", "path": ""},
|
|
{"task_id": "E1", "path": "a.txt"}]))
|
|
joined = "\n".join(bodies)
|
|
self.assertIn("a.txt", joined)
|
|
self.assertIn("hello-file", joined)
|
|
|
|
def test_task_files_dir_list_failure(self):
|
|
worker, worktree = self._worktree_worker()
|
|
Path(worktree, "a.txt").write_text("x", encoding="utf-8")
|
|
self._insert("E1", status="running", parent="PM")
|
|
with mock.patch("os.listdir", side_effect=OSError("list down")):
|
|
bodies = self._deliver(worker, self._sender(worker),
|
|
_base_result(
|
|
task_files=[{"task_id": "E1",
|
|
"path": ""}]))
|
|
self.assertTrue(any("列任务" in b and "目录失败" in b
|
|
for b in bodies))
|
|
|
|
def test_task_files_file_read_failure(self):
|
|
worker, worktree = self._worktree_worker()
|
|
target = Path(worktree, "a.txt")
|
|
target.write_text("secret", encoding="utf-8")
|
|
self._insert("E1", status="running", parent="PM")
|
|
real_open = open
|
|
|
|
def _selective_open(path, *args, **kwargs):
|
|
if os.fspath(path) == str(target):
|
|
raise OSError("read down")
|
|
return real_open(path, *args, **kwargs)
|
|
|
|
with mock.patch("builtins.open", side_effect=_selective_open):
|
|
bodies = self._deliver(worker, self._sender(worker),
|
|
_base_result(
|
|
task_files=[{"task_id": "E1",
|
|
"path": "a.txt"}]))
|
|
self.assertTrue(any("读任务" in b and "失败" in b for b in bodies))
|
|
|
|
def test_task_files_binary_rejected(self):
|
|
worker, worktree = self._worktree_worker()
|
|
Path(worktree, "a.bin").write_bytes(b"\x00\x01\x02abc")
|
|
self._insert("E1", status="running", parent="PM")
|
|
bodies = self._deliver(worker, self._sender(worker), _base_result(
|
|
task_files=[{"task_id": "E1", "path": "a.bin"}]))
|
|
self.assertTrue(any("二进制文件" in b for b in bodies))
|
|
|
|
def test_task_files_invalid_utf8_rejected(self):
|
|
worker, worktree = self._worktree_worker()
|
|
Path(worktree, "bad.bin").write_bytes(b"\xff\xfe\x00abc")
|
|
self._insert("E1", status="running", parent="PM")
|
|
bodies = self._deliver(worker, self._sender(worker), _base_result(
|
|
task_files=[{"task_id": "E1", "path": "bad.bin"}]))
|
|
self.assertTrue(any("二进制文件" in b for b in bodies))
|
|
|
|
def test_task_files_truncation(self):
|
|
worker, worktree = self._worktree_worker()
|
|
Path(worktree, "big.txt").write_text("x" * 21000, encoding="utf-8")
|
|
self._insert("E1", status="running", parent="PM")
|
|
bodies = self._deliver(worker, self._sender(worker), _base_result(
|
|
task_files=[{"task_id": "E1", "path": "big.txt"}]))
|
|
self.assertTrue(any("已截断前 20000 字" in b for b in bodies))
|
|
|
|
|
|
class PollWaitGoalPeerTestCase(DeliverBase):
|
|
def test_poll_and_wait_exceptions(self):
|
|
worker = self._worker()
|
|
with mock.patch.object(worker, "_poll_task_snapshot",
|
|
side_effect=RuntimeError("poll down")), \
|
|
mock.patch.object(worker, "_wait_task_targets",
|
|
side_effect=RuntimeError("wait down")):
|
|
bodies = self._deliver(worker, self._sender(worker),
|
|
_base_result(
|
|
task_polls=[{"task_id": "E1",
|
|
"interval_seconds": 5,
|
|
"timeout_seconds": 0,
|
|
"chars": 2000}],
|
|
wait_threads=[{"task_ids": ["E1"],
|
|
"interval_seconds": 5,
|
|
"timeout_seconds": 0,
|
|
"chars": 2000}]))
|
|
joined = "\n".join(bodies)
|
|
self.assertIn("轮询任务 E1 失败", joined)
|
|
self.assertIn("等待任务失败", joined)
|
|
|
|
def test_goal_converts_queued_after_report(self):
|
|
worker = self._worker()
|
|
bodies = self._deliver(worker, self._sender(worker), _base_result(
|
|
goal_converts=[{"goal": "做目标"}]))
|
|
self.assertEqual(bodies, ["/goal 做目标"])
|
|
|
|
def test_peer_messages_delivery_error(self):
|
|
worker = self._worker()
|
|
sender = self._sender(worker)
|
|
with mock.patch("cloud.worker._deliver_peer_message",
|
|
side_effect=RuntimeError("peer down")):
|
|
worker._deliver_peer_messages(sender, {
|
|
"status": "completed",
|
|
"task_messages": [{"to_task_id": "E1", "body": "hi",
|
|
"mode": "followup"}]})
|
|
conn = sqlite3.connect(self.db_path)
|
|
try:
|
|
tables = [r[0] for r in conn.execute(
|
|
"SELECT name FROM sqlite_master"
|
|
" WHERE type = 'table'").fetchall()]
|
|
if "audit_log" in tables:
|
|
rows = conn.execute(
|
|
"SELECT detail FROM audit_log"
|
|
" WHERE action = 'worker_peer_messages'").fetchall()
|
|
self.assertTrue(any("rejected=1" in (r[0] or "")
|
|
for r in rows), rows)
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|