2ff023a65b
- Add OpenClawClient wrapping CLI commands (create/delete/list crons) - Implement pool initialization, scale up/down, auto-scaling logic - Implement cron state sync and stuck cron detection - Add pool status and manual scaling APIs - Add 13 unit tests and 5 integration tests Pool automatically scales between 5-20 crons based on load. All 778 tests passing.
296 lines
9.4 KiB
Python
296 lines
9.4 KiB
Python
"""Unit tests for cron pool management."""
|
|
|
|
from datetime import timedelta
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
from sqlmodel import Session, select
|
|
|
|
from agenteval.intelligent_eval import cron_pool
|
|
from agenteval.intelligent_eval.openclaw_client import OpenClawClient, OpenClawCron
|
|
from agenteval.storage.db import OpenClawCronPoolDB, utc_now
|
|
|
|
|
|
@pytest.fixture()
|
|
def mock_openclaw_client():
|
|
"""Mock OpenClaw client."""
|
|
client = MagicMock(spec=OpenClawClient)
|
|
|
|
# Make create_cron return unique IDs
|
|
counter = {"value": 0}
|
|
|
|
async def create_cron_impl(**kwargs):
|
|
counter["value"] += 1
|
|
return f"cron-{counter['value']}"
|
|
|
|
client.create_cron = AsyncMock(side_effect=create_cron_impl)
|
|
client.delete_cron = AsyncMock()
|
|
client.list_crons = AsyncMock(return_value=[])
|
|
return client
|
|
|
|
|
|
async def test_initialize_pool_empty(db_session: Session, mock_openclaw_client):
|
|
"""Test pool initialization when pool is empty."""
|
|
created = await cron_pool.initialize_pool(db_session, mock_openclaw_client)
|
|
assert created == cron_pool.MIN_POOL_SIZE
|
|
|
|
# Verify crons created in DB
|
|
crons = db_session.exec(select(OpenClawCronPoolDB)).all()
|
|
assert len(crons) == cron_pool.MIN_POOL_SIZE
|
|
assert all(c.status == "idle" for c in crons)
|
|
|
|
|
|
async def test_initialize_pool_already_exists(db_session: Session, mock_openclaw_client):
|
|
"""Test pool initialization when pool already exists."""
|
|
# Create existing cron
|
|
existing = OpenClawCronPoolDB(
|
|
openclaw_cron_id="existing-cron",
|
|
status="idle",
|
|
last_active_at=utc_now(),
|
|
)
|
|
db_session.add(existing)
|
|
db_session.commit()
|
|
|
|
created = await cron_pool.initialize_pool(db_session, mock_openclaw_client)
|
|
assert created == 0
|
|
|
|
# Verify no new crons created
|
|
crons = db_session.exec(select(OpenClawCronPoolDB)).all()
|
|
assert len(crons) == 1
|
|
|
|
|
|
async def test_scale_up(db_session: Session, mock_openclaw_client):
|
|
"""Test scaling up the pool."""
|
|
created = await cron_pool.scale_up(3, db_session, mock_openclaw_client)
|
|
assert created == 3
|
|
|
|
crons = db_session.exec(select(OpenClawCronPoolDB)).all()
|
|
assert len(crons) == 3
|
|
|
|
|
|
async def test_scale_up_max_limit(db_session: Session, mock_openclaw_client):
|
|
"""Test scaling up respects max pool size."""
|
|
# Fill pool to max
|
|
for i in range(cron_pool.MAX_POOL_SIZE):
|
|
cron = OpenClawCronPoolDB(
|
|
openclaw_cron_id=f"cron-{i}",
|
|
status="idle",
|
|
last_active_at=utc_now(),
|
|
)
|
|
db_session.add(cron)
|
|
db_session.commit()
|
|
|
|
created = await cron_pool.scale_up(5, db_session, mock_openclaw_client)
|
|
assert created == 0
|
|
|
|
|
|
async def test_scale_down(db_session: Session, mock_openclaw_client):
|
|
"""Test scaling down the pool."""
|
|
# Create 10 idle crons
|
|
for i in range(10):
|
|
cron = OpenClawCronPoolDB(
|
|
openclaw_cron_id=f"cron-{i}",
|
|
status="idle",
|
|
last_active_at=utc_now(),
|
|
)
|
|
db_session.add(cron)
|
|
db_session.commit()
|
|
|
|
deleted = await cron_pool.scale_down(3, db_session, mock_openclaw_client)
|
|
assert deleted == 3
|
|
|
|
crons = db_session.exec(select(OpenClawCronPoolDB)).all()
|
|
assert len(crons) == 7
|
|
|
|
|
|
async def test_scale_down_min_limit(db_session: Session, mock_openclaw_client):
|
|
"""Test scaling down respects min pool size."""
|
|
# Create exactly MIN_POOL_SIZE crons
|
|
for i in range(cron_pool.MIN_POOL_SIZE):
|
|
cron = OpenClawCronPoolDB(
|
|
openclaw_cron_id=f"cron-{i}",
|
|
status="idle",
|
|
last_active_at=utc_now(),
|
|
)
|
|
db_session.add(cron)
|
|
db_session.commit()
|
|
|
|
deleted = await cron_pool.scale_down(3, db_session, mock_openclaw_client)
|
|
assert deleted == 0
|
|
|
|
|
|
async def test_scale_down_only_idle(db_session: Session, mock_openclaw_client):
|
|
"""Test scaling down only deletes idle crons."""
|
|
# Create 5 idle and 5 busy crons
|
|
for i in range(5):
|
|
idle_cron = OpenClawCronPoolDB(
|
|
openclaw_cron_id=f"idle-{i}",
|
|
status="idle",
|
|
last_active_at=utc_now(),
|
|
)
|
|
busy_cron = OpenClawCronPoolDB(
|
|
openclaw_cron_id=f"busy-{i}",
|
|
status="busy",
|
|
last_active_at=utc_now(),
|
|
)
|
|
db_session.add_all([idle_cron, busy_cron])
|
|
db_session.commit()
|
|
|
|
deleted = await cron_pool.scale_down(3, db_session, mock_openclaw_client)
|
|
assert deleted == 3
|
|
|
|
# Verify only idle crons deleted
|
|
remaining = db_session.exec(select(OpenClawCronPoolDB)).all()
|
|
assert len(remaining) == 7
|
|
assert sum(1 for c in remaining if c.status == "busy") == 5
|
|
assert sum(1 for c in remaining if c.status == "idle") == 2
|
|
|
|
|
|
async def test_auto_scale_up(db_session: Session, mock_openclaw_client):
|
|
"""Test auto-scaling up when busy/total > 0.8."""
|
|
# Create 10 crons, 9 busy (90% > 80%)
|
|
for i in range(10):
|
|
status = "busy" if i < 9 else "idle"
|
|
cron = OpenClawCronPoolDB(
|
|
openclaw_cron_id=f"existing-cron-{i}",
|
|
status=status,
|
|
last_active_at=utc_now(),
|
|
)
|
|
db_session.add(cron)
|
|
db_session.commit()
|
|
|
|
scaled_up, scaled_down = await cron_pool.auto_scale(db_session, mock_openclaw_client)
|
|
assert scaled_up == 1
|
|
assert scaled_down == 0
|
|
|
|
|
|
async def test_auto_scale_down(db_session: Session, mock_openclaw_client):
|
|
"""Test auto-scaling down when idle > min_size * 2."""
|
|
# Create 15 idle crons (15 > 5 * 2)
|
|
for i in range(15):
|
|
cron = OpenClawCronPoolDB(
|
|
openclaw_cron_id=f"cron-{i}",
|
|
status="idle",
|
|
last_active_at=utc_now(),
|
|
)
|
|
db_session.add(cron)
|
|
db_session.commit()
|
|
|
|
scaled_up, scaled_down = await cron_pool.auto_scale(db_session, mock_openclaw_client)
|
|
assert scaled_up == 0
|
|
assert scaled_down == 1
|
|
|
|
|
|
def test_get_pool_status(db_session: Session):
|
|
"""Test getting pool status."""
|
|
# Create mixed crons
|
|
for i in range(10):
|
|
status = ["idle", "busy", "stuck"][i % 3]
|
|
cron = OpenClawCronPoolDB(
|
|
openclaw_cron_id=f"cron-{i}",
|
|
status=status,
|
|
last_active_at=utc_now(),
|
|
)
|
|
db_session.add(cron)
|
|
db_session.commit()
|
|
|
|
status = cron_pool.get_pool_status(db_session)
|
|
assert status["total"] == 10
|
|
assert status["idle"] == 4 # 0, 3, 6, 9
|
|
assert status["busy"] == 3 # 1, 4, 7
|
|
assert status["stuck"] == 3 # 2, 5, 8
|
|
assert status["min_size"] == cron_pool.MIN_POOL_SIZE
|
|
assert status["max_size"] == cron_pool.MAX_POOL_SIZE
|
|
|
|
|
|
async def test_sync_cron_states(db_session: Session, mock_openclaw_client):
|
|
"""Test syncing cron states from OpenClaw."""
|
|
# Mock OpenClaw returning 2 crons
|
|
mock_openclaw_client.list_crons.return_value = [
|
|
OpenClawCron(id="cron-1", name="worker-1", schedule="* * * * *", enabled=True, state={"status": "busy"}),
|
|
OpenClawCron(id="cron-2", name="worker-2", schedule="* * * * *", enabled=True, state={"status": "idle"}),
|
|
]
|
|
|
|
synced = await cron_pool.sync_cron_states(db_session, mock_openclaw_client)
|
|
assert synced == 2
|
|
|
|
# Verify crons added to DB
|
|
crons = db_session.exec(select(OpenClawCronPoolDB)).all()
|
|
assert len(crons) == 2
|
|
|
|
|
|
def test_detect_stuck_crons(db_session: Session):
|
|
"""Test detecting stuck crons."""
|
|
# Create crons with different last_active_at
|
|
now = utc_now()
|
|
active_cron = OpenClawCronPoolDB(
|
|
openclaw_cron_id="active",
|
|
status="busy",
|
|
last_active_at=now,
|
|
)
|
|
stuck_cron = OpenClawCronPoolDB(
|
|
openclaw_cron_id="stuck",
|
|
status="busy",
|
|
last_active_at=now - timedelta(minutes=15), # 15 minutes ago
|
|
)
|
|
idle_cron = OpenClawCronPoolDB(
|
|
openclaw_cron_id="idle",
|
|
status="idle",
|
|
last_active_at=now - timedelta(minutes=20),
|
|
)
|
|
db_session.add_all([active_cron, stuck_cron, idle_cron])
|
|
db_session.commit()
|
|
|
|
stuck = cron_pool.detect_stuck_crons(db_session)
|
|
assert len(stuck) == 1
|
|
assert stuck[0].openclaw_cron_id == "stuck"
|
|
|
|
|
|
async def test_handle_stuck_cron(db_session: Session, mock_openclaw_client):
|
|
"""Test handling a stuck cron."""
|
|
from agenteval.storage.db import IntelligentEvalTaskQueueDB
|
|
|
|
# Create stuck cron with assigned task
|
|
cron = OpenClawCronPoolDB(
|
|
openclaw_cron_id="stuck-cron",
|
|
status="busy",
|
|
current_eval_id="eval-1",
|
|
last_active_at=utc_now() - timedelta(minutes=15),
|
|
)
|
|
db_session.add(cron)
|
|
|
|
task = IntelligentEvalTaskQueueDB(
|
|
eval_id="eval-1",
|
|
status="assigned",
|
|
priority=1,
|
|
reason="slot_due",
|
|
assigned_cron_id="stuck-cron",
|
|
)
|
|
db_session.add(task)
|
|
db_session.commit()
|
|
|
|
# Handle stuck cron
|
|
await cron_pool.handle_stuck_cron(cron, db_session, mock_openclaw_client)
|
|
|
|
# Verify task marked as failed
|
|
db_session.refresh(task)
|
|
assert task.status == "failed"
|
|
assert task.error == "Cron stuck"
|
|
|
|
# Verify new task created for retry
|
|
new_task = db_session.exec(
|
|
select(IntelligentEvalTaskQueueDB).where(
|
|
IntelligentEvalTaskQueueDB.eval_id == "eval-1",
|
|
IntelligentEvalTaskQueueDB.status == "pending",
|
|
)
|
|
).first()
|
|
assert new_task is not None
|
|
assert new_task.reason == "cron_stuck_retry"
|
|
assert new_task.priority == 1
|
|
|
|
# Verify cron deleted
|
|
deleted_cron = db_session.exec(
|
|
select(OpenClawCronPoolDB).where(OpenClawCronPoolDB.openclaw_cron_id == "stuck-cron")
|
|
).first()
|
|
assert deleted_cron is None
|