mirror of
https://github.com/home-assistant/core.git
synced 2025-07-22 04:37:06 +00:00
Axis - Improve tests (#30415)
This commit is contained in:
parent
c1936f6fe4
commit
9b961632af
@ -1,12 +1,11 @@
|
|||||||
"""Axis binary sensor platform tests."""
|
"""Axis binary sensor platform tests."""
|
||||||
|
|
||||||
from unittest.mock import Mock
|
|
||||||
|
|
||||||
from homeassistant import config_entries
|
|
||||||
from homeassistant.components import axis
|
from homeassistant.components import axis
|
||||||
import homeassistant.components.binary_sensor as binary_sensor
|
import homeassistant.components.binary_sensor as binary_sensor
|
||||||
from homeassistant.setup import async_setup_component
|
from homeassistant.setup import async_setup_component
|
||||||
|
|
||||||
|
from .test_device import NAME, setup_axis_integration
|
||||||
|
|
||||||
EVENTS = [
|
EVENTS = [
|
||||||
{
|
{
|
||||||
"operation": "Initialized",
|
"operation": "Initialized",
|
||||||
@ -24,52 +23,6 @@ EVENTS = [
|
|||||||
},
|
},
|
||||||
]
|
]
|
||||||
|
|
||||||
ENTRY_CONFIG = {
|
|
||||||
axis.CONF_DEVICE: {
|
|
||||||
axis.config_flow.CONF_HOST: "1.2.3.4",
|
|
||||||
axis.config_flow.CONF_USERNAME: "user",
|
|
||||||
axis.config_flow.CONF_PASSWORD: "pass",
|
|
||||||
axis.config_flow.CONF_PORT: 80,
|
|
||||||
},
|
|
||||||
axis.config_flow.CONF_MAC: "1234ABCD",
|
|
||||||
axis.config_flow.CONF_MODEL: "model",
|
|
||||||
axis.config_flow.CONF_NAME: "model 0",
|
|
||||||
}
|
|
||||||
|
|
||||||
ENTRY_OPTIONS = {
|
|
||||||
axis.CONF_CAMERA: False,
|
|
||||||
axis.CONF_EVENTS: True,
|
|
||||||
axis.CONF_TRIGGER_TIME: 0,
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
async def setup_device(hass):
|
|
||||||
"""Load the Axis binary sensor platform."""
|
|
||||||
from axis import AxisDevice
|
|
||||||
|
|
||||||
loop = Mock()
|
|
||||||
|
|
||||||
config_entry = config_entries.ConfigEntry(
|
|
||||||
1,
|
|
||||||
axis.DOMAIN,
|
|
||||||
"Mock Title",
|
|
||||||
ENTRY_CONFIG,
|
|
||||||
"test",
|
|
||||||
config_entries.CONN_CLASS_LOCAL_PUSH,
|
|
||||||
system_options={},
|
|
||||||
options=ENTRY_OPTIONS,
|
|
||||||
)
|
|
||||||
device = axis.AxisNetworkDevice(hass, config_entry)
|
|
||||||
device.api = AxisDevice(loop=loop, **config_entry.data[axis.CONF_DEVICE])
|
|
||||||
hass.data[axis.DOMAIN] = {device.serial: device}
|
|
||||||
device.api.enable_events(event_callback=device.async_event_callback)
|
|
||||||
|
|
||||||
await hass.config_entries.async_forward_entry_setup(config_entry, "binary_sensor")
|
|
||||||
# To flush out the service call to update the group
|
|
||||||
await hass.async_block_till_done()
|
|
||||||
|
|
||||||
return device
|
|
||||||
|
|
||||||
|
|
||||||
async def test_platform_manually_configured(hass):
|
async def test_platform_manually_configured(hass):
|
||||||
"""Test that nothing happens when platform is manually configured."""
|
"""Test that nothing happens when platform is manually configured."""
|
||||||
@ -85,25 +38,25 @@ async def test_platform_manually_configured(hass):
|
|||||||
|
|
||||||
async def test_no_binary_sensors(hass):
|
async def test_no_binary_sensors(hass):
|
||||||
"""Test that no sensors in Axis results in no sensor entities."""
|
"""Test that no sensors in Axis results in no sensor entities."""
|
||||||
await setup_device(hass)
|
await setup_axis_integration(hass)
|
||||||
|
|
||||||
assert len(hass.states.async_all()) == 0
|
assert not hass.states.async_entity_ids("binary_sensor")
|
||||||
|
|
||||||
|
|
||||||
async def test_binary_sensors(hass):
|
async def test_binary_sensors(hass):
|
||||||
"""Test that sensors are loaded properly."""
|
"""Test that sensors are loaded properly."""
|
||||||
device = await setup_device(hass)
|
device = await setup_axis_integration(hass)
|
||||||
|
|
||||||
for event in EVENTS:
|
for event in EVENTS:
|
||||||
device.api.stream.event.manage_event(event)
|
device.api.stream.event.manage_event(event)
|
||||||
await hass.async_block_till_done()
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
assert len(hass.states.async_all()) == 2
|
assert len(hass.states.async_entity_ids("binary_sensor")) == 2
|
||||||
|
|
||||||
pir = hass.states.get("binary_sensor.model_0_pir_0")
|
pir = hass.states.get(f"binary_sensor.{NAME}_pir_0")
|
||||||
assert pir.state == "off"
|
assert pir.state == "off"
|
||||||
assert pir.name == "model 0 PIR 0"
|
assert pir.name == f"{NAME} PIR 0"
|
||||||
|
|
||||||
vmd4 = hass.states.get("binary_sensor.model_0_vmd4_camera1profile1")
|
vmd4 = hass.states.get(f"binary_sensor.{NAME}_vmd4_camera1profile1")
|
||||||
assert vmd4.state == "on"
|
assert vmd4.state == "on"
|
||||||
assert vmd4.name == "model 0 VMD4 Camera1Profile1"
|
assert vmd4.name == f"{NAME} VMD4 Camera1Profile1"
|
||||||
|
@ -1,57 +1,10 @@
|
|||||||
"""Axis camera platform tests."""
|
"""Axis camera platform tests."""
|
||||||
|
|
||||||
from unittest.mock import Mock
|
|
||||||
|
|
||||||
from homeassistant import config_entries
|
|
||||||
from homeassistant.components import axis
|
from homeassistant.components import axis
|
||||||
import homeassistant.components.camera as camera
|
import homeassistant.components.camera as camera
|
||||||
from homeassistant.setup import async_setup_component
|
from homeassistant.setup import async_setup_component
|
||||||
|
|
||||||
ENTRY_CONFIG = {
|
from .test_device import NAME, setup_axis_integration
|
||||||
axis.CONF_DEVICE: {
|
|
||||||
axis.config_flow.CONF_HOST: "1.2.3.4",
|
|
||||||
axis.config_flow.CONF_USERNAME: "user",
|
|
||||||
axis.config_flow.CONF_PASSWORD: "pass",
|
|
||||||
axis.config_flow.CONF_PORT: 80,
|
|
||||||
},
|
|
||||||
axis.config_flow.CONF_MAC: "1234ABCD",
|
|
||||||
axis.config_flow.CONF_MODEL: "model",
|
|
||||||
axis.config_flow.CONF_NAME: "model 0",
|
|
||||||
}
|
|
||||||
|
|
||||||
ENTRY_OPTIONS = {
|
|
||||||
axis.CONF_CAMERA: False,
|
|
||||||
axis.CONF_EVENTS: True,
|
|
||||||
axis.CONF_TRIGGER_TIME: 0,
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
async def setup_device(hass):
|
|
||||||
"""Load the Axis binary sensor platform."""
|
|
||||||
from axis import AxisDevice
|
|
||||||
|
|
||||||
loop = Mock()
|
|
||||||
|
|
||||||
config_entry = config_entries.ConfigEntry(
|
|
||||||
1,
|
|
||||||
axis.DOMAIN,
|
|
||||||
"Mock Title",
|
|
||||||
ENTRY_CONFIG,
|
|
||||||
"test",
|
|
||||||
config_entries.CONN_CLASS_LOCAL_PUSH,
|
|
||||||
system_options={},
|
|
||||||
options=ENTRY_OPTIONS,
|
|
||||||
)
|
|
||||||
device = axis.AxisNetworkDevice(hass, config_entry)
|
|
||||||
device.api = AxisDevice(loop=loop, **config_entry.data[axis.CONF_DEVICE])
|
|
||||||
hass.data[axis.DOMAIN] = {device.serial: device}
|
|
||||||
device.api.enable_events(event_callback=device.async_event_callback)
|
|
||||||
|
|
||||||
await hass.config_entries.async_forward_entry_setup(config_entry, "camera")
|
|
||||||
# To flush out the service call to update the group
|
|
||||||
await hass.async_block_till_done()
|
|
||||||
|
|
||||||
return device
|
|
||||||
|
|
||||||
|
|
||||||
async def test_platform_manually_configured(hass):
|
async def test_platform_manually_configured(hass):
|
||||||
@ -68,12 +21,10 @@ async def test_platform_manually_configured(hass):
|
|||||||
|
|
||||||
async def test_camera(hass):
|
async def test_camera(hass):
|
||||||
"""Test that Axis camera platform is loaded properly."""
|
"""Test that Axis camera platform is loaded properly."""
|
||||||
await setup_device(hass)
|
await setup_axis_integration(hass)
|
||||||
|
|
||||||
await hass.async_block_till_done()
|
assert len(hass.states.async_entity_ids("camera")) == 1
|
||||||
|
|
||||||
assert len(hass.states.async_all()) == 1
|
cam = hass.states.get(f"camera.{NAME}")
|
||||||
|
|
||||||
cam = hass.states.get("camera.model_0")
|
|
||||||
assert cam.state == "idle"
|
assert cam.state == "idle"
|
||||||
assert cam.name == "model 0"
|
assert cam.name == NAME
|
||||||
|
@ -60,7 +60,7 @@ async def test_flow_works(hass):
|
|||||||
)
|
)
|
||||||
|
|
||||||
assert result["type"] == "create_entry"
|
assert result["type"] == "create_entry"
|
||||||
assert result["title"] == "{} - {}".format("prodnbr", "serialnumber")
|
assert result["title"] == f"prodnbr - serialnumber"
|
||||||
assert result["data"] == {
|
assert result["data"] == {
|
||||||
axis.CONF_DEVICE: {
|
axis.CONF_DEVICE: {
|
||||||
config_flow.CONF_HOST: "1.2.3.4",
|
config_flow.CONF_HOST: "1.2.3.4",
|
||||||
|
@ -1,173 +1,180 @@
|
|||||||
"""Test Axis device."""
|
"""Test Axis device."""
|
||||||
from unittest.mock import Mock, patch
|
from copy import deepcopy
|
||||||
|
|
||||||
|
from asynctest import Mock, patch
|
||||||
|
import axis as axislib
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
from homeassistant.components.axis import device, errors
|
from homeassistant import config_entries
|
||||||
from homeassistant.components.axis.camera import AxisCamera
|
from homeassistant.components import axis
|
||||||
|
|
||||||
from tests.common import MockConfigEntry, mock_coro
|
MAC = "00408C12345"
|
||||||
|
MODEL = "model"
|
||||||
|
NAME = "name"
|
||||||
|
|
||||||
DEVICE_DATA = {
|
DEVICE_DATA = {
|
||||||
device.CONF_HOST: "1.2.3.4",
|
axis.device.CONF_HOST: "1.2.3.4",
|
||||||
device.CONF_USERNAME: "username",
|
axis.device.CONF_USERNAME: "username",
|
||||||
device.CONF_PASSWORD: "password",
|
axis.device.CONF_PASSWORD: "password",
|
||||||
device.CONF_PORT: 1234,
|
axis.device.CONF_PORT: 1234,
|
||||||
}
|
}
|
||||||
|
|
||||||
ENTRY_OPTIONS = {device.CONF_CAMERA: True, device.CONF_EVENTS: True}
|
ENTRY_OPTIONS = {axis.device.CONF_CAMERA: True, axis.device.CONF_EVENTS: True}
|
||||||
|
|
||||||
ENTRY_CONFIG = {
|
ENTRY_CONFIG = {
|
||||||
device.CONF_DEVICE: DEVICE_DATA,
|
axis.device.CONF_DEVICE: DEVICE_DATA,
|
||||||
device.CONF_MAC: "mac",
|
axis.device.CONF_MAC: MAC,
|
||||||
device.CONF_MODEL: "model",
|
axis.device.CONF_MODEL: MODEL,
|
||||||
device.CONF_NAME: "name",
|
axis.device.CONF_NAME: NAME,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
DEFAULT_BRAND = """root.Brand.Brand=AXIS
|
||||||
|
root.Brand.ProdFullName=AXIS M1065-LW Network Camera
|
||||||
|
root.Brand.ProdNbr=M1065-LW
|
||||||
|
root.Brand.ProdShortName=AXIS M1065-LW
|
||||||
|
root.Brand.ProdType=Network Camera
|
||||||
|
root.Brand.ProdVariant=
|
||||||
|
root.Brand.WebURL=http://www.axis.com
|
||||||
|
"""
|
||||||
|
|
||||||
async def test_device_setup():
|
DEFAULT_PORTS = """root.Input.NbrOfInputs=1
|
||||||
"""Successful setup."""
|
root.IOPort.I0.Configurable=no
|
||||||
hass = Mock()
|
root.IOPort.I0.Direction=input
|
||||||
entry = Mock()
|
root.IOPort.I0.Input.Name=PIR sensor
|
||||||
entry.data = ENTRY_CONFIG
|
root.IOPort.I0.Input.Trig=closed
|
||||||
entry.options = ENTRY_OPTIONS
|
root.Output.NbrOfOutputs=0
|
||||||
api = Mock()
|
"""
|
||||||
|
|
||||||
axis_device = device.AxisNetworkDevice(hass, entry)
|
DEFAULT_PROPERTIES = """root.Properties.API.HTTP.Version=3
|
||||||
axis_device.start = Mock()
|
root.Properties.API.Metadata.Metadata=yes
|
||||||
|
root.Properties.API.Metadata.Version=1.0
|
||||||
|
root.Properties.Firmware.BuildDate=Feb 15 2019 09:42
|
||||||
|
root.Properties.Firmware.BuildNumber=26
|
||||||
|
root.Properties.Firmware.Version=9.10.1
|
||||||
|
root.Properties.Image.Format=jpeg,mjpeg,h264
|
||||||
|
root.Properties.Image.NbrOfViews=2
|
||||||
|
root.Properties.Image.Resolution=1920x1080,1280x960,1280x720,1024x768,1024x576,800x600,640x480,640x360,352x240,320x240
|
||||||
|
root.Properties.Image.Rotation=0,180
|
||||||
|
root.Properties.System.SerialNumber=ACCC12345678
|
||||||
|
"""
|
||||||
|
|
||||||
assert axis_device.host == DEVICE_DATA[device.CONF_HOST]
|
|
||||||
assert axis_device.model == ENTRY_CONFIG[device.CONF_MODEL]
|
|
||||||
assert axis_device.name == ENTRY_CONFIG[device.CONF_NAME]
|
|
||||||
assert axis_device.serial == ENTRY_CONFIG[device.CONF_MAC]
|
|
||||||
|
|
||||||
with patch.object(device, "get_device", return_value=mock_coro(api)):
|
async def setup_axis_integration(
|
||||||
assert await axis_device.async_setup() is True
|
hass,
|
||||||
|
config=ENTRY_CONFIG,
|
||||||
assert axis_device.api is api
|
options=ENTRY_OPTIONS,
|
||||||
assert len(hass.config_entries.async_forward_entry_setup.mock_calls) == 3
|
brand=DEFAULT_BRAND,
|
||||||
assert hass.config_entries.async_forward_entry_setup.mock_calls[0][1] == (
|
ports=DEFAULT_PORTS,
|
||||||
entry,
|
properties=DEFAULT_PROPERTIES,
|
||||||
"camera",
|
):
|
||||||
)
|
"""Create the Axis device."""
|
||||||
assert hass.config_entries.async_forward_entry_setup.mock_calls[1][1] == (
|
config_entry = config_entries.ConfigEntry(
|
||||||
entry,
|
version=1,
|
||||||
"binary_sensor",
|
domain=axis.DOMAIN,
|
||||||
)
|
title="Mock Title",
|
||||||
assert hass.config_entries.async_forward_entry_setup.mock_calls[2][1] == (
|
data=deepcopy(config),
|
||||||
entry,
|
source="test",
|
||||||
"switch",
|
connection_class=config_entries.CONN_CLASS_LOCAL_PUSH,
|
||||||
|
system_options={},
|
||||||
|
options=deepcopy(options),
|
||||||
|
entry_id="1",
|
||||||
)
|
)
|
||||||
|
|
||||||
|
def mock_request(self, method, path, json=None):
|
||||||
|
if method == "get":
|
||||||
|
if path == "/axis-cgi/param.cgi?action=list&group=root.Brand":
|
||||||
|
return brand
|
||||||
|
if path in [
|
||||||
|
"/axis-cgi/param.cgi?action=list&group=root.Input",
|
||||||
|
"/axis-cgi/param.cgi?action=list&group=root.IOPort",
|
||||||
|
"/axis-cgi/param.cgi?action=list&group=root.Output",
|
||||||
|
]:
|
||||||
|
return ports
|
||||||
|
if path == "/axis-cgi/param.cgi?action=list&group=root.Properties":
|
||||||
|
return properties
|
||||||
|
|
||||||
async def test_device_signal_new_address(hass):
|
return None
|
||||||
"""Successful setup."""
|
|
||||||
entry = MockConfigEntry(
|
|
||||||
domain=device.DOMAIN, data=ENTRY_CONFIG, options=ENTRY_OPTIONS
|
|
||||||
)
|
|
||||||
|
|
||||||
api = Mock()
|
with patch("axis.vapix.Vapix.request", new=mock_request), patch(
|
||||||
api.vapix.get_param.return_value = "1234"
|
"axis.AxisDevice.start", return_value=True
|
||||||
|
):
|
||||||
axis_device = device.AxisNetworkDevice(hass, entry)
|
await axis.async_setup_entry(hass, config_entry)
|
||||||
hass.data[device.DOMAIN] = {axis_device.serial: axis_device}
|
|
||||||
|
|
||||||
with patch.object(device, "get_device", return_value=mock_coro(api)), patch.object(
|
|
||||||
AxisCamera, "_new_address"
|
|
||||||
) as new_address_mock:
|
|
||||||
await axis_device.async_setup()
|
|
||||||
await hass.async_block_till_done()
|
|
||||||
|
|
||||||
assert len(hass.states.async_all()) == 1
|
|
||||||
assert len(axis_device.listeners) == 2
|
|
||||||
|
|
||||||
entry.data[device.CONF_DEVICE][device.CONF_HOST] = "2.3.4.5"
|
|
||||||
hass.config_entries.async_update_entry(entry, data=entry.data)
|
|
||||||
await hass.async_block_till_done()
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
assert axis_device.host == "2.3.4.5"
|
hass.config_entries._entries.append(config_entry)
|
||||||
assert axis_device.api.config.host == "2.3.4.5"
|
|
||||||
assert len(new_address_mock.mock_calls) == 1
|
return hass.data[axis.DOMAIN].get(config[axis.CONF_MAC])
|
||||||
|
|
||||||
|
|
||||||
|
async def test_device_setup(hass):
|
||||||
|
"""Successful setup."""
|
||||||
|
with patch(
|
||||||
|
"homeassistant.config_entries.ConfigEntries.async_forward_entry_setup",
|
||||||
|
return_value=True,
|
||||||
|
) as forward_entry_setup:
|
||||||
|
device = await setup_axis_integration(hass)
|
||||||
|
|
||||||
|
entry = device.config_entry
|
||||||
|
|
||||||
|
assert len(forward_entry_setup.mock_calls) == 3
|
||||||
|
assert forward_entry_setup.mock_calls[0][1] == (entry, "camera")
|
||||||
|
assert forward_entry_setup.mock_calls[1][1] == (entry, "binary_sensor")
|
||||||
|
assert forward_entry_setup.mock_calls[2][1] == (entry, "switch")
|
||||||
|
|
||||||
|
assert device.host == DEVICE_DATA[axis.device.CONF_HOST]
|
||||||
|
assert device.model == ENTRY_CONFIG[axis.device.CONF_MODEL]
|
||||||
|
assert device.name == ENTRY_CONFIG[axis.device.CONF_NAME]
|
||||||
|
assert device.serial == ENTRY_CONFIG[axis.device.CONF_MAC]
|
||||||
|
|
||||||
|
|
||||||
|
async def test_update_address(hass):
|
||||||
|
"""Test update address works."""
|
||||||
|
device = await setup_axis_integration(hass)
|
||||||
|
assert device.api.config.host == "1.2.3.4"
|
||||||
|
|
||||||
|
await hass.config_entries.flow.async_init(
|
||||||
|
axis.DOMAIN,
|
||||||
|
data={
|
||||||
|
"host": "2.3.4.5",
|
||||||
|
"port": 80,
|
||||||
|
"hostname": "name",
|
||||||
|
"properties": {"macaddress": MAC},
|
||||||
|
},
|
||||||
|
context={"source": "zeroconf"},
|
||||||
|
)
|
||||||
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
|
assert device.api.config.host == "2.3.4.5"
|
||||||
|
|
||||||
|
|
||||||
async def test_device_unavailable(hass):
|
async def test_device_unavailable(hass):
|
||||||
"""Successful setup."""
|
"""Successful setup."""
|
||||||
entry = MockConfigEntry(
|
device = await setup_axis_integration(hass)
|
||||||
domain=device.DOMAIN, data=ENTRY_CONFIG, options=ENTRY_OPTIONS
|
device.async_connection_status_callback(status=False)
|
||||||
)
|
assert not device.available
|
||||||
|
|
||||||
api = Mock()
|
|
||||||
api.vapix.get_param.return_value = "1234"
|
|
||||||
|
|
||||||
axis_device = device.AxisNetworkDevice(hass, entry)
|
|
||||||
hass.data[device.DOMAIN] = {axis_device.serial: axis_device}
|
|
||||||
|
|
||||||
with patch.object(device, "get_device", return_value=mock_coro(api)), patch.object(
|
|
||||||
device, "async_dispatcher_send"
|
|
||||||
) as mock_dispatcher:
|
|
||||||
await axis_device.async_setup()
|
|
||||||
await hass.async_block_till_done()
|
|
||||||
|
|
||||||
axis_device.async_connection_status_callback(status=False)
|
|
||||||
|
|
||||||
assert not axis_device.available
|
|
||||||
assert len(mock_dispatcher.mock_calls) == 1
|
|
||||||
|
|
||||||
|
|
||||||
async def test_device_reset(hass):
|
async def test_device_reset(hass):
|
||||||
"""Successfully reset device."""
|
"""Successfully reset device."""
|
||||||
entry = MockConfigEntry(
|
device = await setup_axis_integration(hass)
|
||||||
domain=device.DOMAIN, data=ENTRY_CONFIG, options=ENTRY_OPTIONS
|
result = await device.async_reset()
|
||||||
)
|
assert result is True
|
||||||
|
|
||||||
api = Mock()
|
|
||||||
api.vapix.get_param.return_value = "1234"
|
|
||||||
|
|
||||||
axis_device = device.AxisNetworkDevice(hass, entry)
|
|
||||||
hass.data[device.DOMAIN] = {axis_device.serial: axis_device}
|
|
||||||
|
|
||||||
with patch.object(device, "get_device", return_value=mock_coro(api)):
|
|
||||||
await axis_device.async_setup()
|
|
||||||
await hass.async_block_till_done()
|
|
||||||
|
|
||||||
await axis_device.async_reset()
|
|
||||||
|
|
||||||
assert len(api.stop.mock_calls) == 1
|
|
||||||
assert len(hass.states.async_all()) == 0
|
|
||||||
assert len(axis_device.listeners) == 0
|
|
||||||
|
|
||||||
|
|
||||||
async def test_device_not_accessible():
|
async def test_device_not_accessible(hass):
|
||||||
"""Failed setup schedules a retry of setup."""
|
"""Failed setup schedules a retry of setup."""
|
||||||
hass = Mock()
|
|
||||||
hass.data = dict()
|
|
||||||
entry = Mock()
|
|
||||||
entry.data = ENTRY_CONFIG
|
|
||||||
entry.options = ENTRY_OPTIONS
|
|
||||||
|
|
||||||
axis_device = device.AxisNetworkDevice(hass, entry)
|
|
||||||
|
|
||||||
with patch.object(
|
with patch.object(
|
||||||
device, "get_device", side_effect=errors.CannotConnect
|
axis.device, "get_device", side_effect=axis.errors.CannotConnect
|
||||||
), pytest.raises(device.ConfigEntryNotReady):
|
), pytest.raises(axis.device.ConfigEntryNotReady):
|
||||||
await axis_device.async_setup()
|
await setup_axis_integration(hass)
|
||||||
|
assert hass.data[axis.DOMAIN] == {}
|
||||||
assert not hass.helpers.event.async_call_later.mock_calls
|
|
||||||
|
|
||||||
|
|
||||||
async def test_device_unknown_error():
|
async def test_device_unknown_error(hass):
|
||||||
"""Unknown errors are handled."""
|
"""Unknown errors are handled."""
|
||||||
hass = Mock()
|
with patch.object(axis.device, "get_device", side_effect=Exception):
|
||||||
entry = Mock()
|
await setup_axis_integration(hass)
|
||||||
entry.data = ENTRY_CONFIG
|
assert hass.data[axis.DOMAIN] == {}
|
||||||
entry.options = ENTRY_OPTIONS
|
|
||||||
|
|
||||||
axis_device = device.AxisNetworkDevice(hass, entry)
|
|
||||||
|
|
||||||
with patch.object(device, "get_device", side_effect=Exception):
|
|
||||||
assert await axis_device.async_setup() is False
|
|
||||||
|
|
||||||
assert not hass.helpers.event.async_call_later.mock_calls
|
|
||||||
|
|
||||||
|
|
||||||
async def test_new_event_sends_signal(hass):
|
async def test_new_event_sends_signal(hass):
|
||||||
@ -175,9 +182,9 @@ async def test_new_event_sends_signal(hass):
|
|||||||
entry = Mock()
|
entry = Mock()
|
||||||
entry.data = ENTRY_CONFIG
|
entry.data = ENTRY_CONFIG
|
||||||
|
|
||||||
axis_device = device.AxisNetworkDevice(hass, entry)
|
axis_device = axis.device.AxisNetworkDevice(hass, entry)
|
||||||
|
|
||||||
with patch.object(device, "async_dispatcher_send") as mock_dispatch_send:
|
with patch.object(axis.device, "async_dispatcher_send") as mock_dispatch_send:
|
||||||
axis_device.async_event_callback(action="add", event_id="event")
|
axis_device.async_event_callback(action="add", event_id="event")
|
||||||
await hass.async_block_till_done()
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
@ -191,7 +198,7 @@ async def test_shutdown():
|
|||||||
entry = Mock()
|
entry = Mock()
|
||||||
entry.data = ENTRY_CONFIG
|
entry.data = ENTRY_CONFIG
|
||||||
|
|
||||||
axis_device = device.AxisNetworkDevice(hass, entry)
|
axis_device = axis.device.AxisNetworkDevice(hass, entry)
|
||||||
axis_device.api = Mock()
|
axis_device.api = Mock()
|
||||||
|
|
||||||
axis_device.shutdown(None)
|
axis_device.shutdown(None)
|
||||||
@ -199,39 +206,25 @@ async def test_shutdown():
|
|||||||
assert len(axis_device.api.stop.mock_calls) == 1
|
assert len(axis_device.api.stop.mock_calls) == 1
|
||||||
|
|
||||||
|
|
||||||
async def test_get_device(hass):
|
|
||||||
"""Successful call."""
|
|
||||||
with patch("axis.param_cgi.Params.update_brand", return_value=mock_coro()), patch(
|
|
||||||
"axis.param_cgi.Params.update_properties", return_value=mock_coro()
|
|
||||||
), patch("axis.port_cgi.Ports.update", return_value=mock_coro()):
|
|
||||||
assert await device.get_device(hass, DEVICE_DATA)
|
|
||||||
|
|
||||||
|
|
||||||
async def test_get_device_fails(hass):
|
async def test_get_device_fails(hass):
|
||||||
"""Device unauthorized yields authentication required error."""
|
"""Device unauthorized yields authentication required error."""
|
||||||
import axis
|
|
||||||
|
|
||||||
with patch(
|
with patch(
|
||||||
"axis.param_cgi.Params.update_brand", side_effect=axis.Unauthorized
|
"axis.param_cgi.Params.update_brand", side_effect=axislib.Unauthorized
|
||||||
), pytest.raises(errors.AuthenticationRequired):
|
), pytest.raises(axis.errors.AuthenticationRequired):
|
||||||
await device.get_device(hass, DEVICE_DATA)
|
await axis.device.get_device(hass, DEVICE_DATA)
|
||||||
|
|
||||||
|
|
||||||
async def test_get_device_device_unavailable(hass):
|
async def test_get_device_device_unavailable(hass):
|
||||||
"""Device unavailable yields cannot connect error."""
|
"""Device unavailable yields cannot connect error."""
|
||||||
import axis
|
|
||||||
|
|
||||||
with patch(
|
with patch(
|
||||||
"axis.param_cgi.Params.update_brand", side_effect=axis.RequestError
|
"axis.param_cgi.Params.update_brand", side_effect=axislib.RequestError
|
||||||
), pytest.raises(errors.CannotConnect):
|
), pytest.raises(axis.errors.CannotConnect):
|
||||||
await device.get_device(hass, DEVICE_DATA)
|
await axis.device.get_device(hass, DEVICE_DATA)
|
||||||
|
|
||||||
|
|
||||||
async def test_get_device_unknown_error(hass):
|
async def test_get_device_unknown_error(hass):
|
||||||
"""Device yield unknown error."""
|
"""Device yield unknown error."""
|
||||||
import axis
|
|
||||||
|
|
||||||
with patch(
|
with patch(
|
||||||
"axis.param_cgi.Params.update_brand", side_effect=axis.AxisException
|
"axis.param_cgi.Params.update_brand", side_effect=axislib.AxisException
|
||||||
), pytest.raises(errors.AuthenticationRequired):
|
), pytest.raises(axis.errors.AuthenticationRequired):
|
||||||
await device.get_device(hass, DEVICE_DATA)
|
await axis.device.get_device(hass, DEVICE_DATA)
|
||||||
|
@ -2,11 +2,12 @@
|
|||||||
|
|
||||||
from unittest.mock import Mock, call as mock_call
|
from unittest.mock import Mock, call as mock_call
|
||||||
|
|
||||||
from homeassistant import config_entries
|
|
||||||
from homeassistant.components import axis
|
from homeassistant.components import axis
|
||||||
import homeassistant.components.switch as switch
|
import homeassistant.components.switch as switch
|
||||||
from homeassistant.setup import async_setup_component
|
from homeassistant.setup import async_setup_component
|
||||||
|
|
||||||
|
from .test_device import NAME, setup_axis_integration
|
||||||
|
|
||||||
EVENTS = [
|
EVENTS = [
|
||||||
{
|
{
|
||||||
"operation": "Initialized",
|
"operation": "Initialized",
|
||||||
@ -26,52 +27,6 @@ EVENTS = [
|
|||||||
},
|
},
|
||||||
]
|
]
|
||||||
|
|
||||||
ENTRY_CONFIG = {
|
|
||||||
axis.CONF_DEVICE: {
|
|
||||||
axis.config_flow.CONF_HOST: "1.2.3.4",
|
|
||||||
axis.config_flow.CONF_USERNAME: "user",
|
|
||||||
axis.config_flow.CONF_PASSWORD: "pass",
|
|
||||||
axis.config_flow.CONF_PORT: 80,
|
|
||||||
},
|
|
||||||
axis.config_flow.CONF_MAC: "1234ABCD",
|
|
||||||
axis.config_flow.CONF_MODEL: "model",
|
|
||||||
axis.config_flow.CONF_NAME: "model 0",
|
|
||||||
}
|
|
||||||
|
|
||||||
ENTRY_OPTIONS = {
|
|
||||||
axis.CONF_CAMERA: False,
|
|
||||||
axis.CONF_EVENTS: True,
|
|
||||||
axis.CONF_TRIGGER_TIME: 0,
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
async def setup_device(hass):
|
|
||||||
"""Load the Axis switch platform."""
|
|
||||||
from axis import AxisDevice
|
|
||||||
|
|
||||||
loop = Mock()
|
|
||||||
|
|
||||||
config_entry = config_entries.ConfigEntry(
|
|
||||||
1,
|
|
||||||
axis.DOMAIN,
|
|
||||||
"Mock Title",
|
|
||||||
ENTRY_CONFIG,
|
|
||||||
"test",
|
|
||||||
config_entries.CONN_CLASS_LOCAL_PUSH,
|
|
||||||
system_options={},
|
|
||||||
options=ENTRY_OPTIONS,
|
|
||||||
)
|
|
||||||
device = axis.AxisNetworkDevice(hass, config_entry)
|
|
||||||
device.api = AxisDevice(loop=loop, **config_entry.data[axis.CONF_DEVICE])
|
|
||||||
hass.data[axis.DOMAIN] = {device.serial: device}
|
|
||||||
device.api.enable_events(event_callback=device.async_event_callback)
|
|
||||||
|
|
||||||
await hass.config_entries.async_forward_entry_setup(config_entry, "switch")
|
|
||||||
# To flush out the service call to update the group
|
|
||||||
await hass.async_block_till_done()
|
|
||||||
|
|
||||||
return device
|
|
||||||
|
|
||||||
|
|
||||||
async def test_platform_manually_configured(hass):
|
async def test_platform_manually_configured(hass):
|
||||||
"""Test that nothing happens when platform is manually configured."""
|
"""Test that nothing happens when platform is manually configured."""
|
||||||
@ -84,14 +39,15 @@ async def test_platform_manually_configured(hass):
|
|||||||
|
|
||||||
async def test_no_switches(hass):
|
async def test_no_switches(hass):
|
||||||
"""Test that no output events in Axis results in no switch entities."""
|
"""Test that no output events in Axis results in no switch entities."""
|
||||||
await setup_device(hass)
|
await setup_axis_integration(hass)
|
||||||
|
|
||||||
assert not hass.states.async_entity_ids("switch")
|
assert not hass.states.async_entity_ids("switch")
|
||||||
|
|
||||||
|
|
||||||
async def test_switches(hass):
|
async def test_switches(hass):
|
||||||
"""Test that switches are loaded properly."""
|
"""Test that switches are loaded properly."""
|
||||||
device = await setup_device(hass)
|
device = await setup_axis_integration(hass)
|
||||||
|
|
||||||
device.api.vapix.ports = {"0": Mock(), "1": Mock()}
|
device.api.vapix.ports = {"0": Mock(), "1": Mock()}
|
||||||
device.api.vapix.ports["0"].name = "Doorbell"
|
device.api.vapix.ports["0"].name = "Doorbell"
|
||||||
device.api.vapix.ports["1"].name = ""
|
device.api.vapix.ports["1"].name = ""
|
||||||
@ -100,24 +56,24 @@ async def test_switches(hass):
|
|||||||
device.api.stream.event.manage_event(event)
|
device.api.stream.event.manage_event(event)
|
||||||
await hass.async_block_till_done()
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
assert len(hass.states.async_all()) == 3
|
assert len(hass.states.async_entity_ids("switch")) == 2
|
||||||
|
|
||||||
relay_0 = hass.states.get("switch.model_0_doorbell")
|
relay_0 = hass.states.get(f"switch.{NAME}_doorbell")
|
||||||
assert relay_0.state == "off"
|
assert relay_0.state == "off"
|
||||||
assert relay_0.name == "model 0 Doorbell"
|
assert relay_0.name == f"{NAME} Doorbell"
|
||||||
|
|
||||||
relay_1 = hass.states.get("switch.model_0_relay_1")
|
relay_1 = hass.states.get(f"switch.{NAME}_relay_1")
|
||||||
assert relay_1.state == "on"
|
assert relay_1.state == "on"
|
||||||
assert relay_1.name == "model 0 Relay 1"
|
assert relay_1.name == f"{NAME} Relay 1"
|
||||||
|
|
||||||
device.api.vapix.ports["0"].action = Mock()
|
device.api.vapix.ports["0"].action = Mock()
|
||||||
|
|
||||||
await hass.services.async_call(
|
await hass.services.async_call(
|
||||||
"switch", "turn_on", {"entity_id": "switch.model_0_doorbell"}, blocking=True
|
"switch", "turn_on", {"entity_id": f"switch.{NAME}_doorbell"}, blocking=True
|
||||||
)
|
)
|
||||||
|
|
||||||
await hass.services.async_call(
|
await hass.services.async_call(
|
||||||
"switch", "turn_off", {"entity_id": "switch.model_0_doorbell"}, blocking=True
|
"switch", "turn_off", {"entity_id": f"switch.{NAME}_doorbell"}, blocking=True
|
||||||
)
|
)
|
||||||
|
|
||||||
assert device.api.vapix.ports["0"].action.call_args_list == [
|
assert device.api.vapix.ports["0"].action.call_args_list == [
|
||||||
|
Loading…
x
Reference in New Issue
Block a user