"""Test deCONZ gateway.""" import asyncio from copy import deepcopy from unittest.mock import patch import pydeconz from pydeconz.websocket import State import pytest from homeassistant.components import ssdp from homeassistant.components.alarm_control_panel import ( DOMAIN as ALARM_CONTROL_PANEL_DOMAIN, ) from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR_DOMAIN from homeassistant.components.button import DOMAIN as BUTTON_DOMAIN from homeassistant.components.climate import DOMAIN as CLIMATE_DOMAIN from homeassistant.components.cover import DOMAIN as COVER_DOMAIN from homeassistant.components.deconz.config_flow import DECONZ_MANUFACTURERURL from homeassistant.components.deconz.const import DOMAIN as DECONZ_DOMAIN from homeassistant.components.deconz.errors import AuthenticationRequired, CannotConnect from homeassistant.components.deconz.gateway import ( get_deconz_session, get_gateway_from_config_entry, ) from homeassistant.components.fan import DOMAIN as FAN_DOMAIN from homeassistant.components.light import DOMAIN as LIGHT_DOMAIN from homeassistant.components.lock import DOMAIN as LOCK_DOMAIN from homeassistant.components.number import DOMAIN as NUMBER_DOMAIN from homeassistant.components.scene import DOMAIN as SCENE_DOMAIN from homeassistant.components.select import DOMAIN as SELECT_DOMAIN from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN from homeassistant.components.siren import DOMAIN as SIREN_DOMAIN from homeassistant.components.ssdp import ( ATTR_UPNP_MANUFACTURER_URL, ATTR_UPNP_SERIAL, ATTR_UPNP_UDN, ) from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN from homeassistant.config_entries import SOURCE_HASSIO, SOURCE_SSDP, SOURCE_USER from homeassistant.const import ( CONF_API_KEY, CONF_HOST, CONF_PORT, CONTENT_TYPE_JSON, STATE_OFF, STATE_UNAVAILABLE, ) from homeassistant.core import HomeAssistant from homeassistant.helpers import device_registry as dr from tests.common import MockConfigEntry from tests.test_util.aiohttp import AiohttpClientMocker API_KEY = "1234567890ABCDEF" BRIDGEID = "01234E56789A" HOST = "1.2.3.4" PORT = 80 DEFAULT_URL = f"http://{HOST}:{PORT}/api/{API_KEY}" ENTRY_CONFIG = {CONF_API_KEY: API_KEY, CONF_HOST: HOST, CONF_PORT: PORT} ENTRY_OPTIONS = {} DECONZ_CONFIG = { "bridgeid": BRIDGEID, "ipaddress": HOST, "mac": "00:11:22:33:44:55", "modelid": "deCONZ", "name": "deCONZ mock gateway", "sw_version": "2.05.69", "uuid": "1234", "websocketport": 1234, } DECONZ_WEB_REQUEST = { "config": DECONZ_CONFIG, "groups": {}, "lights": {}, "sensors": {}, } def mock_deconz_request(aioclient_mock, config, data): """Mock a deCONZ get request.""" host = config[CONF_HOST] port = config[CONF_PORT] api_key = config[CONF_API_KEY] aioclient_mock.get( f"http://{host}:{port}/api/{api_key}", json=deepcopy(data), headers={"content-type": CONTENT_TYPE_JSON}, ) def mock_deconz_put_request(aioclient_mock, config, path): """Mock a deCONZ put request.""" host = config[CONF_HOST] port = config[CONF_PORT] api_key = config[CONF_API_KEY] aioclient_mock.put( f"http://{host}:{port}/api/{api_key}{path}", json={}, headers={"content-type": CONTENT_TYPE_JSON}, ) async def setup_deconz_integration( hass, aioclient_mock=None, *, config=ENTRY_CONFIG, options=ENTRY_OPTIONS, get_state_response=DECONZ_WEB_REQUEST, entry_id="1", unique_id=BRIDGEID, source=SOURCE_USER, ): """Create the deCONZ gateway.""" config_entry = MockConfigEntry( domain=DECONZ_DOMAIN, source=source, data=deepcopy(config), options=deepcopy(options), entry_id=entry_id, unique_id=unique_id, ) config_entry.add_to_hass(hass) if aioclient_mock: mock_deconz_request(aioclient_mock, config, get_state_response) await hass.config_entries.async_setup(config_entry.entry_id) await hass.async_block_till_done() return config_entry async def test_gateway_setup( hass: HomeAssistant, aioclient_mock: AiohttpClientMocker ) -> None: """Successful setup.""" with patch( "homeassistant.config_entries.ConfigEntries.async_forward_entry_setup", return_value=True, ) as forward_entry_setup: config_entry = await setup_deconz_integration(hass, aioclient_mock) gateway = get_gateway_from_config_entry(hass, config_entry) assert gateway.bridgeid == BRIDGEID assert gateway.master is True assert gateway.option_allow_clip_sensor is False assert gateway.option_allow_deconz_groups is True assert gateway.option_allow_new_devices is True assert len(gateway.deconz_ids) == 0 assert len(hass.states.async_all()) == 0 assert forward_entry_setup.mock_calls[0][1] == ( config_entry, ALARM_CONTROL_PANEL_DOMAIN, ) assert forward_entry_setup.mock_calls[1][1] == ( config_entry, BINARY_SENSOR_DOMAIN, ) assert forward_entry_setup.mock_calls[2][1] == (config_entry, BUTTON_DOMAIN) assert forward_entry_setup.mock_calls[3][1] == (config_entry, CLIMATE_DOMAIN) assert forward_entry_setup.mock_calls[4][1] == (config_entry, COVER_DOMAIN) assert forward_entry_setup.mock_calls[5][1] == (config_entry, FAN_DOMAIN) assert forward_entry_setup.mock_calls[6][1] == (config_entry, LIGHT_DOMAIN) assert forward_entry_setup.mock_calls[7][1] == (config_entry, LOCK_DOMAIN) assert forward_entry_setup.mock_calls[8][1] == (config_entry, NUMBER_DOMAIN) assert forward_entry_setup.mock_calls[9][1] == (config_entry, SCENE_DOMAIN) assert forward_entry_setup.mock_calls[10][1] == (config_entry, SELECT_DOMAIN) assert forward_entry_setup.mock_calls[11][1] == (config_entry, SENSOR_DOMAIN) assert forward_entry_setup.mock_calls[12][1] == (config_entry, SIREN_DOMAIN) assert forward_entry_setup.mock_calls[13][1] == (config_entry, SWITCH_DOMAIN) device_registry = dr.async_get(hass) gateway_entry = device_registry.async_get_device( identifiers={(DECONZ_DOMAIN, gateway.bridgeid)} ) assert gateway_entry.configuration_url == f"http://{HOST}:{PORT}" assert gateway_entry.entry_type is dr.DeviceEntryType.SERVICE async def test_gateway_device_configuration_url_when_addon( hass: HomeAssistant, aioclient_mock: AiohttpClientMocker ) -> None: """Successful setup.""" with patch( "homeassistant.config_entries.ConfigEntries.async_forward_entry_setup", return_value=True, ): config_entry = await setup_deconz_integration( hass, aioclient_mock, source=SOURCE_HASSIO ) gateway = get_gateway_from_config_entry(hass, config_entry) device_registry = dr.async_get(hass) gateway_entry = device_registry.async_get_device( identifiers={(DECONZ_DOMAIN, gateway.bridgeid)} ) assert ( gateway_entry.configuration_url == "homeassistant://hassio/ingress/core_deconz" ) async def test_connection_status_signalling( hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, mock_deconz_websocket ) -> None: """Make sure that connection status triggers a dispatcher send.""" data = { "sensors": { "1": { "name": "presence", "type": "ZHAPresence", "state": {"presence": False}, "config": {"on": True, "reachable": True}, "uniqueid": "00:00:00:00:00:00:00:00-00", } } } with patch.dict(DECONZ_WEB_REQUEST, data): await setup_deconz_integration(hass, aioclient_mock) assert hass.states.get("binary_sensor.presence").state == STATE_OFF await mock_deconz_websocket(state=State.RETRYING) await hass.async_block_till_done() assert hass.states.get("binary_sensor.presence").state == STATE_UNAVAILABLE await mock_deconz_websocket(state=State.RUNNING) await hass.async_block_till_done() assert hass.states.get("binary_sensor.presence").state == STATE_OFF async def test_update_address( hass: HomeAssistant, aioclient_mock: AiohttpClientMocker ) -> None: """Make sure that connection status triggers a dispatcher send.""" config_entry = await setup_deconz_integration(hass, aioclient_mock) gateway = get_gateway_from_config_entry(hass, config_entry) assert gateway.api.host == "1.2.3.4" with patch( "homeassistant.components.deconz.async_setup_entry", return_value=True, ) as mock_setup_entry: await hass.config_entries.flow.async_init( DECONZ_DOMAIN, data=ssdp.SsdpServiceInfo( ssdp_st="mock_st", ssdp_usn="mock_usn", ssdp_location="http://2.3.4.5:80/", upnp={ ATTR_UPNP_MANUFACTURER_URL: DECONZ_MANUFACTURERURL, ATTR_UPNP_SERIAL: BRIDGEID, ATTR_UPNP_UDN: "uuid:456DEF", }, ), context={"source": SOURCE_SSDP}, ) await hass.async_block_till_done() assert gateway.api.host == "2.3.4.5" assert len(mock_setup_entry.mock_calls) == 1 async def test_reset_after_successful_setup( hass: HomeAssistant, aioclient_mock: AiohttpClientMocker ) -> None: """Make sure that connection status triggers a dispatcher send.""" config_entry = await setup_deconz_integration(hass, aioclient_mock) gateway = get_gateway_from_config_entry(hass, config_entry) result = await gateway.async_reset() await hass.async_block_till_done() assert result is True async def test_get_deconz_session(hass: HomeAssistant) -> None: """Successful call.""" with patch("pydeconz.DeconzSession.refresh_state", return_value=True): assert await get_deconz_session(hass, ENTRY_CONFIG) @pytest.mark.parametrize( ("side_effect", "raised_exception"), [ (asyncio.TimeoutError, CannotConnect), (pydeconz.RequestError, CannotConnect), (pydeconz.ResponseError, CannotConnect), (pydeconz.Unauthorized, AuthenticationRequired), ], ) async def test_get_deconz_session_fails( hass: HomeAssistant, side_effect, raised_exception ) -> None: """Failed call.""" with patch( "pydeconz.DeconzSession.refresh_state", side_effect=side_effect, ), pytest.raises(raised_exception): assert await get_deconz_session(hass, ENTRY_CONFIG)