Adapt Axis integration to library refactoring (#110898)

* Adapt Axis integration to library refactoring

* Bump axis to v49
pull/111892/head
Robert Svensson 2024-02-28 14:36:32 +01:00 committed by GitHub
parent 2b3630b054
commit c478b1416c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
17 changed files with 403 additions and 199 deletions

View File

@ -5,6 +5,10 @@ from collections.abc import Callable
from datetime import datetime, timedelta
from axis.models.event import Event, EventGroup, EventOperation, EventTopic
from axis.vapix.interfaces.applications.fence_guard import FenceGuardHandler
from axis.vapix.interfaces.applications.loitering_guard import LoiteringGuardHandler
from axis.vapix.interfaces.applications.motion_guard import MotionGuardHandler
from axis.vapix.interfaces.applications.vmd4 import Vmd4Handler
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
@ -111,17 +115,33 @@ class AxisBinarySensor(AxisEventEntity, BinarySensorEntity):
self._attr_name = self.device.api.vapix.ports[event.id].name
elif event.group == EventGroup.MOTION:
for event_topic, event_data in (
(EventTopic.FENCE_GUARD, self.device.api.vapix.fence_guard),
(EventTopic.LOITERING_GUARD, self.device.api.vapix.loitering_guard),
(EventTopic.MOTION_GUARD, self.device.api.vapix.motion_guard),
(EventTopic.OBJECT_ANALYTICS, self.device.api.vapix.object_analytics),
(EventTopic.MOTION_DETECTION_4, self.device.api.vapix.vmd4),
event_data: FenceGuardHandler | LoiteringGuardHandler | MotionGuardHandler | Vmd4Handler | None = None
if event.topic_base == EventTopic.FENCE_GUARD:
event_data = self.device.api.vapix.fence_guard
elif event.topic_base == EventTopic.LOITERING_GUARD:
event_data = self.device.api.vapix.loitering_guard
elif event.topic_base == EventTopic.MOTION_GUARD:
event_data = self.device.api.vapix.motion_guard
elif event.topic_base == EventTopic.MOTION_DETECTION_4:
event_data = self.device.api.vapix.vmd4
if (
event_data
and event_data.initialized
and (profiles := event_data["0"].profiles)
):
if (
event.topic_base == event_topic
and event_data
and event.id in event_data
):
self._attr_name = f"{self._event_type} {event_data[event.id].name}"
break
for profile_id, profile in profiles.items():
camera_id = profile.camera
if event.id == f"Camera{camera_id}Profile{profile_id}":
self._attr_name = f"{self._event_type} {profile.name}"
return
if (
event.topic_base == EventTopic.OBJECT_ANALYTICS
and self.device.api.vapix.object_analytics.initialized
and (scenarios := self.device.api.vapix.object_analytics["0"].scenarios)
):
for scenario_id, scenario in scenarios.items():
device_id = scenario.devices[0]["id"]
if event.id == f"Device{device_id}Scenario{scenario_id}":
self._attr_name = f"{self._event_type} {scenario.name}"
break

View File

@ -24,7 +24,10 @@ async def async_setup_entry(
device: AxisNetworkDevice = hass.data[AXIS_DOMAIN][config_entry.entry_id]
if not device.api.vapix.params.image_format:
if (
not (prop := device.api.vapix.params.property_handler.get("0"))
or not prop.image_format
):
return
async_add_entities([AxisCamera(device)])

View File

@ -249,7 +249,10 @@ class AxisOptionsFlowHandler(config_entries.OptionsFlowWithConfigEntry):
# Stream profiles
if vapix.stream_profiles or vapix.params.stream_profiles_max_groups > 0:
if vapix.stream_profiles or (
(profiles := vapix.params.stream_profile_handler.get("0"))
and profiles.max_groups > 0
):
stream_profiles = [DEFAULT_STREAM_PROFILE]
for profile in vapix.streaming_profiles:
stream_profiles.append(profile.name)
@ -262,14 +265,17 @@ class AxisOptionsFlowHandler(config_entries.OptionsFlowWithConfigEntry):
# Video sources
if vapix.params.image_nbrofviews > 0:
await vapix.params.update_image()
video_sources = {DEFAULT_VIDEO_SOURCE: DEFAULT_VIDEO_SOURCE}
for idx, video_source in vapix.params.image_sources.items():
if not video_source["Enabled"]:
if (
properties := vapix.params.property_handler.get("0")
) and properties.image_number_of_views > 0:
await vapix.params.image_handler.update()
video_sources: dict[int | str, str] = {
DEFAULT_VIDEO_SOURCE: DEFAULT_VIDEO_SOURCE
}
for idx, video_source in vapix.params.image_handler.items():
if not video_source.enabled:
continue
video_sources[idx + 1] = video_source["Name"]
video_sources[int(idx) + 1] = video_source.name
schema[
vol.Optional(CONF_VIDEO_SOURCE, default=self.device.option_video_source)

View File

@ -9,6 +9,7 @@ from axis.configuration import Configuration
from axis.errors import Unauthorized
from axis.stream_manager import Signal, State
from axis.vapix.interfaces.mqtt import mqtt_json_to_event
from axis.vapix.models.mqtt import ClientState
from homeassistant.components import mqtt
from homeassistant.components.mqtt import DOMAIN as MQTT_DOMAIN
@ -188,9 +189,8 @@ class AxisNetworkDevice:
status = await self.api.vapix.mqtt.get_client_status()
except Unauthorized:
# This means the user has too low privileges
status = {}
if status.get("data", {}).get("status", {}).get("state") == "active":
return
if status.status.state == ClientState.ACTIVE:
self.config_entry.async_on_unload(
await mqtt.async_subscribe(
hass, f"{self.api.vapix.serial_number}/#", self.mqtt_message
@ -209,7 +209,6 @@ class AxisNetworkDevice:
def async_setup_events(self) -> None:
"""Set up the device events."""
if self.option_events:
self.api.stream.connection_status_callback.append(
self.async_connection_status_callback
@ -217,7 +216,7 @@ class AxisNetworkDevice:
self.api.enable_events()
self.api.stream.start()
if self.api.vapix.mqtt:
if self.api.vapix.mqtt.supported:
async_when_setup(self.hass, MQTT_DOMAIN, self.async_use_mqtt)
@callback

View File

@ -33,13 +33,13 @@ async def async_get_config_entry_diagnostics(
if device.api.vapix.basic_device_info:
diag["basic_device_info"] = async_redact_data(
{attr.id: attr.raw for attr in device.api.vapix.basic_device_info.values()},
device.api.vapix.basic_device_info["0"],
REDACT_BASIC_DEVICE_INFO,
)
if device.api.vapix.params:
diag["params"] = async_redact_data(
{param.id: param.raw for param in device.api.vapix.params.values()},
device.api.vapix.params.items(),
REDACT_VAPIX_PARAMS,
)

View File

@ -69,12 +69,12 @@ class AxisLight(AxisEventEntity, LightEntity):
self._light_id
)
)
self.current_intensity = current_intensity["data"]["intensity"]
self.current_intensity = current_intensity
max_intensity = await self.device.api.vapix.light_control.get_valid_intensity(
self._light_id
)
self.max_intensity = max_intensity["data"]["ranges"][0]["high"]
self.max_intensity = max_intensity.high
@callback
def async_event_callback(self, event: Event) -> None:
@ -110,4 +110,4 @@ class AxisLight(AxisEventEntity, LightEntity):
self._light_id
)
)
self.current_intensity = current_intensity["data"]["intensity"]
self.current_intensity = current_intensity

View File

@ -26,7 +26,7 @@
"iot_class": "local_push",
"loggers": ["axis"],
"quality_scale": "platinum",
"requirements": ["axis==48"],
"requirements": ["axis==49"],
"ssdp": [
{
"manufacturer": "AXIS"

View File

@ -39,7 +39,6 @@ class AxisSwitch(AxisEventEntity, SwitchEntity):
def __init__(self, event: Event, device: AxisNetworkDevice) -> None:
"""Initialize the Axis switch."""
super().__init__(event, device)
if event.id and device.api.vapix.ports[event.id].name:
self._attr_name = device.api.vapix.ports[event.id].name
self._attr_is_on = event.is_tripped
@ -52,8 +51,8 @@ class AxisSwitch(AxisEventEntity, SwitchEntity):
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on switch."""
await self.device.api.vapix.ports[self._event_id].close()
await self.device.api.vapix.ports.close(self._event_id)
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off switch."""
await self.device.api.vapix.ports[self._event_id].open()
await self.device.api.vapix.ports.open(self._event_id)

View File

@ -514,7 +514,7 @@ aurorapy==0.2.7
# avion==0.10
# homeassistant.components.axis
axis==48
axis==49
# homeassistant.components.azure_event_hub
azure-eventhub==5.11.1

View File

@ -454,7 +454,7 @@ auroranoaa==0.0.3
aurorapy==0.2.7
# homeassistant.components.axis
axis==48
axis==49
# homeassistant.components.azure_event_hub
azure-eventhub==5.11.1

View File

@ -99,7 +99,9 @@ def options_fixture(request):
@pytest.fixture(name="mock_vapix_requests")
def default_request_fixture(respx_mock):
def default_request_fixture(
respx_mock, port_management_payload, param_properties_payload, param_ports_payload
):
"""Mock default Vapix requests responses."""
def __mock_default_requests(host):
@ -113,7 +115,7 @@ def default_request_fixture(respx_mock):
json=BASIC_DEVICE_INFO_RESPONSE,
)
respx.post(f"{path}/axis-cgi/io/portmanagement.cgi").respond(
json=PORT_MANAGEMENT_RESPONSE,
json=port_management_payload,
)
respx.post(f"{path}/axis-cgi/mqtt/client.cgi").respond(
json=MQTT_CLIENT_RESPONSE,
@ -124,38 +126,58 @@ def default_request_fixture(respx_mock):
respx.post(f"{path}/axis-cgi/viewarea/info.cgi").respond(
json=VIEW_AREAS_RESPONSE
)
respx.get(f"{path}/axis-cgi/param.cgi?action=list&group=root.Brand").respond(
respx.post(
f"{path}/axis-cgi/param.cgi",
data={"action": "list", "group": "root.Brand"},
).respond(
text=BRAND_RESPONSE,
headers={"Content-Type": "text/plain"},
)
respx.get(f"{path}/axis-cgi/param.cgi?action=list&group=root.Image").respond(
respx.post(
f"{path}/axis-cgi/param.cgi",
data={"action": "list", "group": "root.Image"},
).respond(
text=IMAGE_RESPONSE,
headers={"Content-Type": "text/plain"},
)
respx.get(f"{path}/axis-cgi/param.cgi?action=list&group=root.Input").respond(
text=PORTS_RESPONSE,
headers={"Content-Type": "text/plain"},
)
respx.get(f"{path}/axis-cgi/param.cgi?action=list&group=root.IOPort").respond(
text=PORTS_RESPONSE,
headers={"Content-Type": "text/plain"},
)
respx.get(f"{path}/axis-cgi/param.cgi?action=list&group=root.Output").respond(
text=PORTS_RESPONSE,
headers={"Content-Type": "text/plain"},
)
respx.get(
f"{path}/axis-cgi/param.cgi?action=list&group=root.Properties"
respx.post(
f"{path}/axis-cgi/param.cgi",
data={"action": "list", "group": "root.Input"},
).respond(
text=PROPERTIES_RESPONSE,
text=PORTS_RESPONSE,
headers={"Content-Type": "text/plain"},
)
respx.get(f"{path}/axis-cgi/param.cgi?action=list&group=root.PTZ").respond(
respx.post(
f"{path}/axis-cgi/param.cgi",
data={"action": "list", "group": "root.IOPort"},
).respond(
text=param_ports_payload,
headers={"Content-Type": "text/plain"},
)
respx.post(
f"{path}/axis-cgi/param.cgi",
data={"action": "list", "group": "root.Output"},
).respond(
text=PORTS_RESPONSE,
headers={"Content-Type": "text/plain"},
)
respx.post(
f"{path}/axis-cgi/param.cgi",
data={"action": "list", "group": "root.Properties"},
).respond(
text=param_properties_payload,
headers={"Content-Type": "text/plain"},
)
respx.post(
f"{path}/axis-cgi/param.cgi",
data={"action": "list", "group": "root.PTZ"},
).respond(
text=PTZ_RESPONSE,
headers={"Content-Type": "text/plain"},
)
respx.get(
f"{path}/axis-cgi/param.cgi?action=list&group=root.StreamProfile"
respx.post(
f"{path}/axis-cgi/param.cgi",
data={"action": "list", "group": "root.StreamProfile"},
).respond(
text=STREAM_PROFILES_RESPONSE,
headers={"Content-Type": "text/plain"},
@ -184,6 +206,24 @@ def api_discovery_fixture(api_discovery_items):
respx.post(f"http://{DEFAULT_HOST}:80/axis-cgi/apidiscovery.cgi").respond(json=data)
@pytest.fixture(name="port_management_payload")
def io_port_management_data_fixture():
"""Property parameter data."""
return PORT_MANAGEMENT_RESPONSE
@pytest.fixture(name="param_properties_payload")
def param_properties_data_fixture():
"""Property parameter data."""
return PROPERTIES_RESPONSE
@pytest.fixture(name="param_ports_payload")
def param_ports_data_fixture():
"""Property parameter data."""
return PORTS_RESPONSE
@pytest.fixture(name="setup_default_vapix_requests")
def default_vapix_requests_fixture(mock_vapix_requests):
"""Mock default Vapix requests responses."""

View File

@ -1,5 +1,6 @@
"""Constants for Axis integration tests."""
from axis.vapix.models.api import CONTEXT
MAC = "00408C123456"
FORMATTED_MAC = "00:40:8c:12:34:56"
@ -12,6 +13,7 @@ DEFAULT_HOST = "1.2.3.4"
API_DISCOVERY_RESPONSE = {
"method": "getApiList",
"apiVersion": "1.0",
"context": CONTEXT,
"data": {
"apiList": [
{"id": "api-discovery", "version": "1.0", "name": "API Discovery Service"},
@ -38,27 +40,45 @@ APPLICATIONS_LIST_RESPONSE = """<reply result="ok">
BASIC_DEVICE_INFO_RESPONSE = {
"apiVersion": "1.1",
"context": CONTEXT,
"data": {
"propertyList": {
"ProdNbr": "M1065-LW",
"ProdType": "Network Camera",
"SerialNumber": MAC,
"Version": "9.80.1",
"Architecture": "str",
"Brand": "str",
"BuildDate": "str",
"HardwareID": "str",
"ProdFullName": "str",
"ProdShortName": "str",
"ProdVariant": "str",
"Soc": "str",
"SocSerialNumber": "str",
"WebURL": "str",
}
},
}
MQTT_CLIENT_RESPONSE = {
"apiVersion": "1.0",
"context": "some context",
"method": "getClientStatus",
"data": {"status": {"state": "active", "connectionStatus": "Connected"}},
"apiVersion": "1.0",
"context": CONTEXT,
"data": {
"status": {"state": "active", "connectionStatus": "Connected"},
"config": {
"server": {"protocol": "tcp", "host": "192.168.0.90", "port": 1883},
},
},
}
PORT_MANAGEMENT_RESPONSE = {
"apiVersion": "1.0",
"method": "getPorts",
"context": CONTEXT,
"data": {
"numberOfPorts": 1,
"items": [
@ -78,12 +98,13 @@ PORT_MANAGEMENT_RESPONSE = {
VMD4_RESPONSE = {
"apiVersion": "1.4",
"method": "getConfiguration",
"context": "Axis library",
"context": CONTEXT,
"data": {
"cameras": [{"id": 1, "rotation": 0, "active": True}],
"profiles": [
{"filters": [], "camera": 1, "triggers": [], "name": "Profile 1", "uid": 1}
],
"configurationStatus": 2,
},
}
@ -102,6 +123,95 @@ root.Image.I0.Source=0
root.Image.I1.Enabled=no
root.Image.I1.Name=View Area 2
root.Image.I1.Source=0
root.Image.I0.Appearance.ColorEnabled=yes
root.Image.I0.Appearance.Compression=30
root.Image.I0.Appearance.MirrorEnabled=no
root.Image.I0.Appearance.Resolution=1920x1080
root.Image.I0.Appearance.Rotation=0
root.Image.I0.MPEG.Complexity=50
root.Image.I0.MPEG.ConfigHeaderInterval=1
root.Image.I0.MPEG.FrameSkipMode=drop
root.Image.I0.MPEG.ICount=1
root.Image.I0.MPEG.PCount=31
root.Image.I0.MPEG.UserDataEnabled=no
root.Image.I0.MPEG.UserDataInterval=1
root.Image.I0.MPEG.ZChromaQPMode=off
root.Image.I0.MPEG.ZFpsMode=fixed
root.Image.I0.MPEG.ZGopMode=fixed
root.Image.I0.MPEG.ZMaxGopLength=300
root.Image.I0.MPEG.ZMinFps=0
root.Image.I0.MPEG.ZStrength=10
root.Image.I0.MPEG.H264.Profile=high
root.Image.I0.MPEG.H264.PSEnabled=no
root.Image.I0.Overlay.Enabled=no
root.Image.I0.Overlay.XPos=0
root.Image.I0.Overlay.YPos=0
root.Image.I0.Overlay.MaskWindows.Color=black
root.Image.I0.RateControl.MaxBitrate=0
root.Image.I0.RateControl.Mode=vbr
root.Image.I0.RateControl.Priority=framerate
root.Image.I0.RateControl.TargetBitrate=0
root.Image.I0.SizeControl.MaxFrameSize=0
root.Image.I0.Stream.Duration=0
root.Image.I0.Stream.FPS=0
root.Image.I0.Stream.NbrOfFrames=0
root.Image.I0.Text.BGColor=black
root.Image.I0.Text.ClockEnabled=no
root.Image.I0.Text.Color=white
root.Image.I0.Text.DateEnabled=no
root.Image.I0.Text.Position=top
root.Image.I0.Text.String=
root.Image.I0.Text.TextEnabled=no
root.Image.I0.Text.TextSize=medium
root.Image.I0.TriggerData.AudioEnabled=yes
root.Image.I0.TriggerData.MotionDetectionEnabled=yes
root.Image.I0.TriggerData.MotionLevelEnabled=no
root.Image.I0.TriggerData.TamperingEnabled=yes
root.Image.I0.TriggerData.UserTriggers=
root.Image.I1.Appearance.ColorEnabled=yes
root.Image.I1.Appearance.Compression=30
root.Image.I1.Appearance.MirrorEnabled=no
root.Image.I1.Appearance.Resolution=1920x1080
root.Image.I1.Appearance.Rotation=0
root.Image.I1.MPEG.Complexity=50
root.Image.I1.MPEG.ConfigHeaderInterval=1
root.Image.I1.MPEG.FrameSkipMode=drop
root.Image.I1.MPEG.ICount=1
root.Image.I1.MPEG.PCount=31
root.Image.I1.MPEG.UserDataEnabled=no
root.Image.I1.MPEG.UserDataInterval=1
root.Image.I1.MPEG.ZChromaQPMode=off
root.Image.I1.MPEG.ZFpsMode=fixed
root.Image.I1.MPEG.ZGopMode=fixed
root.Image.I1.MPEG.ZMaxGopLength=300
root.Image.I1.MPEG.ZMinFps=0
root.Image.I1.MPEG.ZStrength=10
root.Image.I1.MPEG.H264.Profile=high
root.Image.I1.MPEG.H264.PSEnabled=no
root.Image.I1.Overlay.Enabled=no
root.Image.I1.Overlay.XPos=0
root.Image.I1.Overlay.YPos=0
root.Image.I1.RateControl.MaxBitrate=0
root.Image.I1.RateControl.Mode=vbr
root.Image.I1.RateControl.Priority=framerate
root.Image.I1.RateControl.TargetBitrate=0
root.Image.I1.SizeControl.MaxFrameSize=0
root.Image.I1.Stream.Duration=0
root.Image.I1.Stream.FPS=0
root.Image.I1.Stream.NbrOfFrames=0
root.Image.I1.Text.BGColor=black
root.Image.I1.Text.ClockEnabled=no
root.Image.I1.Text.Color=white
root.Image.I1.Text.DateEnabled=no
root.Image.I1.Text.Position=top
root.Image.I1.Text.String=
root.Image.I1.Text.TextEnabled=no
root.Image.I1.Text.TextSize=medium
root.Image.I1.TriggerData.AudioEnabled=yes
root.Image.I1.TriggerData.MotionDetectionEnabled=yes
root.Image.I1.TriggerData.MotionLevelEnabled=no
root.Image.I1.TriggerData.TamperingEnabled=yes
root.Image.I1.TriggerData.UserTriggers=
"""
PORTS_RESPONSE = """root.Input.NbrOfInputs=1

View File

@ -19,10 +19,8 @@
}),
]),
'basic_device_info': dict({
'ProdNbr': 'M1065-LW',
'ProdType': 'Network Camera',
'SerialNumber': '**REDACTED**',
'Version': '9.80.1',
'__type': "<class 'axis.vapix.models.basic_device_info.DeviceInformation'>",
'repr': "DeviceInformation(id='0', architecture='str', brand='str', build_date='str', firmware_version='9.80.1', hardware_id='str', product_full_name='str', product_number='M1065-LW', product_short_name='str', product_type='Network Camera', product_variant='str', serial_number='00408C123456', soc='str', soc_serial_number='str', web_url='str')",
}),
'camera_sources': dict({
'Image': 'http://1.2.3.4:80/axis-cgi/jpg/image.cgi',
@ -53,41 +51,8 @@
'version': 3,
}),
'params': dict({
'root.IOPort': dict({
'I0.Configurable': 'no',
'I0.Direction': 'input',
'I0.Input.Name': 'PIR sensor',
'I0.Input.Trig': 'closed',
}),
'root.Input': dict({
'NbrOfInputs': '1',
}),
'root.Output': dict({
'NbrOfOutputs': '0',
}),
'root.Properties': dict({
'API.HTTP.Version': '3',
'API.Metadata.Metadata': 'yes',
'API.Metadata.Version': '1.0',
'EmbeddedDevelopment.Version': '2.16',
'Firmware.BuildDate': 'Feb 15 2019 09:42',
'Firmware.BuildNumber': '26',
'Firmware.Version': '9.10.1',
'Image.Format': 'jpeg,mjpeg,h264',
'Image.NbrOfViews': '2',
'Image.Resolution': '1920x1080,1280x960,1280x720,1024x768,1024x576,800x600,640x480,640x360,352x240,320x240',
'Image.Rotation': '0,180',
'System.SerialNumber': '**REDACTED**',
}),
'root.StreamProfile': dict({
'MaxGroups': '26',
'S0.Description': 'profile_1_description',
'S0.Name': 'profile_1',
'S0.Parameters': 'videocodec=h264',
'S1.Description': 'profile_2_description',
'S1.Name': 'profile_2',
'S1.Parameters': 'videocodec=h265',
}),
'__type': "<class 'dict_items'>",
'repr': "dict_items([('Properties', {'API': {'HTTP': {'Version': 3}, 'Metadata': {'Metadata': True, 'Version': '1.0'}}, 'EmbeddedDevelopment': {'Version': '2.16'}, 'Firmware': {'BuildDate': 'Feb 15 2019 09:42', 'BuildNumber': 26, 'Version': '9.10.1'}, 'Image': {'Format': 'jpeg,mjpeg,h264', 'NbrOfViews': 2, 'Resolution': '1920x1080,1280x960,1280x720,1024x768,1024x576,800x600,640x480,640x360,352x240,320x240', 'Rotation': '0,180'}, 'System': {'SerialNumber': '00408C123456'}}), ('Input', {'NbrOfInputs': 1}), ('IOPort', {'I0': {'Configurable': False, 'Direction': 'input', 'Input': {'Name': 'PIR sensor', 'Trig': 'closed'}}}), ('Output', {'NbrOfOutputs': 0}), ('StreamProfile', {'MaxGroups': 26, 'S0': {'Description': 'profile_1_description', 'Name': 'profile_1', 'Parameters': 'videocodec=h264'}, 'S1': {'Description': 'profile_2_description', 'Name': 'profile_2', 'Parameters': 'videocodec=h265'}})])",
}),
})
# ---

View File

@ -1,5 +1,4 @@
"""Axis camera platform tests."""
from unittest.mock import patch
import pytest
@ -13,7 +12,7 @@ from homeassistant.const import STATE_IDLE
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from .const import NAME
from .const import MAC, NAME
async def test_platform_manually_configured(hass: HomeAssistant) -> None:
@ -72,9 +71,19 @@ async def test_camera_with_stream_profile(
)
property_data = f"""root.Properties.API.HTTP.Version=3
root.Properties.API.Metadata.Metadata=yes
root.Properties.API.Metadata.Version=1.0
root.Properties.EmbeddedDevelopment.Version=2.16
root.Properties.Firmware.BuildDate=Feb 15 2019 09:42
root.Properties.Firmware.BuildNumber=26
root.Properties.Firmware.Version=9.10.1
root.Properties.System.SerialNumber={MAC}
"""
@pytest.mark.parametrize("param_properties_payload", [property_data])
async def test_camera_disabled(hass: HomeAssistant, prepare_config_entry) -> None:
"""Test that Axis camera platform is loaded properly but does not create camera entity."""
with patch("axis.vapix.vapix.Params.image_format", new=None):
await prepare_config_entry()
await prepare_config_entry()
assert len(hass.states.async_entity_ids(CAMERA_DOMAIN)) == 0

View File

@ -227,7 +227,7 @@ async def test_shutdown(config) -> None:
async def test_get_device_fails(hass: HomeAssistant, config) -> None:
"""Device unauthorized yields authentication required error."""
with patch(
"axis.vapix.vapix.Vapix.request", side_effect=axislib.Unauthorized
"axis.vapix.vapix.Vapix.initialize", side_effect=axislib.Unauthorized
), pytest.raises(axis.errors.AuthenticationRequired):
await axis.device.get_axis_device(hass, config)

View File

@ -1,6 +1,7 @@
"""Axis light platform tests."""
from unittest.mock import patch
from axis.vapix.models.api import CONTEXT
import pytest
import respx
@ -49,10 +50,18 @@ def light_control_fixture(light_control_items):
"""Light control mock response."""
data = {
"apiVersion": "1.1",
"context": CONTEXT,
"method": "getLightInformation",
"data": {"items": light_control_items},
}
respx.post(f"http://{DEFAULT_HOST}:80/axis-cgi/lightcontrol.cgi").respond(
respx.post(
f"http://{DEFAULT_HOST}:80/axis-cgi/lightcontrol.cgi",
json={
"apiVersion": "1.1",
"context": CONTEXT,
"method": "getLightInformation",
},
).respond(
json=data,
)
@ -90,24 +99,56 @@ async def test_no_light_entity_without_light_control_representation(
@pytest.mark.parametrize("api_discovery_items", [API_DISCOVERY_LIGHT_CONTROL])
async def test_lights(hass: HomeAssistant, setup_config_entry, mock_rtsp_event) -> None:
async def test_lights(
hass: HomeAssistant,
respx_mock,
setup_config_entry,
mock_rtsp_event,
api_discovery_items,
) -> None:
"""Test that lights are loaded properly."""
# Add light
with patch(
"axis.vapix.interfaces.light_control.LightControl.get_current_intensity",
return_value={"data": {"intensity": 100}},
), patch(
"axis.vapix.interfaces.light_control.LightControl.get_valid_intensity",
return_value={"data": {"ranges": [{"high": 150}]}},
):
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()
respx.post(
f"http://{DEFAULT_HOST}:80/axis-cgi/lightcontrol.cgi",
json={
"apiVersion": "1.1",
"context": CONTEXT,
"method": "getCurrentIntensity",
"params": {"lightID": "led0"},
},
).respond(
json={
"apiVersion": "1.1",
"context": "Axis library",
"method": "getCurrentIntensity",
"data": {"intensity": 100},
},
)
respx.post(
f"http://{DEFAULT_HOST}:80/axis-cgi/lightcontrol.cgi",
json={
"apiVersion": "1.1",
"context": CONTEXT,
"method": "getValidIntensity",
"params": {"lightID": "led0"},
},
).respond(
json={
"apiVersion": "1.1",
"context": "Axis library",
"method": "getValidIntensity",
"data": {"ranges": [{"low": 0, "high": 150}]},
},
)
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
@ -118,14 +159,9 @@ async def test_lights(hass: HomeAssistant, setup_config_entry, mock_rtsp_event)
assert light_0.name == f"{NAME} IR Light 0"
# Turn on, set brightness, light already on
with patch(
"axis.vapix.interfaces.light_control.LightControl.activate_light"
) as mock_activate, patch(
"axis.vapix.interfaces.light_control.LightControl.set_manual_intensity"
) as mock_set_intensity, patch(
"axis.vapix.interfaces.light_control.LightControl.get_current_intensity",
return_value={"data": {"intensity": 100}},
):
with patch("axis.vapix.vapix.LightHandler.activate_light") as mock_activate, patch(
"axis.vapix.vapix.LightHandler.set_manual_intensity"
) as mock_set_intensity:
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_ON,
@ -136,12 +172,7 @@ async def test_lights(hass: HomeAssistant, setup_config_entry, mock_rtsp_event)
mock_set_intensity.assert_called_once_with("led0", 29)
# Turn off
with patch(
"axis.vapix.interfaces.light_control.LightControl.deactivate_light"
) as mock_deactivate, patch(
"axis.vapix.interfaces.light_control.LightControl.get_current_intensity",
return_value={"data": {"intensity": 100}},
):
with patch("axis.vapix.vapix.LightHandler.deactivate_light") as mock_deactivate:
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_OFF,
@ -164,14 +195,9 @@ async def test_lights(hass: HomeAssistant, setup_config_entry, mock_rtsp_event)
assert light_0.state == STATE_OFF
# Turn on, set brightness
with patch(
"axis.vapix.interfaces.light_control.LightControl.activate_light"
) as mock_activate, patch(
"axis.vapix.interfaces.light_control.LightControl.set_manual_intensity"
) as mock_set_intensity, patch(
"axis.vapix.interfaces.light_control.LightControl.get_current_intensity",
return_value={"data": {"intensity": 100}},
):
with patch("axis.vapix.vapix.LightHandler.activate_light") as mock_activate, patch(
"axis.vapix.vapix.LightHandler.set_manual_intensity"
) as mock_set_intensity:
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_ON,
@ -182,12 +208,7 @@ async def test_lights(hass: HomeAssistant, setup_config_entry, mock_rtsp_event)
mock_set_intensity.assert_not_called()
# Turn off, light already off
with patch(
"axis.vapix.interfaces.light_control.LightControl.deactivate_light"
) as mock_deactivate, patch(
"axis.vapix.interfaces.light_control.LightControl.get_current_intensity",
return_value={"data": {"intensity": 100}},
):
with patch("axis.vapix.vapix.LightHandler.deactivate_light") as mock_deactivate:
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_OFF,

View File

@ -1,6 +1,7 @@
"""Axis switch platform tests."""
from unittest.mock import AsyncMock
from unittest.mock import patch
from axis.vapix.models.api import CONTEXT
import pytest
from homeassistant.components.axis.const import DOMAIN as AXIS_DOMAIN
@ -32,18 +33,22 @@ async def test_no_switches(hass: HomeAssistant, setup_config_entry) -> None:
assert not hass.states.async_entity_ids(SWITCH_DOMAIN)
PORT_DATA = """root.IOPort.I0.Configurable=yes
root.IOPort.I0.Direction=output
root.IOPort.I0.Output.Name=Doorbell
root.IOPort.I0.Output.Active=closed
root.IOPort.I1.Configurable=yes
root.IOPort.I1.Direction=output
root.IOPort.I1.Output.Name=
root.IOPort.I1.Output.Active=open
"""
@pytest.mark.parametrize("param_ports_payload", [PORT_DATA])
async def test_switches_with_port_cgi(
hass: HomeAssistant, setup_config_entry, mock_rtsp_event
) -> None:
"""Test that switches are loaded properly using port.cgi."""
device = hass.data[AXIS_DOMAIN][setup_config_entry.entry_id]
device.api.vapix.ports = {"0": AsyncMock(), "1": AsyncMock()}
device.api.vapix.ports["0"].name = "Doorbell"
device.api.vapix.ports["0"].open = AsyncMock()
device.api.vapix.ports["0"].close = AsyncMock()
device.api.vapix.ports["1"].name = ""
mock_rtsp_event(
topic="tns1:Device/Trigger/Relay",
data_type="LogicalState",
@ -72,36 +77,61 @@ async def test_switches_with_port_cgi(
assert relay_0.state == STATE_OFF
assert relay_0.name == f"{NAME} Doorbell"
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
device.api.vapix.ports["0"].close.assert_called_once()
with patch("axis.vapix.vapix.Ports.close") as mock_turn_on:
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
mock_turn_on.assert_called_once_with("0")
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
device.api.vapix.ports["0"].open.assert_called_once()
with patch("axis.vapix.vapix.Ports.open") as mock_turn_off:
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
mock_turn_off.assert_called_once_with("0")
PORT_MANAGEMENT_RESPONSE = {
"apiVersion": "1.0",
"method": "getPorts",
"context": CONTEXT,
"data": {
"numberOfPorts": 2,
"items": [
{
"port": "0",
"configurable": True,
"usage": "",
"name": "Doorbell",
"direction": "output",
"state": "open",
"normalState": "open",
},
{
"port": "1",
"configurable": True,
"usage": "",
"name": "",
"direction": "output",
"state": "open",
"normalState": "open",
},
],
},
}
@pytest.mark.parametrize("api_discovery_items", [API_DISCOVERY_PORT_MANAGEMENT])
@pytest.mark.parametrize("port_management_payload", [PORT_MANAGEMENT_RESPONSE])
async def test_switches_with_port_management(
hass: HomeAssistant, setup_config_entry, mock_rtsp_event
) -> None:
"""Test that switches are loaded properly using port management."""
device = hass.data[AXIS_DOMAIN][setup_config_entry.entry_id]
device.api.vapix.ports = {"0": AsyncMock(), "1": AsyncMock()}
device.api.vapix.ports["0"].name = "Doorbell"
device.api.vapix.ports["0"].open = AsyncMock()
device.api.vapix.ports["0"].close = AsyncMock()
device.api.vapix.ports["1"].name = ""
mock_rtsp_event(
topic="tns1:Device/Trigger/Relay",
data_type="LogicalState",
@ -143,18 +173,20 @@ async def test_switches_with_port_management(
assert hass.states.get(f"{SWITCH_DOMAIN}.{NAME}_relay_1").state == STATE_ON
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
device.api.vapix.ports["0"].close.assert_called_once()
with patch("axis.vapix.vapix.IoPortManagement.close") as mock_turn_on:
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
mock_turn_on.assert_called_once_with("0")
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
device.api.vapix.ports["0"].open.assert_called_once()
with patch("axis.vapix.vapix.IoPortManagement.open") as mock_turn_off:
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
mock_turn_off.assert_called_once_with("0")