Add fixtures for Axis rtsp client and adapt tests to use them

homeassistant/components/axis/__init__.py

@@ -31,7 +31,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
 

homeassistant/components/axis/device.py

@@ -263,9 +263,7 @@ async def start_platforms():
     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):

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'<tt:SimpleItem Name="{source_name}" Value="{source_idx}"/>'
+
+        event = f"""<?xml version="1.0" encoding="UTF-8"?>
+<tt:MetadataStream xmlns:tt="http://www.onvif.org/ver10/schema">
+    <tt:Event>
+        <wsnt:NotificationMessage xmlns:tns1="http://www.onvif.org/ver10/topics"
+                                  xmlns:tnsaxis="http://www.axis.com/2009/event/topics"
+                                  xmlns:wsnt="http://docs.oasis-open.org/wsn/b-2"
+                                  xmlns:wsa5="http://www.w3.org/2005/08/addressing">
+            <wsnt:Topic Dialect="http://docs.oasis-open.org/wsn/t-1/TopicExpression/Simple">
+                {topic}
+            </wsnt:Topic>
+            <wsnt:ProducerReference>
+                <wsa5:Address>
+                    uri://bf32a3b9-e5e7-4d57-a48d-1b5be9ae7b16/ProducerReference
+                </wsa5:Address>
+            </wsnt:ProducerReference>
+            <wsnt:Message>
+                <tt:Message UtcTime="2020-11-03T20:21:48.346022Z"
+                            PropertyOperation="{operation}">
+                    <tt:Source>{source}</tt:Source>
+                    <tt:Key></tt:Key>
+                    <tt:Data>
+                        <tt:SimpleItem Name="{data_type}" Value="{data_value}"/>
+                    </tt:Data>
+                </tt:Message>
+            </wsnt:Message>
+        </wsnt:NotificationMessage>
+    </tt:Event>
+</tt:MetadataStream>
+"""
+
+        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

tests/components/axis/test_binary_sensor.py

@@ -10,31 +10,6 @@
 
 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

tests/components/axis/test_device.py

@@ -23,7 +23,9 @@
     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):

tests/components/axis/test_light.py

@@ -27,24 +27,6 @@
     "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]
-
-    device.api.event.update([EVENT_ON])
+        await setup_axis_integration(hass)
+
+    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)