evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add fixtures for Axis rtsp client and adapt tests to use them (#47901)

Browse source 
* Add a fixture for rtsp client and adapt tests to use it

* Better fixtures for RTSP events and signals
This commit is contained in:
Robert Svensson 2021-04-09 10:56:53 +02:00 • committed by GitHub
parent e7e53b879e
commit 31ae121645
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
7 changed files with 230 additions and 90 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
homeassistant/components/axis/__init__.py
View file

@ -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

4
homeassistant/components/axis/device.py
View file

@ -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):

112
tests/components/axis/conftest.py
View file

@ -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

51
tests/components/axis/test_binary_sensor.py
View file

@ -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

40
tests/components/axis/test_device.py
View file

@ -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):

54
tests/components/axis/test_light.py
View file

@ -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)

55
tests/components/axis/test_switch.py
View file

@ -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