auto-sync: 2026-05-17 05:53:43

tests/test_inbox.py

@@ -1,6 +1,6 @@
 """F7 Inbox JSONL Watcher 单元测试
 
-按测试计划 test-plan-v2.6.md §F7：
+按 test-plan-v2.6.md §F7：
 - T1: 写入+消费（P0）
 - T2: truncate（P0）
 - T3: 并发写入（P0）
@@ -10,9 +10,9 @@
 
 import asyncio
 import json
-import threading
 import pytest
 from pathlib import Path
+from typing import Any, Dict, List
 
 from src.daemon.inbox import InboxWatcher
 
@@ -32,47 +32,51 @@ def watcher(inbox_path):
 # ---------------------------------------------------------------------------
 
 class TestWriteConsume:
-    def test_single_event(self, watcher, inbox_path):
-        """写入单条事件 → 消费到"""
-        event = {"type": "task_done", "task_id": "t1", "agent": "a"}
-        watcher.write_event(event)
+    def test_single_event(self, inbox_path, watcher):
+        """写入一个事件 → poll 消费到"""
+        event = {"type": "agent_output", "task_id": "t1", "agent": "a1",
+                 "project_id": "p1"}
+        InboxWatcher.write_event(inbox_path, event)
 
-        events = watcher.consume()
-        assert len(events) == 1
-        assert events[0]["type"] == "task_done"
-        assert events[0]["task_id"] == "t1"
+        result = asyncio.run(watcher.poll())
+        assert len(result) == 1
+        assert result[0]["type"] == "agent_output"
+        assert result[0]["task_id"] == "t1"
 
-    def test_multiple_events(self, watcher, inbox_path):
-        """写入多条事件 → 全部消费到"""
-        for i in range(5):
-            watcher.write_event({"type": "progress", "task_id": f"t{i}", "pct": i * 20})
+    def test_multiple_events(self, inbox_path, watcher):
+        """写入多个事件 → 全部消费"""
+        events = [
+            {"type": "agent_claim", "task_id": f"t{i}", "agent": "a1",
+             "project_id": "p1"}
+            for i in range(5)
+        ]
+        InboxWatcher.write_events(inbox_path, events)
 
-        events = watcher.consume()
-        assert len(events) == 5
-        assert events[0]["task_id"] == "t0"
-        assert events[4]["task_id"] == "t4"
+        result = asyncio.run(watcher.poll())
+        assert len(result) == 5
 
-    def test_consume_increments_counter(self, watcher, inbox_path):
-        """total_consumed 递增"""
-        assert watcher.total_consumed == 0
-        watcher.write_event({"type": "test"})
-        watcher.consume()
-        assert watcher.total_consumed == 1
-        watcher.write_event({"type": "test2"})
-        watcher.write_event({"type": "test3"})
-        watcher.consume()
-        assert watcher.total_consumed == 3
+    def test_callback_called(self, inbox_path):
+        """回调函数被调用"""
+        received: List[Dict] = []
 
-    def test_consume_empty_file_returns_empty(self, watcher, inbox_path):
-        """文件不存在时返回空列表"""
-        assert watcher.consume() == []
+        async def callback(event):
+            received.append(event)
 
-    def test_static_write(self, inbox_path):
-        """静态方法写入"""
-        InboxWatcher.write_event_to(inbox_path, {"type": "static"})
-        events = InboxWatcher(inbox_path).consume()
-        assert len(events) == 1
-        assert events[0]["type"] == "static"
+        w = InboxWatcher(inbox_path, process_callback=callback, watch_interval=0.05)
+        InboxWatcher.write_event(inbox_path, {"type": "test", "x": 1})
+
+        asyncio.run(w.poll())
+        assert len(received) == 1
+        assert received[0]["x"] == 1
+
+    def test_counter_increments(self, inbox_path, watcher):
+        """total_processed 计数正确"""
+        assert watcher.total_processed == 0
+        InboxWatcher.write_events(inbox_path, [
+            {"type": "a"}, {"type": "b"}, {"type": "c"},
+        ])
+        asyncio.run(watcher.poll())
+        assert watcher.total_processed == 3
 
 
 # ---------------------------------------------------------------------------
@@ -80,36 +84,34 @@ class TestWriteConsume:
 # ---------------------------------------------------------------------------
 
 class TestTruncate:
-    def test_file_cleared_after_consume(self, watcher, inbox_path):
+    def test_file_truncated_after_poll(self, inbox_path, watcher):
         """消费后文件被清空"""
-        watcher.write_event({"type": "test"})
+        InboxWatcher.write_event(inbox_path, {"type": "test"})
         assert inbox_path.exists()
         assert inbox_path.stat().st_size > 0
 
-        watcher.consume()
+        asyncio.run(watcher.poll())
+        assert inbox_path.exists()  # 文件还在
+        assert inbox_path.stat().st_size == 0  # 但内容被清空
 
-        # 文件存在但为空
-        assert inbox_path.exists()
-        assert inbox_path.stat().st_size == 0
+    def test_second_poll_gets_nothing(self, inbox_path, watcher):
+        """第二次 poll 无新事件"""
+        InboxWatcher.write_event(inbox_path, {"type": "test"})
+        r1 = asyncio.run(watcher.poll())
+        assert len(r1) == 1
 
-    def test_second_consume_returns_empty(self, watcher, inbox_path):
-        """二次消费无新事件"""
-        watcher.write_event({"type": "test"})
-        events1 = watcher.consume()
-        assert len(events1) == 1
+        r2 = asyncio.run(watcher.poll())
+        assert len(r2) == 0
 
-        events2 = watcher.consume()
-        assert len(events2) == 0
+    def test_truncate_then_write(self, inbox_path, watcher):
+        """truncate 后再写入，能正常消费"""
+        InboxWatcher.write_event(inbox_path, {"type": "first"})
+        asyncio.run(watcher.poll())
 
-    def test_write_after_truncate(self, watcher, inbox_path):
-        """truncate 后可以继续写入"""
-        watcher.write_event({"type": "first"})
-        watcher.consume()
-
-        watcher.write_event({"type": "second"})
-        events = watcher.consume()
-        assert len(events) == 1
-        assert events[0]["type"] == "second"
+        InboxWatcher.write_event(inbox_path, {"type": "second"})
+        r = asyncio.run(watcher.poll())
+        assert len(r) == 1
+        assert r[0]["type"] == "second"
 
 
 # ---------------------------------------------------------------------------
@@ -117,57 +119,52 @@ class TestTruncate:
 # ---------------------------------------------------------------------------
 
 class TestConcurrentWrite:
-    def test_multi_agent_concurrent(self, inbox_path):
-        """多 Agent 同时写入不同事件，全部消费到"""
-        n_agents = 5
-        n_events_per_agent = 10
-        barrier = threading.Barrier(n_agents)
+    def test_concurrent_writers(self, inbox_path, watcher):
+        """多个模拟 Agent 同时写入不同行"""
+        import threading
 
-        def writer(agent_id):
-            barrier.wait()
-            for i in range(n_events_per_agent):
-                InboxWatcher.write_event_to(inbox_path, {
+        def write_batch(agent_id, count):
+            for i in range(count):
+                InboxWatcher.write_event(inbox_path, {
+                    "type": "agent_output",
                     "agent": agent_id,
-                    "seq": i,
+                    "task_id": f"{agent_id}-t{i}",
                 })
 
-        threads = [threading.Thread(target=writer, args=(f"agent-{j}",))
-                   for j in range(n_agents)]
+        threads = [
+            threading.Thread(target=write_batch, args=(f"agent-{i}", 10))
+            for i in range(3)
+        ]
         for t in threads:
             t.start()
         for t in threads:
-            t.join(timeout=5)
+            t.join()
 
-        watcher = InboxWatcher(inbox_path)
-        events = watcher.consume()
-        assert len(events) == n_agents * n_events_per_agent
-
-        # 每个 agent 的事件都到齐
-        agents_seen = {e["agent"] for e in events}
-        assert len(agents_seen) == n_agents
+        result = asyncio.run(watcher.poll())
+        assert len(result) == 30
 
 
 # ---------------------------------------------------------------------------
 # T4: 损坏行恢复（P1）
 # ---------------------------------------------------------------------------
 
-class TestBadLineRecovery:
-    def test_skip_invalid_json(self, watcher, inbox_path):
+class TestCorruptRecovery:
+    def test_invalid_json_skipped(self, inbox_path, watcher):
         """非法 JSON 行跳过不崩溃"""
         inbox_path.parent.mkdir(parents=True, exist_ok=True)
         with open(inbox_path, "w") as f:
-            f.write('{"type": "good"}\n')
+            f.write('{"type": "valid"}\n')
             f.write('not json at all\n')
-            f.write('{"type": "also_good"}\n')
-            f.write('{broken json\n')
+            f.write('{"type": "also_valid"}\n')
 
-        events = watcher.consume()
-        assert len(events) == 2
-        assert events[0]["type"] == "good"
-        assert events[1]["type"] == "also_good"
+        result = asyncio.run(watcher.poll())
+        assert len(result) == 2
+        assert result[0]["type"] == "valid"
+        assert result[1]["type"] == "also_valid"
+        assert watcher.total_errors == 1
 
-    def test_non_dict_json_skipped(self, watcher, inbox_path):
-        """合法 JSON 但非 dict 也跳过"""
+    def test_non_dict_skipped(self, inbox_path, watcher):
+        """非 dict 类型的 JSON 值跳过"""
         inbox_path.parent.mkdir(parents=True, exist_ok=True)
         with open(inbox_path, "w") as f:
             f.write('{"type": "good"}\n')
@@ -175,18 +172,20 @@ class TestBadLineRecovery:
             f.write('"just a string"\n')
             f.write('42\n')
 
-        events = watcher.consume()
-        assert len(events) == 1
-        assert events[0]["type"] == "good"
+        result = asyncio.run(watcher.poll())
+        assert len(result) == 1
+        assert watcher.total_errors >= 1
 
-    def test_empty_lines_skipped(self, watcher, inbox_path):
-        """空行不影响"""
+    def test_all_corrupt(self, inbox_path, watcher):
+        """全部损坏行 → 返回空列表，不崩溃"""
         inbox_path.parent.mkdir(parents=True, exist_ok=True)
         with open(inbox_path, "w") as f:
-            f.write('\n\n{"type": "good"}\n\n\n')
+            f.write("bad line 1\n")
+            f.write("bad line 2\n")
 
-        events = watcher.consume()
-        assert len(events) == 1
+        result = asyncio.run(watcher.poll())
+        assert len(result) == 0
+        assert watcher.total_errors == 2
 
 
 # ---------------------------------------------------------------------------
@@ -194,19 +193,24 @@ class TestBadLineRecovery:
 # ---------------------------------------------------------------------------
 
 class TestEmptyFile:
-    def test_empty_file_no_events(self, watcher, inbox_path):
-        """空文件返回空列表"""
+    def test_nonexistent_file(self, inbox_path, watcher):
+        """文件不存在 → 返回空"""
+        result = asyncio.run(watcher.poll())
+        assert len(result) == 0
+
+    def test_empty_file(self, inbox_path, watcher):
+        """空文件 → 返回空"""
         inbox_path.parent.mkdir(parents=True, exist_ok=True)
         inbox_path.touch()
-        events = watcher.consume()
-        assert events == []
+        result = asyncio.run(watcher.poll())
+        assert len(result) == 0
 
-    def test_whitespace_only_file(self, watcher, inbox_path):
-        """只有空白的文件返回空列表"""
+    def test_whitespace_only_file(self, inbox_path, watcher):
+        """只有空白字符的文件 → 返回空"""
         inbox_path.parent.mkdir(parents=True, exist_ok=True)
-        inbox_path.write_text("   \n  \n  \n")
-        events = watcher.consume()
-        assert events == []
+        inbox_path.write_text("  \n  \n\n  ")
+        result = asyncio.run(watcher.poll())
+        assert len(result) == 0
 
 
 # ---------------------------------------------------------------------------
@@ -214,33 +218,38 @@ class TestEmptyFile:
 # ---------------------------------------------------------------------------
 
 class TestWatcherLifecycle:
-    def test_start_stop(self, watcher, inbox_path):
-        """可以启动和停止"""
+    def test_start_stop(self, inbox_path):
+        """watcher 可以启动和停止"""
+        w = InboxWatcher(inbox_path, watch_interval=0.05)
+        assert not w.is_running
+
         async def run():
-            await watcher.start()
-            assert watcher.is_running
-            await asyncio.sleep(0.2)
-            await watcher.stop()
-            assert not watcher.is_running
+            await w.start()
+            assert w.is_running
+            await asyncio.sleep(0.1)
+            await w.stop()
 
         asyncio.run(run())
+        assert not w.is_running
 
-    def test_auto_consume_via_callback(self, watcher, inbox_path):
-        """watcher 循环自动消费并通过回调传递事件"""
-        collected = []
+    def test_auto_consumes_events(self, inbox_path):
+        """watcher 运行期间自动消费事件"""
+        received: List[Dict] = []
 
-        async def on_events(events):
-            collected.extend(events)
+        async def callback(event):
+            received.append(event)
 
-        watcher.on_events = on_events
+        w = InboxWatcher(inbox_path, process_callback=callback, watch_interval=0.05)
 
         async def run():
-            await watcher.start()
+            await w.start()
+            # 等一下让 watcher 启动
+            await asyncio.sleep(0.1)
             # 写入事件
-            watcher.write_event({"type": "auto"})
-            await asyncio.sleep(0.3)  # 等待 watcher 消费
-            await watcher.stop()
+            InboxWatcher.write_event(inbox_path, {"type": "test", "seq": 1})
+            # 等一下让 watcher 消费
+            await asyncio.sleep(0.2)
+            await w.stop()
 
         asyncio.run(run())
-        assert len(collected) >= 1
-        assert collected[0]["type"] == "auto"
+        assert len(received) >= 1