128 lines
4.4 KiB
Python
128 lines
4.4 KiB
Python
"""Unit tests for cloud.store.CloudStore (task 2.5)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from cloud.pool import HostRegistration, PooledDevice
|
|
from cloud.store import CloudStore
|
|
|
|
|
|
def _pooled(
|
|
device_id: str,
|
|
host_id: str,
|
|
*,
|
|
status: str = "idle",
|
|
driver_type: str = "wda",
|
|
tags: list[str] | None = None,
|
|
) -> PooledDevice:
|
|
from datetime import UTC, datetime
|
|
|
|
return PooledDevice(
|
|
device_id=device_id,
|
|
host_id=host_id,
|
|
driver_type=driver_type,
|
|
status=status, # type: ignore[arg-type]
|
|
capability_tags=list(tags or []),
|
|
synced_at=datetime(2026, 1, 1, tzinfo=UTC),
|
|
)
|
|
|
|
|
|
def test_upsert_host_and_replace_devices_round_trip(tmp_path) -> None:
|
|
store = CloudStore(tmp_path / "cloud.sqlite3")
|
|
from datetime import UTC, datetime
|
|
|
|
ts = datetime(2026, 7, 6, 12, 0, tzinfo=UTC)
|
|
store.upsert_host("host-a", address="10.0.0.1:8000", last_seen_at=ts)
|
|
store.replace_host_devices(
|
|
"host-a",
|
|
[_pooled("dev-1", "host-a"), _pooled("dev-2", "host-a", status="busy")],
|
|
)
|
|
|
|
hosts = store.list_hosts()
|
|
assert len(hosts) == 1
|
|
assert hosts[0].host_id == "host-a"
|
|
assert hosts[0].address == "10.0.0.1:8000"
|
|
assert hosts[0].last_seen_at == ts
|
|
|
|
devices = store.list_devices()
|
|
assert {d.device_id for d in devices} == {"dev-1", "dev-2"}
|
|
by_id = {d.device_id: d for d in devices}
|
|
assert by_id["dev-1"].host_id == "host-a"
|
|
assert by_id["dev-2"].status == "busy"
|
|
|
|
fetched = store.get_device("dev-1")
|
|
assert fetched is not None
|
|
assert fetched.host_id == "host-a"
|
|
assert store.get_device("does-not-exist") is None
|
|
|
|
|
|
def test_second_sync_fully_replaces_host_devices(tmp_path) -> None:
|
|
store = CloudStore(tmp_path / "cloud.sqlite3")
|
|
from datetime import UTC, datetime
|
|
|
|
store.upsert_host("host-a", address=None, last_seen_at=datetime(2026, 1, 1, tzinfo=UTC))
|
|
store.replace_host_devices(
|
|
"host-a",
|
|
[_pooled("dev-1", "host-a"), _pooled("dev-2", "host-a"), _pooled("dev-3", "host-a")],
|
|
)
|
|
|
|
# Second sync: only dev-2 plus a new dev-4. dev-1/dev-3 must be gone.
|
|
store.replace_host_devices(
|
|
"host-a",
|
|
[_pooled("dev-2", "host-a"), _pooled("dev-4", "host-a")],
|
|
)
|
|
|
|
devices = store.list_devices()
|
|
assert {d.device_id for d in devices} == {"dev-2", "dev-4"}
|
|
assert all(d.host_id == "host-a" for d in devices)
|
|
|
|
|
|
def test_devices_from_two_hosts_coexist(tmp_path) -> None:
|
|
store = CloudStore(tmp_path / "cloud.sqlite3")
|
|
from datetime import UTC, datetime
|
|
|
|
store.upsert_host("host-a", address="a", last_seen_at=datetime(2026, 1, 1, tzinfo=UTC))
|
|
store.upsert_host("host-b", address="b", last_seen_at=datetime(2026, 1, 2, tzinfo=UTC))
|
|
store.replace_host_devices("host-a", [_pooled("a-dev-1", "host-a")])
|
|
store.replace_host_devices("host-b", [_pooled("b-dev-1", "host-b"), _pooled("b-dev-2", "host-b")])
|
|
|
|
devices = store.list_devices()
|
|
assert {d.device_id for d in devices} == {"a-dev-1", "b-dev-1", "b-dev-2"}
|
|
by_host = {d.device_id: d.host_id for d in devices}
|
|
assert by_host == {"a-dev-1": "host-a", "b-dev-1": "host-b", "b-dev-2": "host-b"}
|
|
|
|
# Replacing host-a's devices must not touch host-b.
|
|
store.replace_host_devices("host-a", [_pooled("a-dev-9", "host-a")])
|
|
devices = store.list_devices()
|
|
assert {d.device_id for d in devices} == {"a-dev-9", "b-dev-1", "b-dev-2"}
|
|
|
|
|
|
def test_upsert_host_preserves_address_when_none(tmp_path) -> None:
|
|
store = CloudStore(tmp_path / "cloud.sqlite3")
|
|
from datetime import UTC, datetime
|
|
|
|
store.upsert_host("host-a", address="10.0.0.1:8000", last_seen_at=datetime(2026, 1, 1, tzinfo=UTC))
|
|
# Subsequent sync with address=None should not clobber the existing address.
|
|
store.upsert_host("host-a", address=None, last_seen_at=datetime(2026, 1, 2, tzinfo=UTC))
|
|
|
|
host = store.get_host("host-a")
|
|
assert host is not None
|
|
assert host.address == "10.0.0.1:8000"
|
|
assert host.last_seen_at == datetime(2026, 1, 2, tzinfo=UTC)
|
|
|
|
|
|
def test_capability_tags_round_trip(tmp_path) -> None:
|
|
store = CloudStore(tmp_path / "cloud.sqlite3")
|
|
from datetime import UTC, datetime
|
|
|
|
store.upsert_host("host-a", address="a", last_seen_at=datetime(2026, 1, 1, tzinfo=UTC))
|
|
store.replace_host_devices(
|
|
"host-a",
|
|
[_pooled("dev-1", "host-a", tags=["ios", "physical"])],
|
|
)
|
|
|
|
device = store.get_device("dev-1")
|
|
assert device is not None
|
|
assert device.capability_tags == ["ios", "physical"]
|