auto-sync: 2026-06-05 23:18:33

tests/e2e/test_spawner_e2e.py

@@ -0,0 +1,80 @@
+"""F9 Agent Spawner 集成测试 — 超时/失败/spawn 真实流程"""
+
+import asyncio
+import pytest
+from pathlib import Path
+
+from src.blackboard.operations import Blackboard
+from src.daemon.spawner import AgentSpawner
+
+pytestmark = [pytest.mark.e2e, pytest.mark.slow]
+
+skip_no_integration = pytest.mark.skipif(
+    not __import__("os").environ.get("RUN_INTEGRATION"),
+    reason="Set RUN_INTEGRATION=1 to run real spawn tests",
+)
+
+
+@pytest.fixture
+def db_path(tmp_path):
+    return tmp_path / "blackboard.db"
+
+
+@pytest.fixture
+def bb(db_path):
+    return Blackboard(db_path)
+
+
+@pytest.fixture
+def real_spawner(db_path):
+    return AgentSpawner(db_path=db_path, dry_run=False, agent_timeout=2.0)
+
+
+# ---------------------------------------------------------------------------
+# T2: 超时处理
+# ---------------------------------------------------------------------------
+
+@skip_no_integration
+class TestTimeout:
+    def test_timeout_kills_process(self, tmp_path):
+        """超时后 kill 进程"""
+        db_path = tmp_path / "blackboard.db"
+        Blackboard(db_path)  # init
+        spawner = AgentSpawner(db_path=db_path, dry_run=False, agent_timeout=0.5)
+
+        # Spawn a long-running process (sleep 10)
+        session_id = asyncio.run(
+            spawner.spawn_full_agent(
+                "test-agent",
+                "sleep 10",
+                task_id=None,
+            )
+        )
+        # Wait for timeout
+        asyncio.run(asyncio.sleep(1.0))
+
+        session = spawner.get_session(session_id)
+        if session:
+            assert session["status"] in ("timed_out", "running", "completed")
+
+
+# ---------------------------------------------------------------------------
+# T3: spawn 失败
+# ---------------------------------------------------------------------------
+
+@skip_no_integration
+class TestSpawnFailure:
+    def test_nonexistent_command(self, real_spawner, db_path, bb):
+        """命令不存在 → spawn_failed"""
+        bb.create_task(
+            __import__("src.blackboard.models", fromlist=["Task"]).Task(
+                id="t1", title="T", status="pending", assigned_by="d"
+            )
+        )
+
+        try:
+            asyncio.run(
+                real_spawner.spawn_full_agent("test", "msg", task_id="t1")
+            )
+        except Exception:
+            pass  # Expected - command may fail