233 lines
9.1 KiB
Python
233 lines
9.1 KiB
Python
"""自动化与分诊收件箱模块单测(cloud/automations.py)。
|
|
|
|
不跑真实模型、不联网：临时 SQLite 文件 + 模块函数直调。
|
|
覆盖：建表幂等、创建校验与同名规则、到期边界、claim 抢单
|
|
防重、取消语义、产出记录、分诊四段聚合。
|
|
"""
|
|
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
try:
|
|
from cloud.automations import (
|
|
cancel_automation,
|
|
claim_due_automation,
|
|
create_automation,
|
|
due_automations,
|
|
ensure_automations_tables,
|
|
get_automation,
|
|
is_due,
|
|
list_automation_runs,
|
|
list_automations,
|
|
record_automation_run,
|
|
triage_inbox,
|
|
)
|
|
from cloud.peer_messages import ensure_followups_table
|
|
except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
|
from cloud.automations import (
|
|
cancel_automation,
|
|
claim_due_automation,
|
|
create_automation,
|
|
due_automations,
|
|
ensure_automations_tables,
|
|
get_automation,
|
|
is_due,
|
|
list_automation_runs,
|
|
list_automations,
|
|
record_automation_run,
|
|
triage_inbox,
|
|
)
|
|
from cloud.peer_messages import ensure_followups_table
|
|
|
|
|
|
class AutomationsTestCase(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.db_path = os.path.join(self.tmp.name, "control.db")
|
|
self.conn = sqlite3.connect(self.db_path)
|
|
self.conn.row_factory = sqlite3.Row
|
|
self.conn.execute(
|
|
"CREATE TABLE tasks(id TEXT PRIMARY KEY, project_id TEXT,"
|
|
" prompt TEXT, status TEXT, created_at DOUBLE PRECISION,"
|
|
" updated_at DOUBLE PRECISION, error_text TEXT)")
|
|
ensure_automations_tables(self.conn)
|
|
ensure_followups_table(self.conn)
|
|
self.conn.commit()
|
|
self.now = 1700000000.0
|
|
self.project = "proj-1"
|
|
|
|
def tearDown(self):
|
|
self.conn.close()
|
|
self.tmp.cleanup()
|
|
|
|
def _make(self, name="daily", interval=3600, **kw):
|
|
kw.setdefault("project_id", self.project)
|
|
return create_automation(
|
|
self.conn, kw.pop("project_id"), name, "做每日检查",
|
|
"user-1", interval, now=self.now, **kw)
|
|
|
|
def test_ensure_is_idempotent(self):
|
|
ensure_automations_tables(self.conn)
|
|
ensure_automations_tables(self.conn)
|
|
self.conn.commit()
|
|
rows = self.conn.execute(
|
|
"SELECT name FROM sqlite_master WHERE type = 'table'"
|
|
" AND name IN ('automations', 'automation_runs')").fetchall()
|
|
self.assertEqual(len(rows), 2)
|
|
|
|
def test_create_fields_and_due_immediately(self):
|
|
auto = self._make()
|
|
self.assertEqual(auto["status"], "active")
|
|
self.assertEqual(auto["interval_seconds"], 3600)
|
|
self.assertIsNone(auto["last_run_at"])
|
|
self.assertEqual(auto["next_run_at"], self.now)
|
|
self.assertTrue(auto["due"])
|
|
self.assertTrue(is_due(auto, self.now))
|
|
got = get_automation(self.conn, auto["id"])
|
|
self.assertEqual(got["name"], "daily")
|
|
self.assertIsNone(get_automation(self.conn, "missing"))
|
|
|
|
def test_create_validation(self):
|
|
for name, prompt, interval in (
|
|
("", "x", 3600),
|
|
(" ", "x", 3600),
|
|
("n", "", 3600),
|
|
("n", "x", 59),
|
|
("n", "x", True),
|
|
("n", "x", 3600.5),
|
|
("x" * 101, "x", 3600),
|
|
("n", "x" * 20001, 3600)):
|
|
with self.assertRaises(ValueError, msg=(name, interval)):
|
|
create_automation(
|
|
self.conn, self.project, name, prompt, "user-1",
|
|
interval, now=self.now)
|
|
|
|
def test_duplicate_active_name_rejected_reusable_after_cancel(self):
|
|
self._make(name="dup")
|
|
with self.assertRaises(ValueError):
|
|
self._make(name="dup")
|
|
autos = list_automations(self.conn, self.project, self.now)
|
|
self.assertEqual(len(autos), 1)
|
|
ok, cancelled = cancel_automation(
|
|
self.conn, autos[0]["id"], self.now)
|
|
self.assertTrue(ok)
|
|
self.assertEqual(cancelled["status"], "cancelled")
|
|
again = self._make(name="dup")
|
|
self.assertEqual(again["status"], "active")
|
|
|
|
def test_due_boundary_matches_host_rundue_rule(self):
|
|
auto = self._make(interval=3600)
|
|
self.conn.execute(
|
|
"UPDATE automations SET last_run_at = ? WHERE id = ?",
|
|
(self.now - 3599, auto["id"]))
|
|
self.conn.commit()
|
|
self.assertEqual(due_automations(
|
|
self.conn, self.project, self.now), [])
|
|
self.conn.execute(
|
|
"UPDATE automations SET last_run_at = ? WHERE id = ?",
|
|
(self.now - 3600, auto["id"]))
|
|
self.conn.commit()
|
|
due = due_automations(self.conn, self.project, self.now)
|
|
self.assertEqual([d["id"] for d in due], [auto["id"]])
|
|
|
|
def test_claim_is_the_duplicate_gate(self):
|
|
auto = self._make(interval=3600)
|
|
self.assertTrue(claim_due_automation(
|
|
self.conn, auto["id"], self.now))
|
|
# 同一间隔内第二次抢单必须失败：这就是防重复执行。
|
|
self.assertFalse(claim_due_automation(
|
|
self.conn, auto["id"], self.now))
|
|
self.assertFalse(claim_due_automation(
|
|
self.conn, auto["id"], self.now + 3599))
|
|
# 下个间隔到了才能再抢。
|
|
self.assertTrue(claim_due_automation(
|
|
self.conn, auto["id"], self.now + 3600))
|
|
self.assertFalse(claim_due_automation(
|
|
self.conn, "missing", self.now))
|
|
ok, _ = cancel_automation(self.conn, auto["id"], self.now)
|
|
self.assertTrue(ok)
|
|
self.assertFalse(claim_due_automation(
|
|
self.conn, auto["id"], self.now + 7200))
|
|
|
|
def test_cancel_semantics(self):
|
|
ok, auto = cancel_automation(self.conn, "missing", self.now)
|
|
self.assertFalse(ok)
|
|
self.assertIsNone(auto)
|
|
made = self._make()
|
|
ok, auto = cancel_automation(self.conn, made["id"], self.now)
|
|
self.assertTrue(ok)
|
|
self.assertEqual(auto["status"], "cancelled")
|
|
self.assertFalse(auto["due"])
|
|
ok, auto = cancel_automation(self.conn, made["id"], self.now)
|
|
self.assertFalse(ok)
|
|
self.assertEqual(auto["status"], "cancelled")
|
|
|
|
def test_record_run_and_list_runs(self):
|
|
auto = self._make()
|
|
run = record_automation_run(
|
|
self.conn, auto["id"], self.project, task_id="task-1",
|
|
status="created", detail="claimed", now=self.now)
|
|
self.assertEqual(run["task_id"], "task-1")
|
|
self.assertEqual(run["status"], "created")
|
|
runs = list_automation_runs(self.conn, auto["id"])
|
|
self.assertEqual(len(runs), 1)
|
|
with self.assertRaises(ValueError):
|
|
record_automation_run(
|
|
self.conn, "missing", self.project, now=self.now)
|
|
|
|
def test_triage_inbox_four_sections(self):
|
|
auto = self._make()
|
|
other = self._make(name="weekly", interval=86400)
|
|
cancel_automation(self.conn, other["id"], self.now)
|
|
self.conn.execute(
|
|
"INSERT INTO tasks(id, project_id, prompt, status,"
|
|
" created_at, updated_at, error_text)"
|
|
" VALUES('t-fail', ?, '坏任务', 'failed', ?, ?, 'boom')",
|
|
(self.project, self.now, self.now))
|
|
self.conn.execute(
|
|
"INSERT INTO tasks(id, project_id, prompt, status,"
|
|
" created_at, updated_at, error_text)"
|
|
" VALUES('t-run', ?, '跑中任务', 'running', ?, ?, NULL)",
|
|
(self.project, self.now, self.now))
|
|
self.conn.execute(
|
|
"INSERT INTO task_followups(id, task_id, project_id, body,"
|
|
" mode, status, created_by, sender_task_id, created_at,"
|
|
" updated_at) VALUES('f-1', 't-run', ?, '请看一下',"
|
|
" 'queued', 'pending', 'user-1', NULL, ?, ?)",
|
|
(self.project, self.now, self.now))
|
|
self.conn.execute(
|
|
"INSERT INTO tasks(id, project_id, prompt, status,"
|
|
" created_at, updated_at, error_text)"
|
|
" VALUES('t-done', ?, '已完成', 'completed', ?, ?, NULL)",
|
|
(self.project, self.now, self.now))
|
|
record_automation_run(
|
|
self.conn, auto["id"], self.project, task_id="t-done",
|
|
status="created", detail="", now=self.now)
|
|
self.conn.commit()
|
|
inbox = triage_inbox(self.conn, self.project, self.now)
|
|
self.assertEqual(
|
|
[d["id"] for d in inbox["automations_due"]], [auto["id"]])
|
|
self.assertEqual(
|
|
[t["id"] for t in inbox["failed_tasks"]], ["t-fail"])
|
|
self.assertEqual(
|
|
[f["id"] for f in inbox["pending_followups"]], ["f-1"])
|
|
self.assertEqual(len(inbox["recent_runs"]), 1)
|
|
self.assertEqual(inbox["recent_runs"][0]["task_status"],
|
|
"completed")
|
|
self.assertEqual(inbox["counts"], {
|
|
"automations_due": 1,
|
|
"failed_tasks": 1,
|
|
"pending_followups": 1,
|
|
"recent_runs": 1,
|
|
})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|