""" tests/unit/test_device_manager.py Unit tests for DeviceManager, DeviceInfo, DeviceHealth. TDD: Red phase — tests written before implementation. """ from unittest.mock import MagicMock import pytest from core.device.device_manager import DeviceInfo, DeviceHealth, DeviceManager # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- def _make_mock_kp_api(devices): """Build a minimal kp API mock whose scan_devices() returns a descriptor list.""" descriptor_list = MagicMock() descriptor_list.device_descriptor_number = len(devices) mock_descs = [] for d in devices: desc = MagicMock() desc.usb_port_id = d["port_id"] desc.product_id = d["product_id"] desc.kn_number = d.get("kn_number", 0) mock_descs.append(desc) descriptor_list.device_descriptor_list = mock_descs kp_api = MagicMock() kp_api.core.scan_devices.return_value = descriptor_list return kp_api @pytest.fixture def two_device_kp(): """Mock kp API returning one KL520 and one KL720.""" return _make_mock_kp_api([ {"port_id": 1, "product_id": 0x100}, # KL520 {"port_id": 2, "product_id": 0x720}, # KL720 ]) @pytest.fixture def empty_kp(): """Mock kp API returning no devices.""" descriptor_list = MagicMock() descriptor_list.device_descriptor_number = 0 descriptor_list.device_descriptor_list = [] kp_api = MagicMock() kp_api.core.scan_devices.return_value = descriptor_list return kp_api # --------------------------------------------------------------------------- # DeviceInfo dataclass # --------------------------------------------------------------------------- class TestDeviceInfo: def test_fields_accessible(self): info = DeviceInfo( device_id="usb-1", series="KL520", product_id=0x100, status="online", gops=2, assigned_stage=None, current_fps=0.0, utilization_pct=0.0, ) assert info.device_id == "usb-1" assert info.series == "KL520" assert info.product_id == 0x100 assert info.status == "online" assert info.gops == 2 assert info.assigned_stage is None assert info.current_fps == 0.0 assert info.utilization_pct == 0.0 # --------------------------------------------------------------------------- # DeviceHealth dataclass # --------------------------------------------------------------------------- class TestDeviceHealth: def test_fields_accessible(self): health = DeviceHealth( device_id="usb-1", temperature_celsius=None, error_count=0, last_error=None, uptime_seconds=120.0, ) assert health.device_id == "usb-1" assert health.temperature_celsius is None assert health.error_count == 0 assert health.last_error is None assert health.uptime_seconds == 120.0 # --------------------------------------------------------------------------- # DeviceManager.scan_devices # --------------------------------------------------------------------------- class TestScanDevices: def test_returns_list_of_device_info(self, two_device_kp): mgr = DeviceManager(kp_api=two_device_kp) devices = mgr.scan_devices() assert isinstance(devices, list) assert len(devices) == 2 assert all(isinstance(d, DeviceInfo) for d in devices) def test_kl520_properties(self, two_device_kp): mgr = DeviceManager(kp_api=two_device_kp) devices = mgr.scan_devices() kl520 = next(d for d in devices if d.series == "KL520") assert kl520.product_id == 0x100 assert kl520.gops == 2 assert kl520.status == "online" def test_kl720_properties(self, two_device_kp): mgr = DeviceManager(kp_api=two_device_kp) devices = mgr.scan_devices() kl720 = next(d for d in devices if d.series == "KL720") assert kl720.product_id == 0x720 assert kl720.gops == 28 assert kl720.status == "online" def test_empty_returns_empty_list(self, empty_kp): mgr = DeviceManager(kp_api=empty_kp) devices = mgr.scan_devices() assert devices == [] def test_device_id_uses_port(self, two_device_kp): mgr = DeviceManager(kp_api=two_device_kp) devices = mgr.scan_devices() ids = {d.device_id for d in devices} assert "usb-1" in ids assert "usb-2" in ids # --------------------------------------------------------------------------- # DeviceManager.assign_device / unassign_device # --------------------------------------------------------------------------- class TestAssignDevice: def test_assign_online_device_returns_true(self, two_device_kp): mgr = DeviceManager(kp_api=two_device_kp) mgr.scan_devices() result = mgr.assign_device("usb-1", "stage-A") assert result is True def test_assigned_device_shows_stage(self, two_device_kp): mgr = DeviceManager(kp_api=two_device_kp) mgr.scan_devices() mgr.assign_device("usb-1", "stage-A") devices = mgr.get_device_statistics() assert devices["usb-1"].assigned_stage == "stage-A" def test_assign_already_assigned_device_returns_false(self, two_device_kp): mgr = DeviceManager(kp_api=two_device_kp) mgr.scan_devices() mgr.assign_device("usb-1", "stage-A") result = mgr.assign_device("usb-1", "stage-B") assert result is False def test_assign_unknown_device_returns_false(self, two_device_kp): mgr = DeviceManager(kp_api=two_device_kp) mgr.scan_devices() result = mgr.assign_device("usb-99", "stage-A") assert result is False def test_unassign_frees_device(self, two_device_kp): mgr = DeviceManager(kp_api=two_device_kp) mgr.scan_devices() mgr.assign_device("usb-1", "stage-A") result = mgr.unassign_device("usb-1") assert result is True devices = mgr.get_device_statistics() assert devices["usb-1"].assigned_stage is None def test_unassign_unknown_device_returns_false(self, two_device_kp): mgr = DeviceManager(kp_api=two_device_kp) mgr.scan_devices() result = mgr.unassign_device("usb-99") assert result is False def test_reassign_after_unassign_succeeds(self, two_device_kp): mgr = DeviceManager(kp_api=two_device_kp) mgr.scan_devices() mgr.assign_device("usb-1", "stage-A") mgr.unassign_device("usb-1") result = mgr.assign_device("usb-1", "stage-B") assert result is True def test_should_reject_assignment_for_offline_device(self): """assign_device returns False when the device status is offline.""" kp_api = _make_mock_kp_api([{"port_id": 5, "product_id": 0x100}]) mgr = DeviceManager(kp_api=kp_api) mgr.scan_devices() mgr._devices["usb-5"].status = "offline" result = mgr.assign_device("usb-5", "stage-A") assert result is False def test_should_allow_reassignment_to_same_stage(self, two_device_kp): """Assigning a device to the same stage twice is idempotent and returns True.""" mgr = DeviceManager(kp_api=two_device_kp) mgr.scan_devices() mgr.assign_device("usb-1", "stage-A") result = mgr.assign_device("usb-1", "stage-A") assert result is True def test_should_reject_reassignment_to_different_stage(self, two_device_kp): """Assigning a device already assigned to a different stage returns False.""" mgr = DeviceManager(kp_api=two_device_kp) mgr.scan_devices() mgr.assign_device("usb-1", "stage-A") result = mgr.assign_device("usb-1", "stage-B") assert result is False # --------------------------------------------------------------------------- # DeviceManager.get_load_balance_recommendation # --------------------------------------------------------------------------- class TestLoadBalanceRecommendation: def test_returns_dict_mapping_stage_to_device(self, two_device_kp): mgr = DeviceManager(kp_api=two_device_kp) mgr.scan_devices() rec = mgr.get_load_balance_recommendation(["stage-A", "stage-B"]) assert isinstance(rec, dict) assert "stage-A" in rec assert "stage-B" in rec def test_high_gops_assigned_to_first_stage(self, two_device_kp): """KL720 (28 GOPS) should be recommended for the first stage.""" mgr = DeviceManager(kp_api=two_device_kp) mgr.scan_devices() rec = mgr.get_load_balance_recommendation(["stage-A", "stage-B"]) # The device recommended for stage-A should be the higher-gops one stats = mgr.get_device_statistics() first_device_id = rec["stage-A"] assert stats[first_device_id].gops == 28 # KL720 def test_recommendation_with_more_stages_than_devices(self, two_device_kp): """Extra stages beyond available devices map to empty string.""" mgr = DeviceManager(kp_api=two_device_kp) mgr.scan_devices() rec = mgr.get_load_balance_recommendation(["s1", "s2", "s3"]) assert rec["s3"] == "" def test_recommendation_with_no_devices(self, empty_kp): mgr = DeviceManager(kp_api=empty_kp) mgr.scan_devices() rec = mgr.get_load_balance_recommendation(["stage-A"]) assert rec["stage-A"] == "" # --------------------------------------------------------------------------- # DeviceManager.get_device_health # --------------------------------------------------------------------------- class TestGetDeviceHealth: def test_returns_device_health(self, two_device_kp): mgr = DeviceManager(kp_api=two_device_kp) mgr.scan_devices() health = mgr.get_device_health("usb-1") assert isinstance(health, DeviceHealth) assert health.device_id == "usb-1" assert health.temperature_celsius is None # SDK does not support it assert health.error_count == 0 # --------------------------------------------------------------------------- # DeviceManager.get_device_statistics # --------------------------------------------------------------------------- class TestGetDeviceStatistics: def test_returns_all_known_devices(self, two_device_kp): mgr = DeviceManager(kp_api=two_device_kp) mgr.scan_devices() stats = mgr.get_device_statistics() assert isinstance(stats, dict) assert "usb-1" in stats assert "usb-2" in stats def test_values_are_device_info(self, two_device_kp): mgr = DeviceManager(kp_api=two_device_kp) mgr.scan_devices() stats = mgr.get_device_statistics() assert all(isinstance(v, DeviceInfo) for v in stats.values())