cfdaily
81 lines
2.3 KiB
Python
81 lines
2.3 KiB
Python
"""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
|