Follow Axis library changes and improve tests (#44126)

This commit is contained in:
Robert Svensson 2021-01-13 14:03:54 +01:00 committed by GitHub
parent ff3a1f2050
commit 6325bc8bfe
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
10 changed files with 149 additions and 78 deletions

View file

@ -1,26 +1,12 @@
"""Test Axis device."""
from copy import deepcopy
from unittest import mock
from unittest.mock import AsyncMock, Mock, patch
from unittest.mock import Mock, patch
import axis as axislib
from axis.api_discovery import URL as API_DISCOVERY_URL
from axis.applications import URL_LIST as APPLICATIONS_URL
from axis.applications.vmd4 import URL as VMD4_URL
from axis.basic_device_info import URL as BASIC_DEVICE_INFO_URL
from axis.event_stream import OPERATION_INITIALIZED
from axis.light_control import URL as LIGHT_CONTROL_URL
from axis.mqtt import URL_CLIENT as MQTT_CLIENT_URL
from axis.param_cgi import (
BRAND as BRAND_URL,
INPUT as INPUT_URL,
IOPORT as IOPORT_URL,
OUTPUT as OUTPUT_URL,
PROPERTIES as PROPERTIES_URL,
STREAM_PROFILES as STREAM_PROFILES_URL,
)
from axis.port_management import URL as PORT_MANAGEMENT_URL
import pytest
import respx
from homeassistant import config_entries
from homeassistant.components import axis
@ -47,10 +33,12 @@ MAC = "00408C12345"
MODEL = "model"
NAME = "name"
DEFAULT_HOST = "1.2.3.4"
ENTRY_OPTIONS = {CONF_EVENTS: True}
ENTRY_CONFIG = {
CONF_HOST: "1.2.3.4",
CONF_HOST: DEFAULT_HOST,
CONF_USERNAME: "root",
CONF_PASSWORD: "pass",
CONF_PORT: 80,
@ -166,6 +154,14 @@ root.Brand.ProdVariant=
root.Brand.WebURL=http://www.axis.com
"""
IMAGE_RESPONSE = """root.Image.I0.Enabled=yes
root.Image.I0.Name=View Area 1
root.Image.I0.Source=0
root.Image.I1.Enabled=no
root.Image.I1.Name=View Area 2
root.Image.I1.Source=0
"""
PORTS_RESPONSE = """root.Input.NbrOfInputs=1
root.IOPort.I0.Configurable=no
root.IOPort.I0.Direction=input
@ -188,6 +184,9 @@ root.Properties.Image.Rotation=0,180
root.Properties.System.SerialNumber=00408C12345
"""
PTZ_RESPONSE = ""
STREAM_PROFILES_RESPONSE = """root.StreamProfile.MaxGroups=26
root.StreamProfile.S0.Description=profile_1_description
root.StreamProfile.S0.Name=profile_1
@ -197,31 +196,85 @@ root.StreamProfile.S1.Name=profile_2
root.StreamProfile.S1.Parameters=videocodec=h265
"""
VIEW_AREAS_RESPONSE = {"apiVersion": "1.0", "method": "list", "data": {"viewAreas": []}}
async def vapix_request(self, session, url, **kwargs):
"""Return data based on url."""
if API_DISCOVERY_URL in url:
return API_DISCOVERY_RESPONSE
if APPLICATIONS_URL in url:
return APPLICATIONS_LIST_RESPONSE
if BASIC_DEVICE_INFO_URL in url:
return BASIC_DEVICE_INFO_RESPONSE
if LIGHT_CONTROL_URL in url:
return LIGHT_CONTROL_RESPONSE
if MQTT_CLIENT_URL in url:
return MQTT_CLIENT_RESPONSE
if PORT_MANAGEMENT_URL in url:
return PORT_MANAGEMENT_RESPONSE
if VMD4_URL in url:
return VMD4_RESPONSE
if BRAND_URL in url:
return BRAND_RESPONSE
if IOPORT_URL in url or INPUT_URL in url or OUTPUT_URL in url:
return PORTS_RESPONSE
if PROPERTIES_URL in url:
return PROPERTIES_RESPONSE
if STREAM_PROFILES_URL in url:
return STREAM_PROFILES_RESPONSE
def mock_default_vapix_requests(respx: respx, host: str = DEFAULT_HOST) -> None:
"""Mock default Vapix requests responses."""
respx.post(f"http://{host}:80/axis-cgi/apidiscovery.cgi").respond(
json=API_DISCOVERY_RESPONSE,
)
respx.post(f"http://{host}:80/axis-cgi/basicdeviceinfo.cgi").respond(
json=BASIC_DEVICE_INFO_RESPONSE,
)
respx.post(f"http://{host}:80/axis-cgi/io/portmanagement.cgi").respond(
json=PORT_MANAGEMENT_RESPONSE,
)
respx.post(f"http://{host}:80/axis-cgi/lightcontrol.cgi").respond(
json=LIGHT_CONTROL_RESPONSE,
)
respx.post(f"http://{host}:80/axis-cgi/mqtt/client.cgi").respond(
json=MQTT_CLIENT_RESPONSE,
)
respx.post(f"http://{host}:80/axis-cgi/streamprofile.cgi").respond(
json=STREAM_PROFILES_RESPONSE,
)
respx.post(f"http://{host}:80/axis-cgi/viewarea/info.cgi").respond(
json=VIEW_AREAS_RESPONSE
)
respx.get(
f"http://{host}:80/axis-cgi/param.cgi?action=list&group=root.Brand"
).respond(
text=BRAND_RESPONSE,
headers={"Content-Type": "text/plain"},
)
respx.get(
f"http://{host}:80/axis-cgi/param.cgi?action=list&group=root.Image"
).respond(
text=IMAGE_RESPONSE,
headers={"Content-Type": "text/plain"},
)
respx.get(
f"http://{host}:80/axis-cgi/param.cgi?action=list&group=root.Input"
).respond(
text=PORTS_RESPONSE,
headers={"Content-Type": "text/plain"},
)
respx.get(
f"http://{host}:80/axis-cgi/param.cgi?action=list&group=root.IOPort"
).respond(
text=PORTS_RESPONSE,
headers={"Content-Type": "text/plain"},
)
respx.get(
f"http://{host}:80/axis-cgi/param.cgi?action=list&group=root.Output"
).respond(
text=PORTS_RESPONSE,
headers={"Content-Type": "text/plain"},
)
respx.get(
f"http://{host}:80/axis-cgi/param.cgi?action=list&group=root.Properties"
).respond(
text=PROPERTIES_RESPONSE,
headers={"Content-Type": "text/plain"},
)
respx.get(
f"http://{host}:80/axis-cgi/param.cgi?action=list&group=root.PTZ"
).respond(
text=PTZ_RESPONSE,
headers={"Content-Type": "text/plain"},
)
respx.get(
f"http://{host}:80/axis-cgi/param.cgi?action=list&group=root.StreamProfile"
).respond(
text=STREAM_PROFILES_RESPONSE,
headers={"Content-Type": "text/plain"},
)
respx.post(f"http://{host}:80/axis-cgi/applications/list.cgi").respond(
text=APPLICATIONS_LIST_RESPONSE,
headers={"Content-Type": "text/xml"},
)
respx.post(f"http://{host}:80/local/vmd/control.cgi").respond(json=VMD4_RESPONSE)
async def setup_axis_integration(hass, config=ENTRY_CONFIG, options=ENTRY_OPTIONS):
@ -235,10 +288,8 @@ async def setup_axis_integration(hass, config=ENTRY_CONFIG, options=ENTRY_OPTION
)
config_entry.add_to_hass(hass)
with patch("axis.vapix.Vapix.request", new=vapix_request), patch(
"axis.rtsp.RTSPClient.start",
return_value=True,
):
with patch("axis.rtsp.RTSPClient.start", return_value=True), respx.mock:
mock_default_vapix_requests(respx)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
@ -317,10 +368,11 @@ async def test_update_address(hass):
device = hass.data[AXIS_DOMAIN][config_entry.unique_id]
assert device.api.config.host == "1.2.3.4"
with patch("axis.vapix.Vapix.request", new=vapix_request), patch(
with patch(
"homeassistant.components.axis.async_setup_entry",
return_value=True,
) as mock_setup_entry:
) as mock_setup_entry, respx.mock:
mock_default_vapix_requests(respx, "2.3.4.5")
await hass.config_entries.flow.async_init(
AXIS_DOMAIN,
data={
@ -390,12 +442,10 @@ async def test_shutdown():
axis_device = axis.device.AxisNetworkDevice(hass, entry)
axis_device.api = Mock()
axis_device.api.vapix.close = AsyncMock()
await axis_device.shutdown(None)
assert len(axis_device.api.stream.stop.mock_calls) == 1
assert len(axis_device.api.vapix.close.mock_calls) == 1
async def test_get_device_fails(hass):