72 lines
2.1 KiB
Python
72 lines
2.1 KiB
Python
"""Tests for the cloud DevicePool, focused on the mcp_busy flag plumbing."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from cloud.config import CloudConfig
|
|
from cloud.pool import DevicePool
|
|
from cloud.store import CloudStore
|
|
from core.models import Device
|
|
|
|
|
|
@pytest.fixture
|
|
def pool() -> DevicePool:
|
|
"""Build a fresh DevicePool backed by a temporary SQLite file."""
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
store = CloudStore(Path(tmp) / "cloud.sqlite3")
|
|
try:
|
|
yield DevicePool(store=store, config=CloudConfig())
|
|
finally:
|
|
store.close()
|
|
|
|
|
|
def _device(device_id: str, *, status: str = "idle") -> Device:
|
|
return Device(
|
|
id=device_id,
|
|
driver_type="wda",
|
|
status=status, # type: ignore[arg-type]
|
|
capability_tags=[],
|
|
)
|
|
|
|
|
|
def test_sync_host_devices_marks_mcp_busy_devices(pool: DevicePool) -> None:
|
|
"""When a host reports device-1 as MCP-busy, the pool PooledDevice for
|
|
device-1 has mcp_busy=True."""
|
|
pool.sync_host_devices(
|
|
"host-1",
|
|
[_device("device-1", status="idle")],
|
|
mcp_busy_device_ids=["device-1"],
|
|
)
|
|
devices = pool.list_devices()
|
|
busy = [d for d in devices if d.device_id == "device-1"]
|
|
assert len(busy) == 1
|
|
assert busy[0].mcp_busy is True
|
|
|
|
|
|
def test_sync_host_devices_default_mcp_busy_is_false(pool: DevicePool) -> None:
|
|
pool.sync_host_devices(
|
|
"host-1", [_device("device-1", status="idle")]
|
|
)
|
|
devices = pool.list_devices()
|
|
assert devices[0].mcp_busy is False
|
|
|
|
|
|
def test_sync_host_devices_clears_mcp_busy_on_next_sync(
|
|
pool: DevicePool,
|
|
) -> None:
|
|
"""MCP releases device -> next heartbeat without device in
|
|
mcp_busy_device_ids -> pool reflects mcp_busy=False."""
|
|
pool.sync_host_devices(
|
|
"host-1",
|
|
[_device("device-1", status="idle")],
|
|
mcp_busy_device_ids=["device-1"],
|
|
)
|
|
pool.sync_host_devices(
|
|
"host-1", [_device("device-1", status="idle")]
|
|
)
|
|
devices = pool.list_devices()
|
|
assert devices[0].mcp_busy is False |