"""自动化与分诊收件箱模块单测(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()