diff --git a/homeassistant/components/axis/__init__.py b/homeassistant/components/axis/__init__.py index 378d02bcccd..acbdc2ca782 100644 --- a/homeassistant/components/axis/__init__.py +++ b/homeassistant/components/axis/__init__.py @@ -26,7 +26,9 @@ async def async_setup_entry(hass, config_entry): await device.async_update_device_registry() - hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, device.shutdown) + device.listeners.append( + hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, device.shutdown) + ) return True diff --git a/homeassistant/components/axis/device.py b/homeassistant/components/axis/device.py index f732ad2fb5d..93b63b64122 100644 --- a/homeassistant/components/axis/device.py +++ b/homeassistant/components/axis/device.py @@ -263,9 +263,7 @@ class AxisNetworkDevice: def disconnect_from_stream(self): """Stop stream.""" if self.api.stream.state != STATE_STOPPED: - self.api.stream.connection_status_callback.remove( - self.async_connection_status_callback - ) + self.api.stream.connection_status_callback.clear() self.api.stream.stop() async def shutdown(self, event): diff --git a/tests/components/axis/conftest.py b/tests/components/axis/conftest.py index b3964663767..be448359366 100644 --- a/tests/components/axis/conftest.py +++ b/tests/components/axis/conftest.py @@ -1,2 +1,112 @@ -"""axis conftest.""" +"""Axis conftest.""" + +from typing import Optional +from unittest.mock import patch + +from axis.rtsp import ( + SIGNAL_DATA, + SIGNAL_FAILED, + SIGNAL_PLAYING, + STATE_PLAYING, + STATE_STOPPED, +) +import pytest + from tests.components.light.conftest import mock_light_profiles # noqa: F401 + + +@pytest.fixture(autouse=True) +def mock_axis_rtspclient(): + """No real RTSP communication allowed.""" + with patch("axis.streammanager.RTSPClient") as rtsp_client_mock: + + rtsp_client_mock.return_value.session.state = STATE_STOPPED + + async def start_stream(): + """Set state to playing when calling RTSPClient.start.""" + rtsp_client_mock.return_value.session.state = STATE_PLAYING + + rtsp_client_mock.return_value.start = start_stream + + def stop_stream(): + """Set state to stopped when calling RTSPClient.stop.""" + rtsp_client_mock.return_value.session.state = STATE_STOPPED + + rtsp_client_mock.return_value.stop = stop_stream + + def make_rtsp_call(data: Optional[dict] = None, state: str = ""): + """Generate a RTSP call.""" + axis_streammanager_session_callback = rtsp_client_mock.call_args[0][4] + + if data: + rtsp_client_mock.return_value.rtp.data = data + axis_streammanager_session_callback(signal=SIGNAL_DATA) + elif state: + axis_streammanager_session_callback(signal=state) + else: + raise NotImplementedError + + yield make_rtsp_call + + +@pytest.fixture(autouse=True) +def mock_rtsp_event(mock_axis_rtspclient): + """Fixture to allow mocking received RTSP events.""" + + def send_event( + topic: str, + data_type: str, + data_value: str, + operation: str = "Initialized", + source_name: str = "", + source_idx: str = "", + ) -> None: + source = "" + if source_name != "" and source_idx != "": + source = f'' + + event = f""" + + + + + {topic} + + + + uri://bf32a3b9-e5e7-4d57-a48d-1b5be9ae7b16/ProducerReference + + + + + {source} + + + + + + + + + +""" + + mock_axis_rtspclient(data=event.encode("utf-8")) + + yield send_event + + +@pytest.fixture(autouse=True) +def mock_rtsp_signal_state(mock_axis_rtspclient): + """Fixture to allow mocking RTSP state signalling.""" + + def send_signal(connected: bool) -> None: + """Signal state change of RTSP connection.""" + signal = SIGNAL_PLAYING if connected else SIGNAL_FAILED + mock_axis_rtspclient(state=signal) + + yield send_signal diff --git a/tests/components/axis/test_binary_sensor.py b/tests/components/axis/test_binary_sensor.py index 98ef55282c3..2429ec61855 100644 --- a/tests/components/axis/test_binary_sensor.py +++ b/tests/components/axis/test_binary_sensor.py @@ -10,31 +10,6 @@ from homeassistant.setup import async_setup_component from .test_device import NAME, setup_axis_integration -EVENTS = [ - { - "operation": "Initialized", - "topic": "tns1:Device/tnsaxis:Sensor/PIR", - "source": "sensor", - "source_idx": "0", - "type": "state", - "value": "0", - }, - { - "operation": "Initialized", - "topic": "tns1:PTZController/tnsaxis:PTZPresets/Channel_1", - "source": "PresetToken", - "source_idx": "0", - "type": "on_preset", - "value": "1", - }, - { - "operation": "Initialized", - "topic": "tnsaxis:CameraApplicationPlatform/VMD/Camera1Profile1", - "type": "active", - "value": "1", - }, -] - async def test_platform_manually_configured(hass): """Test that nothing happens when platform is manually configured.""" @@ -57,12 +32,30 @@ async def test_no_binary_sensors(hass): assert not hass.states.async_entity_ids(BINARY_SENSOR_DOMAIN) -async def test_binary_sensors(hass): +async def test_binary_sensors(hass, mock_rtsp_event): """Test that sensors are loaded properly.""" - config_entry = await setup_axis_integration(hass) - device = hass.data[AXIS_DOMAIN][config_entry.unique_id] + await setup_axis_integration(hass) - device.api.event.update(EVENTS) + mock_rtsp_event( + topic="tns1:Device/tnsaxis:Sensor/PIR", + data_type="state", + data_value="0", + source_name="sensor", + source_idx="0", + ) + mock_rtsp_event( + topic="tnsaxis:CameraApplicationPlatform/VMD/Camera1Profile1", + data_type="active", + data_value="1", + ) + # Unsupported event + mock_rtsp_event( + topic="tns1:PTZController/tnsaxis:PTZPresets/Channel_1", + data_type="on_preset", + data_value="1", + source_name="PresetToken", + source_idx="0", + ) await hass.async_block_till_done() assert len(hass.states.async_entity_ids(BINARY_SENSOR_DOMAIN)) == 2 diff --git a/tests/components/axis/test_device.py b/tests/components/axis/test_device.py index a5371395638..cb6e5b1a12b 100644 --- a/tests/components/axis/test_device.py +++ b/tests/components/axis/test_device.py @@ -23,7 +23,9 @@ from homeassistant.const import ( CONF_PASSWORD, CONF_PORT, CONF_USERNAME, + STATE_OFF, STATE_ON, + STATE_UNAVAILABLE, ) from tests.common import MockConfigEntry, async_fire_mqtt_message @@ -288,7 +290,7 @@ async def setup_axis_integration(hass, config=ENTRY_CONFIG, options=ENTRY_OPTION ) config_entry.add_to_hass(hass) - with patch("axis.rtsp.RTSPClient.start", return_value=True), respx.mock: + with respx.mock: mock_default_vapix_requests(respx) await hass.config_entries.async_setup(config_entry.entry_id) await hass.async_block_till_done() @@ -389,12 +391,38 @@ async def test_update_address(hass): assert len(mock_setup_entry.mock_calls) == 1 -async def test_device_unavailable(hass): +async def test_device_unavailable(hass, mock_rtsp_event, mock_rtsp_signal_state): """Successful setup.""" - config_entry = await setup_axis_integration(hass) - device = hass.data[AXIS_DOMAIN][config_entry.unique_id] - device.async_connection_status_callback(status=False) - assert not device.available + await setup_axis_integration(hass) + + # Provide an entity that can be used to verify connection state on + mock_rtsp_event( + topic="tns1:AudioSource/tnsaxis:TriggerLevel", + data_type="triggered", + data_value="10", + source_name="channel", + source_idx="1", + ) + await hass.async_block_till_done() + + assert hass.states.get(f"{BINARY_SENSOR_DOMAIN}.{NAME}_sound_1").state == STATE_OFF + + # Connection to device has failed + + mock_rtsp_signal_state(connected=False) + await hass.async_block_till_done() + + assert ( + hass.states.get(f"{BINARY_SENSOR_DOMAIN}.{NAME}_sound_1").state + == STATE_UNAVAILABLE + ) + + # Connection to device has been restored + + mock_rtsp_signal_state(connected=True) + await hass.async_block_till_done() + + assert hass.states.get(f"{BINARY_SENSOR_DOMAIN}.{NAME}_sound_1").state == STATE_OFF async def test_device_reset(hass): diff --git a/tests/components/axis/test_light.py b/tests/components/axis/test_light.py index db4ba86ceae..db7ca6921fb 100644 --- a/tests/components/axis/test_light.py +++ b/tests/components/axis/test_light.py @@ -27,24 +27,6 @@ API_DISCOVERY_LIGHT_CONTROL = { "name": "Light Control", } -EVENT_ON = { - "operation": "Initialized", - "topic": "tns1:Device/tnsaxis:Light/Status", - "source": "id", - "source_idx": "0", - "type": "state", - "value": "ON", -} - -EVENT_OFF = { - "operation": "Initialized", - "topic": "tns1:Device/tnsaxis:Light/Status", - "source": "id", - "source_idx": "0", - "type": "state", - "value": "OFF", -} - async def test_platform_manually_configured(hass): """Test that nothing happens when platform is manually configured.""" @@ -62,7 +44,9 @@ async def test_no_lights(hass): assert not hass.states.async_entity_ids(LIGHT_DOMAIN) -async def test_no_light_entity_without_light_control_representation(hass): +async def test_no_light_entity_without_light_control_representation( + hass, mock_rtsp_event +): """Verify no lights entities get created without light control representation.""" api_discovery = deepcopy(API_DISCOVERY_RESPONSE) api_discovery["data"]["apiList"].append(API_DISCOVERY_LIGHT_CONTROL) @@ -73,23 +57,27 @@ async def test_no_light_entity_without_light_control_representation(hass): with patch.dict(API_DISCOVERY_RESPONSE, api_discovery), patch.dict( LIGHT_CONTROL_RESPONSE, light_control ): - config_entry = await setup_axis_integration(hass) - device = hass.data[AXIS_DOMAIN][config_entry.unique_id] + await setup_axis_integration(hass) - device.api.event.update([EVENT_ON]) + mock_rtsp_event( + topic="tns1:Device/tnsaxis:Light/Status", + data_type="state", + data_value="ON", + source_name="id", + source_idx="0", + ) await hass.async_block_till_done() assert not hass.states.async_entity_ids(LIGHT_DOMAIN) -async def test_lights(hass): +async def test_lights(hass, mock_rtsp_event): """Test that lights are loaded properly.""" api_discovery = deepcopy(API_DISCOVERY_RESPONSE) api_discovery["data"]["apiList"].append(API_DISCOVERY_LIGHT_CONTROL) with patch.dict(API_DISCOVERY_RESPONSE, api_discovery): - config_entry = await setup_axis_integration(hass) - device = hass.data[AXIS_DOMAIN][config_entry.unique_id] + await setup_axis_integration(hass) # Add light with patch( @@ -99,7 +87,13 @@ async def test_lights(hass): "axis.light_control.LightControl.get_valid_intensity", return_value={"data": {"ranges": [{"high": 150}]}}, ): - device.api.event.update([EVENT_ON]) + mock_rtsp_event( + topic="tns1:Device/tnsaxis:Light/Status", + data_type="state", + data_value="ON", + source_name="id", + source_idx="0", + ) await hass.async_block_till_done() assert len(hass.states.async_entity_ids(LIGHT_DOMAIN)) == 1 @@ -144,7 +138,13 @@ async def test_lights(hass): mock_deactivate.assert_called_once() # Event turn off light - device.api.event.update([EVENT_OFF]) + mock_rtsp_event( + topic="tns1:Device/tnsaxis:Light/Status", + data_type="state", + data_value="OFF", + source_name="id", + source_idx="0", + ) await hass.async_block_till_done() light_0 = hass.states.get(entity_id) diff --git a/tests/components/axis/test_switch.py b/tests/components/axis/test_switch.py index dcbe285cb54..541c377d3ff 100644 --- a/tests/components/axis/test_switch.py +++ b/tests/components/axis/test_switch.py @@ -21,25 +21,6 @@ from .test_device import ( setup_axis_integration, ) -EVENTS = [ - { - "operation": "Initialized", - "topic": "tns1:Device/Trigger/Relay", - "source": "RelayToken", - "source_idx": "0", - "type": "LogicalState", - "value": "inactive", - }, - { - "operation": "Initialized", - "topic": "tns1:Device/Trigger/Relay", - "source": "RelayToken", - "source_idx": "1", - "type": "LogicalState", - "value": "active", - }, -] - async def test_platform_manually_configured(hass): """Test that nothing happens when platform is manually configured.""" @@ -57,7 +38,7 @@ async def test_no_switches(hass): assert not hass.states.async_entity_ids(SWITCH_DOMAIN) -async def test_switches_with_port_cgi(hass): +async def test_switches_with_port_cgi(hass, mock_rtsp_event): """Test that switches are loaded properly using port.cgi.""" config_entry = await setup_axis_integration(hass) device = hass.data[AXIS_DOMAIN][config_entry.unique_id] @@ -68,7 +49,20 @@ async def test_switches_with_port_cgi(hass): device.api.vapix.ports["0"].close = AsyncMock() device.api.vapix.ports["1"].name = "" - device.api.event.update(EVENTS) + mock_rtsp_event( + topic="tns1:Device/Trigger/Relay", + data_type="LogicalState", + data_value="inactive", + source_name="RelayToken", + source_idx="0", + ) + mock_rtsp_event( + topic="tns1:Device/Trigger/Relay", + data_type="LogicalState", + data_value="active", + source_name="RelayToken", + source_idx="1", + ) await hass.async_block_till_done() assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 2 @@ -100,7 +94,9 @@ async def test_switches_with_port_cgi(hass): device.api.vapix.ports["0"].open.assert_called_once() -async def test_switches_with_port_management(hass): +async def test_switches_with_port_management( + hass, mock_axis_rtspclient, mock_rtsp_event +): """Test that switches are loaded properly using port management.""" api_discovery = deepcopy(API_DISCOVERY_RESPONSE) api_discovery["data"]["apiList"].append(API_DISCOVERY_PORT_MANAGEMENT) @@ -115,7 +111,20 @@ async def test_switches_with_port_management(hass): device.api.vapix.ports["0"].close = AsyncMock() device.api.vapix.ports["1"].name = "" - device.api.event.update(EVENTS) + mock_rtsp_event( + topic="tns1:Device/Trigger/Relay", + data_type="LogicalState", + data_value="inactive", + source_name="RelayToken", + source_idx="0", + ) + mock_rtsp_event( + topic="tns1:Device/Trigger/Relay", + data_type="LogicalState", + data_value="active", + source_name="RelayToken", + source_idx="1", + ) await hass.async_block_till_done() assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 2