"""Tests for the USB Discovery integration.""" import os import sys from unittest.mock import MagicMock, Mock, call, patch, sentinel import pytest from homeassistant.components import usb from homeassistant.const import EVENT_HOMEASSISTANT_STARTED, EVENT_HOMEASSISTANT_STOP from homeassistant.core import HomeAssistant from homeassistant.setup import async_setup_component from . import conbee_device, slae_sh_device from tests.typing import WebSocketGenerator @pytest.fixture(name="operating_system") def mock_operating_system(): """Mock running Home Assistant Operating system.""" with patch( "homeassistant.components.usb.system_info.async_get_system_info", return_value={ "hassio": True, "docker": True, }, ): yield @pytest.fixture(name="docker") def mock_docker(): """Mock running Home Assistant in docker container.""" with patch( "homeassistant.components.usb.system_info.async_get_system_info", return_value={ "hassio": False, "docker": True, }, ): yield @pytest.fixture(name="venv") def mock_venv(): """Mock running Home Assistant in a venv container.""" with patch( "homeassistant.components.usb.system_info.async_get_system_info", return_value={ "hassio": False, "docker": False, "virtualenv": True, }, ): yield @pytest.mark.skipif( not sys.platform.startswith("linux"), reason="Only works on linux", ) async def test_observer_discovery( hass: HomeAssistant, hass_ws_client: WebSocketGenerator, venv ) -> None: """Test that observer can discover a device without raising an exception.""" new_usb = [{"domain": "test1", "vid": "3039"}] mock_comports = [ MagicMock( device=slae_sh_device.device, vid=12345, pid=12345, serial_number=slae_sh_device.serial_number, manufacturer=slae_sh_device.manufacturer, description=slae_sh_device.description, ) ] mock_observer = None async def _mock_monitor_observer_callback(callback): await hass.async_add_executor_job( callback, MagicMock(action="create", device_path="/dev/new") ) def _create_mock_monitor_observer(monitor, callback, name): nonlocal mock_observer hass.async_create_task(_mock_monitor_observer_callback(callback)) mock_observer = MagicMock() return mock_observer with patch("pyudev.Context"), patch( "pyudev.MonitorObserver", new=_create_mock_monitor_observer ), patch("pyudev.Monitor.filter_by"), patch( "homeassistant.components.usb.async_get_usb", return_value=new_usb ), patch( "homeassistant.components.usb.comports", return_value=mock_comports ), patch.object( hass.config_entries.flow, "async_init" ) as mock_config_flow: assert await async_setup_component(hass, "usb", {"usb": {}}) await hass.async_block_till_done() hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED) await hass.async_block_till_done() assert len(mock_config_flow.mock_calls) == 1 assert mock_config_flow.mock_calls[0][1][0] == "test1" hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP) await hass.async_block_till_done() assert mock_observer.mock_calls == [call.start(), call.stop()] @pytest.mark.skipif( not sys.platform.startswith("linux"), reason="Only works on linux", ) async def test_removal_by_observer_before_started( hass: HomeAssistant, operating_system ) -> None: """Test a device is removed by the observer before started.""" async def _mock_monitor_observer_callback(callback): await hass.async_add_executor_job( callback, MagicMock(action="remove", device_path="/dev/new") ) def _create_mock_monitor_observer(monitor, callback, name): hass.async_create_task(_mock_monitor_observer_callback(callback)) new_usb = [{"domain": "test1", "vid": "3039", "pid": "3039"}] mock_comports = [ MagicMock( device=slae_sh_device.device, vid=12345, pid=12345, serial_number=slae_sh_device.serial_number, manufacturer=slae_sh_device.manufacturer, description=slae_sh_device.description, ) ] with patch( "homeassistant.components.usb.async_get_usb", return_value=new_usb ), patch( "homeassistant.components.usb.comports", return_value=mock_comports ), patch( "pyudev.MonitorObserver", new=_create_mock_monitor_observer ), patch.object( hass.config_entries.flow, "async_init" ) as mock_config_flow: assert await async_setup_component(hass, "usb", {"usb": {}}) await hass.async_block_till_done() with patch("homeassistant.components.usb.comports", return_value=[]): hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED) await hass.async_block_till_done() assert len(mock_config_flow.mock_calls) == 0 hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP) await hass.async_block_till_done() async def test_discovered_by_websocket_scan( hass: HomeAssistant, hass_ws_client: WebSocketGenerator ) -> None: """Test a device is discovered from websocket scan.""" new_usb = [{"domain": "test1", "vid": "3039", "pid": "3039"}] mock_comports = [ MagicMock( device=slae_sh_device.device, vid=12345, pid=12345, serial_number=slae_sh_device.serial_number, manufacturer=slae_sh_device.manufacturer, description=slae_sh_device.description, ) ] with patch("pyudev.Context", side_effect=ImportError), patch( "homeassistant.components.usb.async_get_usb", return_value=new_usb ), patch( "homeassistant.components.usb.comports", return_value=mock_comports ), patch.object( hass.config_entries.flow, "async_init" ) as mock_config_flow: assert await async_setup_component(hass, "usb", {"usb": {}}) await hass.async_block_till_done() hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED) await hass.async_block_till_done() ws_client = await hass_ws_client(hass) await ws_client.send_json({"id": 1, "type": "usb/scan"}) response = await ws_client.receive_json() assert response["success"] await hass.async_block_till_done() assert len(mock_config_flow.mock_calls) == 1 assert mock_config_flow.mock_calls[0][1][0] == "test1" async def test_discovered_by_websocket_scan_limited_by_description_matcher( hass: HomeAssistant, hass_ws_client: WebSocketGenerator ) -> None: """Test a device is discovered from websocket scan is limited by the description matcher.""" new_usb = [ {"domain": "test1", "vid": "3039", "pid": "3039", "description": "*2652*"} ] mock_comports = [ MagicMock( device=slae_sh_device.device, vid=12345, pid=12345, serial_number=slae_sh_device.serial_number, manufacturer=slae_sh_device.manufacturer, description=slae_sh_device.description, ) ] with patch("pyudev.Context", side_effect=ImportError), patch( "homeassistant.components.usb.async_get_usb", return_value=new_usb ), patch( "homeassistant.components.usb.comports", return_value=mock_comports ), patch.object( hass.config_entries.flow, "async_init" ) as mock_config_flow: assert await async_setup_component(hass, "usb", {"usb": {}}) await hass.async_block_till_done() hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED) await hass.async_block_till_done() ws_client = await hass_ws_client(hass) await ws_client.send_json({"id": 1, "type": "usb/scan"}) response = await ws_client.receive_json() assert response["success"] await hass.async_block_till_done() assert len(mock_config_flow.mock_calls) == 1 assert mock_config_flow.mock_calls[0][1][0] == "test1" async def test_most_targeted_matcher_wins( hass: HomeAssistant, hass_ws_client: WebSocketGenerator ) -> None: """Test that the most targeted matcher is used.""" new_usb = [ {"domain": "less", "vid": "3039", "pid": "3039"}, {"domain": "more", "vid": "3039", "pid": "3039", "description": "*2652*"}, ] mock_comports = [ MagicMock( device=slae_sh_device.device, vid=12345, pid=12345, serial_number=slae_sh_device.serial_number, manufacturer=slae_sh_device.manufacturer, description=slae_sh_device.description, ) ] with patch("pyudev.Context", side_effect=ImportError), patch( "homeassistant.components.usb.async_get_usb", return_value=new_usb ), patch( "homeassistant.components.usb.comports", return_value=mock_comports ), patch.object( hass.config_entries.flow, "async_init" ) as mock_config_flow: assert await async_setup_component(hass, "usb", {"usb": {}}) await hass.async_block_till_done() hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED) await hass.async_block_till_done() ws_client = await hass_ws_client(hass) await ws_client.send_json({"id": 1, "type": "usb/scan"}) response = await ws_client.receive_json() assert response["success"] await hass.async_block_till_done() assert len(mock_config_flow.mock_calls) == 1 assert mock_config_flow.mock_calls[0][1][0] == "more" async def test_discovered_by_websocket_scan_rejected_by_description_matcher( hass: HomeAssistant, hass_ws_client: WebSocketGenerator ) -> None: """Test a device is discovered from websocket scan rejected by the description matcher.""" new_usb = [ {"domain": "test1", "vid": "3039", "pid": "3039", "description": "*not_it*"} ] mock_comports = [ MagicMock( device=slae_sh_device.device, vid=12345, pid=12345, serial_number=slae_sh_device.serial_number, manufacturer=slae_sh_device.manufacturer, description=slae_sh_device.description, ) ] with patch("pyudev.Context", side_effect=ImportError), patch( "homeassistant.components.usb.async_get_usb", return_value=new_usb ), patch( "homeassistant.components.usb.comports", return_value=mock_comports ), patch.object( hass.config_entries.flow, "async_init" ) as mock_config_flow: assert await async_setup_component(hass, "usb", {"usb": {}}) await hass.async_block_till_done() hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED) await hass.async_block_till_done() ws_client = await hass_ws_client(hass) await ws_client.send_json({"id": 1, "type": "usb/scan"}) response = await ws_client.receive_json() assert response["success"] await hass.async_block_till_done() assert len(mock_config_flow.mock_calls) == 0 async def test_discovered_by_websocket_scan_limited_by_serial_number_matcher( hass: HomeAssistant, hass_ws_client: WebSocketGenerator ) -> None: """Test a device is discovered from websocket scan is limited by the serial_number matcher.""" new_usb = [ { "domain": "test1", "vid": "3039", "pid": "3039", "serial_number": "00_12_4b_00*", } ] mock_comports = [ MagicMock( device=slae_sh_device.device, vid=12345, pid=12345, serial_number=slae_sh_device.serial_number, manufacturer=slae_sh_device.manufacturer, description=slae_sh_device.description, ) ] with patch("pyudev.Context", side_effect=ImportError), patch( "homeassistant.components.usb.async_get_usb", return_value=new_usb ), patch( "homeassistant.components.usb.comports", return_value=mock_comports ), patch.object( hass.config_entries.flow, "async_init" ) as mock_config_flow: assert await async_setup_component(hass, "usb", {"usb": {}}) await hass.async_block_till_done() hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED) await hass.async_block_till_done() ws_client = await hass_ws_client(hass) await ws_client.send_json({"id": 1, "type": "usb/scan"}) response = await ws_client.receive_json() assert response["success"] await hass.async_block_till_done() assert len(mock_config_flow.mock_calls) == 1 assert mock_config_flow.mock_calls[0][1][0] == "test1" async def test_discovered_by_websocket_scan_rejected_by_serial_number_matcher( hass: HomeAssistant, hass_ws_client: WebSocketGenerator ) -> None: """Test a device is discovered from websocket scan is rejected by the serial_number matcher.""" new_usb = [ {"domain": "test1", "vid": "3039", "pid": "3039", "serial_number": "123*"} ] mock_comports = [ MagicMock( device=slae_sh_device.device, vid=12345, pid=12345, serial_number=slae_sh_device.serial_number, manufacturer=slae_sh_device.manufacturer, description=slae_sh_device.description, ) ] with patch("pyudev.Context", side_effect=ImportError), patch( "homeassistant.components.usb.async_get_usb", return_value=new_usb ), patch( "homeassistant.components.usb.comports", return_value=mock_comports ), patch.object( hass.config_entries.flow, "async_init" ) as mock_config_flow: assert await async_setup_component(hass, "usb", {"usb": {}}) await hass.async_block_till_done() hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED) await hass.async_block_till_done() ws_client = await hass_ws_client(hass) await ws_client.send_json({"id": 1, "type": "usb/scan"}) response = await ws_client.receive_json() assert response["success"] await hass.async_block_till_done() assert len(mock_config_flow.mock_calls) == 0 async def test_discovered_by_websocket_scan_limited_by_manufacturer_matcher( hass: HomeAssistant, hass_ws_client: WebSocketGenerator ) -> None: """Test a device is discovered from websocket scan is limited by the manufacturer matcher.""" new_usb = [ { "domain": "test1", "vid": "3039", "pid": "3039", "manufacturer": "dresden elektronik ingenieurtechnik*", } ] mock_comports = [ MagicMock( device=conbee_device.device, vid=12345, pid=12345, serial_number=conbee_device.serial_number, manufacturer=conbee_device.manufacturer, description=conbee_device.description, ) ] with patch("pyudev.Context", side_effect=ImportError), patch( "homeassistant.components.usb.async_get_usb", return_value=new_usb ), patch( "homeassistant.components.usb.comports", return_value=mock_comports ), patch.object( hass.config_entries.flow, "async_init" ) as mock_config_flow: assert await async_setup_component(hass, "usb", {"usb": {}}) await hass.async_block_till_done() hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED) await hass.async_block_till_done() ws_client = await hass_ws_client(hass) await ws_client.send_json({"id": 1, "type": "usb/scan"}) response = await ws_client.receive_json() assert response["success"] await hass.async_block_till_done() assert len(mock_config_flow.mock_calls) == 1 assert mock_config_flow.mock_calls[0][1][0] == "test1" async def test_discovered_by_websocket_scan_rejected_by_manufacturer_matcher( hass: HomeAssistant, hass_ws_client: WebSocketGenerator ) -> None: """Test a device is discovered from websocket scan is rejected by the manufacturer matcher.""" new_usb = [ { "domain": "test1", "vid": "3039", "pid": "3039", "manufacturer": "other vendor*", } ] mock_comports = [ MagicMock( device=conbee_device.device, vid=12345, pid=12345, serial_number=conbee_device.serial_number, manufacturer=conbee_device.manufacturer, description=conbee_device.description, ) ] with patch("pyudev.Context", side_effect=ImportError), patch( "homeassistant.components.usb.async_get_usb", return_value=new_usb ), patch( "homeassistant.components.usb.comports", return_value=mock_comports ), patch.object( hass.config_entries.flow, "async_init" ) as mock_config_flow: assert await async_setup_component(hass, "usb", {"usb": {}}) await hass.async_block_till_done() hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED) await hass.async_block_till_done() ws_client = await hass_ws_client(hass) await ws_client.send_json({"id": 1, "type": "usb/scan"}) response = await ws_client.receive_json() assert response["success"] await hass.async_block_till_done() assert len(mock_config_flow.mock_calls) == 0 async def test_discovered_by_websocket_rejected_with_empty_serial_number_only( hass: HomeAssistant, hass_ws_client: WebSocketGenerator ) -> None: """Test a device is discovered from websocket is rejected with empty serial number.""" new_usb = [ {"domain": "test1", "vid": "3039", "pid": "3039", "serial_number": "123*"} ] mock_comports = [ MagicMock( device=conbee_device.device, vid=12345, pid=12345, serial_number=None, manufacturer=None, description=None, ) ] with patch("pyudev.Context", side_effect=ImportError), patch( "homeassistant.components.usb.async_get_usb", return_value=new_usb ), patch( "homeassistant.components.usb.comports", return_value=mock_comports ), patch.object( hass.config_entries.flow, "async_init" ) as mock_config_flow: assert await async_setup_component(hass, "usb", {"usb": {}}) await hass.async_block_till_done() hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED) await hass.async_block_till_done() ws_client = await hass_ws_client(hass) await ws_client.send_json({"id": 1, "type": "usb/scan"}) response = await ws_client.receive_json() assert response["success"] await hass.async_block_till_done() assert len(mock_config_flow.mock_calls) == 0 async def test_discovered_by_websocket_scan_match_vid_only( hass: HomeAssistant, hass_ws_client: WebSocketGenerator ) -> None: """Test a device is discovered from websocket scan only matching vid.""" new_usb = [{"domain": "test1", "vid": "3039"}] mock_comports = [ MagicMock( device=slae_sh_device.device, vid=12345, pid=12345, serial_number=slae_sh_device.serial_number, manufacturer=slae_sh_device.manufacturer, description=slae_sh_device.description, ) ] with patch("pyudev.Context", side_effect=ImportError), patch( "homeassistant.components.usb.async_get_usb", return_value=new_usb ), patch( "homeassistant.components.usb.comports", return_value=mock_comports ), patch.object( hass.config_entries.flow, "async_init" ) as mock_config_flow: assert await async_setup_component(hass, "usb", {"usb": {}}) await hass.async_block_till_done() hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED) await hass.async_block_till_done() ws_client = await hass_ws_client(hass) await ws_client.send_json({"id": 1, "type": "usb/scan"}) response = await ws_client.receive_json() assert response["success"] await hass.async_block_till_done() assert len(mock_config_flow.mock_calls) == 1 assert mock_config_flow.mock_calls[0][1][0] == "test1" async def test_discovered_by_websocket_scan_match_vid_wrong_pid( hass: HomeAssistant, hass_ws_client: WebSocketGenerator ) -> None: """Test a device is discovered from websocket scan only matching vid but wrong pid.""" new_usb = [{"domain": "test1", "vid": "3039", "pid": "9999"}] mock_comports = [ MagicMock( device=slae_sh_device.device, vid=12345, pid=12345, serial_number=slae_sh_device.serial_number, manufacturer=slae_sh_device.manufacturer, description=slae_sh_device.description, ) ] with patch("pyudev.Context", side_effect=ImportError), patch( "homeassistant.components.usb.async_get_usb", return_value=new_usb ), patch( "homeassistant.components.usb.comports", return_value=mock_comports ), patch.object( hass.config_entries.flow, "async_init" ) as mock_config_flow: assert await async_setup_component(hass, "usb", {"usb": {}}) await hass.async_block_till_done() hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED) await hass.async_block_till_done() ws_client = await hass_ws_client(hass) await ws_client.send_json({"id": 1, "type": "usb/scan"}) response = await ws_client.receive_json() assert response["success"] await hass.async_block_till_done() assert len(mock_config_flow.mock_calls) == 0 async def test_discovered_by_websocket_no_vid_pid( hass: HomeAssistant, hass_ws_client: WebSocketGenerator ) -> None: """Test a device is discovered from websocket scan with no vid or pid.""" new_usb = [{"domain": "test1", "vid": "3039", "pid": "9999"}] mock_comports = [ MagicMock( device=slae_sh_device.device, vid=None, pid=None, serial_number=slae_sh_device.serial_number, manufacturer=slae_sh_device.manufacturer, description=slae_sh_device.description, ) ] with patch("pyudev.Context", side_effect=ImportError), patch( "homeassistant.components.usb.async_get_usb", return_value=new_usb ), patch( "homeassistant.components.usb.comports", return_value=mock_comports ), patch.object( hass.config_entries.flow, "async_init" ) as mock_config_flow: assert await async_setup_component(hass, "usb", {"usb": {}}) await hass.async_block_till_done() hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED) await hass.async_block_till_done() ws_client = await hass_ws_client(hass) await ws_client.send_json({"id": 1, "type": "usb/scan"}) response = await ws_client.receive_json() assert response["success"] await hass.async_block_till_done() assert len(mock_config_flow.mock_calls) == 0 @pytest.mark.parametrize("exception_type", [ImportError, OSError]) async def test_non_matching_discovered_by_scanner_after_started( hass: HomeAssistant, exception_type, hass_ws_client: WebSocketGenerator ) -> None: """Test a websocket scan that does not match.""" new_usb = [{"domain": "test1", "vid": "4444", "pid": "4444"}] mock_comports = [ MagicMock( device=slae_sh_device.device, vid=12345, pid=12345, serial_number=slae_sh_device.serial_number, manufacturer=slae_sh_device.manufacturer, description=slae_sh_device.description, ) ] with patch("pyudev.Context", side_effect=exception_type), patch( "homeassistant.components.usb.async_get_usb", return_value=new_usb ), patch( "homeassistant.components.usb.comports", return_value=mock_comports ), patch.object( hass.config_entries.flow, "async_init" ) as mock_config_flow: assert await async_setup_component(hass, "usb", {"usb": {}}) await hass.async_block_till_done() hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED) await hass.async_block_till_done() ws_client = await hass_ws_client(hass) await ws_client.send_json({"id": 1, "type": "usb/scan"}) response = await ws_client.receive_json() assert response["success"] await hass.async_block_till_done() assert len(mock_config_flow.mock_calls) == 0 @pytest.mark.skipif( not sys.platform.startswith("linux"), reason="Only works on linux", ) async def test_observer_on_wsl_fallback_without_throwing_exception( hass: HomeAssistant, hass_ws_client: WebSocketGenerator, venv ) -> None: """Test that observer on WSL failure results in fallback to scanning without raising an exception.""" new_usb = [{"domain": "test1", "vid": "3039"}] mock_comports = [ MagicMock( device=slae_sh_device.device, vid=12345, pid=12345, serial_number=slae_sh_device.serial_number, manufacturer=slae_sh_device.manufacturer, description=slae_sh_device.description, ) ] with patch("pyudev.Context"), patch( "pyudev.Monitor.filter_by", side_effect=ValueError ), patch("homeassistant.components.usb.async_get_usb", return_value=new_usb), patch( "homeassistant.components.usb.comports", return_value=mock_comports ), patch.object( hass.config_entries.flow, "async_init" ) as mock_config_flow: assert await async_setup_component(hass, "usb", {"usb": {}}) await hass.async_block_till_done() hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED) await hass.async_block_till_done() ws_client = await hass_ws_client(hass) await ws_client.send_json({"id": 1, "type": "usb/scan"}) response = await ws_client.receive_json() assert response["success"] await hass.async_block_till_done() assert len(mock_config_flow.mock_calls) == 1 assert mock_config_flow.mock_calls[0][1][0] == "test1" @pytest.mark.skipif( not sys.platform.startswith("linux"), reason="Only works on linux", ) async def test_not_discovered_by_observer_before_started_on_docker( hass: HomeAssistant, docker ) -> None: """Test a device is not discovered since observer is not running on bare docker.""" async def _mock_monitor_observer_callback(callback): await hass.async_add_executor_job( callback, MagicMock(action="add", device_path="/dev/new") ) def _create_mock_monitor_observer(monitor, callback, name): hass.async_create_task(_mock_monitor_observer_callback(callback)) return MagicMock() new_usb = [{"domain": "test1", "vid": "3039", "pid": "3039"}] mock_comports = [ MagicMock( device=slae_sh_device.device, vid=12345, pid=12345, serial_number=slae_sh_device.serial_number, manufacturer=slae_sh_device.manufacturer, description=slae_sh_device.description, ) ] with patch( "homeassistant.components.usb.async_get_usb", return_value=new_usb ), patch( "homeassistant.components.usb.comports", return_value=mock_comports ), patch( "pyudev.MonitorObserver", new=_create_mock_monitor_observer ): assert await async_setup_component(hass, "usb", {"usb": {}}) await hass.async_block_till_done() with patch("homeassistant.components.usb.comports", return_value=[]), patch.object( hass.config_entries.flow, "async_init" ) as mock_config_flow: hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED) await hass.async_block_till_done() assert len(mock_config_flow.mock_calls) == 0 def test_get_serial_by_id_no_dir() -> None: """Test serial by id conversion if there's no /dev/serial/by-id.""" p1 = patch("os.path.isdir", MagicMock(return_value=False)) p2 = patch("os.scandir") with p1 as is_dir_mock, p2 as scan_mock: res = usb.get_serial_by_id(sentinel.path) assert res is sentinel.path assert is_dir_mock.call_count == 1 assert scan_mock.call_count == 0 def test_get_serial_by_id() -> None: """Test serial by id conversion.""" p1 = patch("os.path.isdir", MagicMock(return_value=True)) p2 = patch("os.scandir") def _realpath(path): if path is sentinel.matched_link: return sentinel.path return sentinel.serial_link_path p3 = patch("os.path.realpath", side_effect=_realpath) with p1 as is_dir_mock, p2 as scan_mock, p3: res = usb.get_serial_by_id(sentinel.path) assert res is sentinel.path assert is_dir_mock.call_count == 1 assert scan_mock.call_count == 1 entry1 = MagicMock(spec_set=os.DirEntry) entry1.is_symlink.return_value = True entry1.path = sentinel.some_path entry2 = MagicMock(spec_set=os.DirEntry) entry2.is_symlink.return_value = False entry2.path = sentinel.other_path entry3 = MagicMock(spec_set=os.DirEntry) entry3.is_symlink.return_value = True entry3.path = sentinel.matched_link scan_mock.return_value = [entry1, entry2, entry3] res = usb.get_serial_by_id(sentinel.path) assert res is sentinel.matched_link assert is_dir_mock.call_count == 2 assert scan_mock.call_count == 2 def test_human_readable_device_name() -> None: """Test human readable device name includes the passed data.""" name = usb.human_readable_device_name( "/dev/null", "612020FD", "Silicon Labs", "HubZ Smart Home Controller - HubZ Z-Wave Com Port", "10C4", "8A2A", ) assert "/dev/null" in name assert "612020FD" in name assert "Silicon Labs" in name assert "HubZ Smart Home Controller - HubZ Z-Wave Com Port"[:26] in name assert "10C4" in name assert "8A2A" in name name = usb.human_readable_device_name( "/dev/null", "612020FD", "Silicon Labs", None, "10C4", "8A2A", ) assert "/dev/null" in name assert "612020FD" in name assert "Silicon Labs" in name assert "10C4" in name assert "8A2A" in name async def test_async_is_plugged_in( hass: HomeAssistant, hass_ws_client: WebSocketGenerator ) -> None: """Test async_is_plugged_in.""" new_usb = [{"domain": "test1", "vid": "3039", "pid": "3039"}] mock_comports = [ MagicMock( device=slae_sh_device.device, vid=12345, pid=12345, serial_number=slae_sh_device.serial_number, manufacturer=slae_sh_device.manufacturer, description=slae_sh_device.description, ) ] matcher = { "vid": "3039", "pid": "3039", } with patch("pyudev.Context", side_effect=ImportError), patch( "homeassistant.components.usb.async_get_usb", return_value=new_usb ), patch("homeassistant.components.usb.comports", return_value=[]), patch.object( hass.config_entries.flow, "async_init" ): assert await async_setup_component(hass, "usb", {"usb": {}}) await hass.async_block_till_done() hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED) await hass.async_block_till_done() assert not usb.async_is_plugged_in(hass, matcher) with patch( "homeassistant.components.usb.comports", return_value=mock_comports ), patch.object(hass.config_entries.flow, "async_init"): ws_client = await hass_ws_client(hass) await ws_client.send_json({"id": 1, "type": "usb/scan"}) response = await ws_client.receive_json() assert response["success"] await hass.async_block_till_done() assert usb.async_is_plugged_in(hass, matcher) @pytest.mark.parametrize( "matcher", [ {"vid": "abcd"}, {"pid": "123a"}, {"serial_number": "1234ABCD"}, {"manufacturer": "Some Manufacturer"}, {"description": "A description"}, ], ) async def test_async_is_plugged_in_case_enforcement( hass: HomeAssistant, matcher ) -> None: """Test `async_is_plugged_in` throws an error when incorrect cases are used.""" new_usb = [{"domain": "test1", "vid": "ABCD"}] with patch("pyudev.Context", side_effect=ImportError), patch( "homeassistant.components.usb.async_get_usb", return_value=new_usb ), patch("homeassistant.components.usb.comports", return_value=[]), patch.object( hass.config_entries.flow, "async_init" ): assert await async_setup_component(hass, "usb", {"usb": {}}) await hass.async_block_till_done() hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED) await hass.async_block_till_done() with pytest.raises(ValueError): usb.async_is_plugged_in(hass, matcher) async def test_web_socket_triggers_discovery_request_callbacks( hass: HomeAssistant, hass_ws_client: WebSocketGenerator ) -> None: """Test the websocket call triggers a discovery request callback.""" mock_callback = Mock() with patch("pyudev.Context", side_effect=ImportError), patch( "homeassistant.components.usb.async_get_usb", return_value=[] ), patch("homeassistant.components.usb.comports", return_value=[]), patch.object( hass.config_entries.flow, "async_init" ): assert await async_setup_component(hass, "usb", {"usb": {}}) await hass.async_block_till_done() hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED) await hass.async_block_till_done() cancel = usb.async_register_scan_request_callback(hass, mock_callback) ws_client = await hass_ws_client(hass) await ws_client.send_json({"id": 1, "type": "usb/scan"}) response = await ws_client.receive_json() assert response["success"] await hass.async_block_till_done() assert len(mock_callback.mock_calls) == 1 cancel() await ws_client.send_json({"id": 2, "type": "usb/scan"}) response = await ws_client.receive_json() assert response["success"] await hass.async_block_till_done() assert len(mock_callback.mock_calls) == 1 async def test_initial_scan_callback( hass: HomeAssistant, hass_ws_client: WebSocketGenerator ) -> None: """Test it's possible to register a callback when the initial scan is done.""" mock_callback_1 = Mock() mock_callback_2 = Mock() with patch("pyudev.Context", side_effect=ImportError), patch( "homeassistant.components.usb.async_get_usb", return_value=[] ), patch("homeassistant.components.usb.comports", return_value=[]), patch.object( hass.config_entries.flow, "async_init" ): assert await async_setup_component(hass, "usb", {"usb": {}}) cancel_1 = usb.async_register_initial_scan_callback(hass, mock_callback_1) assert len(mock_callback_1.mock_calls) == 0 await hass.async_block_till_done() assert len(mock_callback_1.mock_calls) == 0 # This triggers the initial scan hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED) await hass.async_block_till_done() assert len(mock_callback_1.mock_calls) == 1 # A callback registered now should be called immediately. The old callback # should not be called again cancel_2 = usb.async_register_initial_scan_callback(hass, mock_callback_2) assert len(mock_callback_1.mock_calls) == 1 assert len(mock_callback_2.mock_calls) == 1 # Calling the cancels should be allowed even if the callback has been called cancel_1() cancel_2() async def test_cancel_initial_scan_callback( hass: HomeAssistant, hass_ws_client: WebSocketGenerator ) -> None: """Test it's possible to cancel an initial scan callback.""" mock_callback = Mock() with patch("pyudev.Context", side_effect=ImportError), patch( "homeassistant.components.usb.async_get_usb", return_value=[] ), patch("homeassistant.components.usb.comports", return_value=[]), patch.object( hass.config_entries.flow, "async_init" ): assert await async_setup_component(hass, "usb", {"usb": {}}) cancel = usb.async_register_initial_scan_callback(hass, mock_callback) assert len(mock_callback.mock_calls) == 0 await hass.async_block_till_done() assert len(mock_callback.mock_calls) == 0 cancel() # This triggers the initial scan hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED) await hass.async_block_till_done() assert len(mock_callback.mock_calls) == 0