Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Axis - Code improvements (#35592) 

* Adapt library improvements
Clean up integration and tests and make them more like latest changes in UniFi integration

* Bump dependency to v26
pull/35624/head
Robert Svensson 2020-05-14 10:49:27 +02:00 committed by GitHub
parent cf50ccb919
commit e6c58c9795
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
16 changed files with 323 additions and 307 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
homeassistant/components/axis/__init__.py
View File

@ -2,19 +2,10 @@
import logging
from homeassistant.const import (
CONF_DEVICE,
CONF_HOST,
CONF_MAC,
CONF_PASSWORD,
CONF_PORT,
CONF_TRIGGER_TIME,
CONF_USERNAME,
EVENT_HOMEASSISTANT_STOP,
)
from homeassistant.const import CONF_DEVICE, EVENT_HOMEASSISTANT_STOP
from .const import CONF_CAMERA, CONF_EVENTS, DEFAULT_TRIGGER_TIME, DOMAIN
from .device import AxisNetworkDevice, get_device
from .const import DOMAIN as AXIS_DOMAIN
from .device import AxisNetworkDevice
LOGGER = logging.getLogger(__name__)
@ -26,11 +17,7 @@ async def async_setup(hass, config):
async def async_setup_entry(hass, config_entry):
"""Set up the Axis component."""
if DOMAIN not in hass.data:
hass.data[DOMAIN] = {}
if not config_entry.options:
await async_populate_options(hass, config_entry)
hass.data.setdefault(AXIS_DOMAIN, {})
device = AxisNetworkDevice(hass, config_entry)
@ -43,7 +30,7 @@ async def async_setup_entry(hass, config_entry):
config_entry, unique_id=device.api.vapix.params.system_serialnumber
)
hass.data[DOMAIN][config_entry.unique_id] = device
hass.data[AXIS_DOMAIN][config_entry.unique_id] = device
await device.async_update_device_registry()
@ -54,32 +41,10 @@ async def async_setup_entry(hass, config_entry):
async def async_unload_entry(hass, config_entry):
"""Unload Axis device config entry."""
device = hass.data[DOMAIN].pop(config_entry.data[CONF_MAC])
device = hass.data[AXIS_DOMAIN].pop(config_entry.unique_id)
return await device.async_reset()
async def async_populate_options(hass, config_entry):
"""Populate default options for device."""
device = await get_device(
hass,
host=config_entry.data[CONF_HOST],
port=config_entry.data[CONF_PORT],
username=config_entry.data[CONF_USERNAME],
password=config_entry.data[CONF_PASSWORD],
)
supported_formats = device.vapix.params.image_format
camera = bool(supported_formats)
options = {
CONF_CAMERA: camera,
CONF_EVENTS: True,
CONF_TRIGGER_TIME: DEFAULT_TRIGGER_TIME,
}
hass.config_entries.async_update_entry(config_entry, options=options)
async def async_migrate_entry(hass, config_entry):
"""Migrate old entry."""
LOGGER.debug("Migrating from version %s", config_entry.version)

5
homeassistant/components/axis/axis_base.py
View File

@ -18,7 +18,7 @@ class AxisEntityBase(Entity):
"""Subscribe device events."""
self.async_on_remove(
async_dispatcher_connect(
self.hass, self.device.event_reachable, self.update_callback
self.hass, self.device.signal_reachable, self.update_callback
)
)
@ -49,15 +49,12 @@ class AxisEventBase(AxisEntityBase):
async def async_added_to_hass(self) -> None:
"""Subscribe sensors events."""
self.event.register_callback(self.update_callback)
await super().async_added_to_hass()
async def async_will_remove_from_hass(self) -> None:
"""Disconnect device object when removed."""
self.event.remove_callback(self.update_callback)
await super().async_will_remove_from_hass()
@property
def device_class(self):
"""Return the class of the event."""

34
homeassistant/components/axis/binary_sensor.py
View File

@ -5,7 +5,6 @@ from datetime import timedelta
from axis.event_stream import CLASS_INPUT, CLASS_OUTPUT
from homeassistant.components.binary_sensor import BinarySensorEntity
from homeassistant.const import CONF_TRIGGER_TIME
from homeassistant.core import callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.event import async_track_point_in_utc_time
@ -22,13 +21,13 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
@callback
def async_add_sensor(event_id):
"""Add binary sensor from Axis device."""
event = device.api.event.events[event_id]
event = device.api.event[event_id]
if event.CLASS != CLASS_OUTPUT:
async_add_entities([AxisBinarySensor(event, device)], True)
device.listeners.append(
async_dispatcher_connect(hass, device.event_new_sensor, async_add_sensor)
async_dispatcher_connect(hass, device.signal_new_event, async_add_sensor)
)
@ -38,7 +37,7 @@ class AxisBinarySensor(AxisEventBase, BinarySensorEntity):
def __init__(self, event, device):
"""Initialize the Axis binary sensor."""
super().__init__(event, device)
self.remove_timer = None
self.cancel_scheduled_update = None
@callback
def update_callback(self, no_delay=False):
@ -46,24 +45,25 @@ class AxisBinarySensor(AxisEventBase, BinarySensorEntity):
Parameter no_delay is True when device_event_reachable is sent.
"""
delay = self.device.config_entry.options[CONF_TRIGGER_TIME]
if self.remove_timer is not None:
self.remove_timer()
self.remove_timer = None
@callback
def scheduled_update(now):
"""Timer callback for sensor update."""
self.cancel_scheduled_update = None
self.async_write_ha_state()
if self.is_on or delay == 0 or no_delay:
if self.cancel_scheduled_update is not None:
self.cancel_scheduled_update()
self.cancel_scheduled_update = None
if self.is_on or self.device.option_trigger_time == 0 or no_delay:
self.async_write_ha_state()
return
@callback
def _delay_update(now):
"""Timer callback for sensor update."""
self.async_write_ha_state()
self.remove_timer = None
self.remove_timer = async_track_point_in_utc_time(
self.hass, _delay_update, utcnow() + timedelta(seconds=delay)
self.cancel_scheduled_update = async_track_point_in_utc_time(
self.hass,
scheduled_update,
utcnow() + timedelta(seconds=self.device.option_trigger_time),
)
@property

2
homeassistant/components/axis/camera.py
View File

@ -59,7 +59,7 @@ class AxisCamera(AxisEntityBase, MjpegCamera):
"""Subscribe camera events."""
self.async_on_remove(
async_dispatcher_connect(
self.hass, self.device.event_new_address, self._new_address
self.hass, self.device.signal_new_address, self._new_address
)
)

3
homeassistant/components/axis/const.py
View File

@ -5,8 +5,11 @@ LOGGER = logging.getLogger(__package__)
DOMAIN = "axis"
ATTR_MANUFACTURER = "Axis Communications AB"
CONF_CAMERA = "camera"
CONF_EVENTS = "events"
CONF_MODEL = "model"
DEFAULT_EVENTS = True
DEFAULT_TRIGGER_TIME = 0

143
homeassistant/components/axis/device.py
View File

@ -4,13 +4,18 @@ import asyncio
import async_timeout
import axis
from axis.event_stream import OPERATION_INITIALIZED
from axis.streammanager import SIGNAL_PLAYING
from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR_DOMAIN
from homeassistant.components.camera import DOMAIN as CAMERA_DOMAIN
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.const import (
CONF_HOST,
CONF_NAME,
CONF_PASSWORD,
CONF_PORT,
CONF_TRIGGER_TIME,
CONF_USERNAME,
)
from homeassistant.core import callback
@ -18,7 +23,16 @@ from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC
from homeassistant.helpers.dispatcher import async_dispatcher_send
from .const import CONF_CAMERA, CONF_EVENTS, CONF_MODEL, DOMAIN, LOGGER
from .const import (
ATTR_MANUFACTURER,
CONF_CAMERA,
CONF_EVENTS,
CONF_MODEL,
DEFAULT_EVENTS,
DEFAULT_TRIGGER_TIME,
DOMAIN as AXIS_DOMAIN,
LOGGER,
)
from .errors import AuthenticationRequired, CannotConnect
@ -57,14 +71,74 @@ class AxisNetworkDevice:
"""Return the serial number of this device."""
return self.config_entry.unique_id
@property
def option_camera(self):
"""Config entry option defining if camera should be used."""
supported_formats = self.api.vapix.params.image_format
return self.config_entry.options.get(CONF_CAMERA, bool(supported_formats))
@property
def option_events(self):
"""Config entry option defining if platforms based on events should be created."""
return self.config_entry.options.get(CONF_EVENTS, DEFAULT_EVENTS)
@property
def option_trigger_time(self):
"""Config entry option defining minimum number of seconds to keep trigger high."""
return self.config_entry.options.get(CONF_TRIGGER_TIME, DEFAULT_TRIGGER_TIME)
@property
def signal_reachable(self):
"""Device specific event to signal a change in connection status."""
return f"axis_reachable_{self.serial}"
@property
def signal_new_event(self):
"""Device specific event to signal new device event available."""
return f"axis_new_event_{self.serial}"
@property
def signal_new_address(self):
"""Device specific event to signal a change in device address."""
return f"axis_new_address_{self.serial}"
@callback
def async_connection_status_callback(self, status):
"""Handle signals of device connection status.
This is called on every RTSP keep-alive message.
Only signal state change if state change is true.
"""
if self.available != (status == SIGNAL_PLAYING):
self.available = not self.available
async_dispatcher_send(self.hass, self.signal_reachable, True)
@callback
def async_event_callback(self, action, event_id):
"""Call to configure events when initialized on event stream."""
if action == OPERATION_INITIALIZED:
async_dispatcher_send(self.hass, self.signal_new_event, event_id)
@staticmethod
async def async_new_address_callback(hass, entry):
"""Handle signals of device getting new address.
This is a static method because a class method (bound method),
can not be used with weak references.
"""
device = hass.data[AXIS_DOMAIN][entry.unique_id]
device.api.config.host = device.host
async_dispatcher_send(hass, device.signal_new_address)
async def async_update_device_registry(self):
"""Update device registry."""
device_registry = await self.hass.helpers.device_registry.async_get_registry()
device_registry.async_get_or_create(
config_entry_id=self.config_entry.entry_id,
connections={(CONNECTION_NETWORK_MAC, self.serial)},
identifiers={(DOMAIN, self.serial)},
manufacturer="Axis Communications AB",
identifiers={(AXIS_DOMAIN, self.serial)},
manufacturer=ATTR_MANUFACTURER,
model=f"{self.model} {self.product_type}",
name=self.name,
sw_version=self.fw_version,
@ -91,15 +165,15 @@ class AxisNetworkDevice:
self.fw_version = self.api.vapix.params.firmware_version
self.product_type = self.api.vapix.params.prodtype
if self.config_entry.options[CONF_CAMERA]:
if self.option_camera:
self.hass.async_create_task(
self.hass.config_entries.async_forward_entry_setup(
self.config_entry, "camera"
self.config_entry, CAMERA_DOMAIN
)
)
if self.config_entry.options[CONF_EVENTS]:
if self.option_events:
self.api.stream.connection_status_callback = (
self.async_connection_status_callback
@ -110,7 +184,7 @@ class AxisNetworkDevice:
self.hass.config_entries.async_forward_entry_setup(
self.config_entry, platform
)
for platform in ["binary_sensor", "switch"]
for platform in [BINARY_SENSOR_DOMAIN, SWITCH_DOMAIN]
]
self.hass.async_create_task(self.start(platform_tasks))
@ -118,50 +192,6 @@ class AxisNetworkDevice:
return True
@property
def event_new_address(self):
"""Device specific event to signal new device address."""
return f"axis_new_address_{self.serial}"
@staticmethod
async def async_new_address_callback(hass, entry):
"""Handle signals of device getting new address.
This is a static method because a class method (bound method),
can not be used with weak references.
"""
device = hass.data[DOMAIN][entry.unique_id]
device.api.config.host = device.host
async_dispatcher_send(hass, device.event_new_address)
@property
def event_reachable(self):
"""Device specific event to signal a change in connection status."""
return f"axis_reachable_{self.serial}"
@callback
def async_connection_status_callback(self, status):
"""Handle signals of device connection status.
This is called on every RTSP keep-alive message.
Only signal state change if state change is true.
"""
if self.available != (status == SIGNAL_PLAYING):
self.available = not self.available
async_dispatcher_send(self.hass, self.event_reachable, True)
@property
def event_new_sensor(self):
"""Device specific event to signal new sensor available."""
return f"axis_add_sensor_{self.serial}"
@callback
def async_event_callback(self, action, event_id):
"""Call to configure events when initialized on event stream."""
if action == "add":
async_dispatcher_send(self.hass, self.event_new_sensor, event_id)
async def start(self, platform_tasks):
"""Start the event stream when all platforms are loaded."""
await asyncio.gather(*platform_tasks)
@ -179,7 +209,7 @@ class AxisNetworkDevice:
if self.config_entry.options[CONF_CAMERA]:
platform_tasks.append(
self.hass.config_entries.async_forward_entry_unload(
self.config_entry, "camera"
self.config_entry, CAMERA_DOMAIN
)
)
@ -189,7 +219,7 @@ class AxisNetworkDevice:
self.hass.config_entries.async_forward_entry_unload(
self.config_entry, platform
)
for platform in ["binary_sensor", "switch"]
for platform in [BINARY_SENSOR_DOMAIN, SWITCH_DOMAIN]
]
await asyncio.gather(*platform_tasks)
@ -205,12 +235,7 @@ async def get_device(hass, host, port, username, password):
"""Create a Axis device."""
device = axis.AxisDevice(
loop=hass.loop,
host=host,
port=port,
username=username,
password=password,
web_proto="http",
host=host, port=port, username=username, password=password, web_proto="http",
)
device.vapix.initialize_params(preload_data=False)

2
homeassistant/components/axis/manifest.json
View File

@ -3,7 +3,7 @@
"name": "Axis",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/axis",
"requirements": ["axis==25"],
"requirements": ["axis==26"],
"zeroconf": ["_axis-video._tcp.local."],
"codeowners": ["@Kane610"]
}

11
homeassistant/components/axis/switch.py
View File

@ -1,6 +1,7 @@
"""Support for Axis switches."""
from axis.event_stream import CLASS_OUTPUT
from axis.port_cgi import ACTION_HIGH, ACTION_LOW
from homeassistant.components.switch import SwitchEntity
from homeassistant.core import callback
@ -17,13 +18,13 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
@callback
def async_add_switch(event_id):
"""Add switch from Axis device."""
event = device.api.event.events[event_id]
event = device.api.event[event_id]
if event.CLASS == CLASS_OUTPUT:
async_add_entities([AxisSwitch(event, device)], True)
device.listeners.append(
async_dispatcher_connect(hass, device.event_new_sensor, async_add_switch)
async_dispatcher_connect(hass, device.signal_new_event, async_add_switch)
)
@ -37,16 +38,14 @@ class AxisSwitch(AxisEventBase, SwitchEntity):
async def async_turn_on(self, **kwargs):
"""Turn on switch."""
action = "/"
await self.hass.async_add_executor_job(
self.device.api.vapix.ports[self.event.id].action, action
self.device.api.vapix.ports[self.event.id].action, ACTION_HIGH
)
async def async_turn_off(self, **kwargs):
"""Turn off switch."""
action = "\\"
await self.hass.async_add_executor_job(
self.device.api.vapix.ports[self.event.id].action, action
self.device.api.vapix.ports[self.event.id].action, ACTION_LOW
)
@property

2
requirements_all.txt
View File

@ -303,7 +303,7 @@ avea==1.4
avri-api==0.1.7
# homeassistant.components.axis
axis==25
axis==26
# homeassistant.components.azure_event_hub
azure-eventhub==1.3.1

2
requirements_test_all.txt
View File

@ -141,7 +141,7 @@ async-upnp-client==0.14.13
av==7.0.1
# homeassistant.components.axis
axis==25
axis==26
# homeassistant.components.homekit
base36==0.1.1

16
tests/components/axis/test_binary_sensor.py
View File

@ -1,7 +1,7 @@
"""Axis binary sensor platform tests."""
from homeassistant.components import axis
import homeassistant.components.binary_sensor as binary_sensor
from homeassistant.components.axis.const import DOMAIN as AXIS_DOMAIN
from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR_DOMAIN
from homeassistant.setup import async_setup_component
from .test_device import NAME, setup_axis_integration
@ -28,19 +28,21 @@ async def test_platform_manually_configured(hass):
"""Test that nothing happens when platform is manually configured."""
assert (
await async_setup_component(
hass, binary_sensor.DOMAIN, {"binary_sensor": {"platform": axis.DOMAIN}}
hass,
BINARY_SENSOR_DOMAIN,
{BINARY_SENSOR_DOMAIN: {"platform": AXIS_DOMAIN}},
)
is True
)
assert axis.DOMAIN not in hass.data
assert AXIS_DOMAIN not in hass.data
async def test_no_binary_sensors(hass):
"""Test that no sensors in Axis results in no sensor entities."""
await setup_axis_integration(hass)
assert not hass.states.async_entity_ids("binary_sensor")
assert not hass.states.async_entity_ids(BINARY_SENSOR_DOMAIN)
async def test_binary_sensors(hass):
@ -48,10 +50,10 @@ async def test_binary_sensors(hass):
device = await setup_axis_integration(hass)
for event in EVENTS:
device.api.stream.event.manage_event(event)
device.api.event.process_event(event)
await hass.async_block_till_done()
assert len(hass.states.async_entity_ids("binary_sensor")) == 2
assert len(hass.states.async_entity_ids(BINARY_SENSOR_DOMAIN)) == 2
pir = hass.states.get(f"binary_sensor.{NAME}_pir_0")
assert pir.state == "off"

10
tests/components/axis/test_camera.py
View File

@ -1,7 +1,7 @@
"""Axis camera platform tests."""
from homeassistant.components import axis
import homeassistant.components.camera as camera
from homeassistant.components.axis.const import DOMAIN as AXIS_DOMAIN
from homeassistant.components.camera import DOMAIN as CAMERA_DOMAIN
from homeassistant.setup import async_setup_component
from .test_device import NAME, setup_axis_integration
@ -11,19 +11,19 @@ async def test_platform_manually_configured(hass):
"""Test that nothing happens when platform is manually configured."""
assert (
await async_setup_component(
hass, camera.DOMAIN, {"camera": {"platform": axis.DOMAIN}}
hass, CAMERA_DOMAIN, {"camera": {"platform": AXIS_DOMAIN}}
)
is True
)
assert axis.DOMAIN not in hass.data
assert AXIS_DOMAIN not in hass.data
async def test_camera(hass):
"""Test that Axis camera platform is loaded properly."""
await setup_axis_integration(hass)
assert len(hass.states.async_entity_ids("camera")) == 1
assert len(hass.states.async_entity_ids(CAMERA_DOMAIN)) == 1
cam = hass.states.get(f"camera.{NAME}")
assert cam.state == "idle"

193
tests/components/axis/test_config_flow.py
View File

@ -1,6 +1,15 @@
"""Test Axis config flow."""
from homeassistant.components import axis
from homeassistant.components.axis import config_flow
from homeassistant.components.axis.const import CONF_MODEL, DOMAIN as AXIS_DOMAIN
from homeassistant.const import (
CONF_HOST,
CONF_MAC,
CONF_NAME,
CONF_PASSWORD,
CONF_PORT,
CONF_USERNAME,
)
from .test_device import MAC, MODEL, NAME, setup_axis_integration
@ -11,9 +20,8 @@ from tests.common import MockConfigEntry
def setup_mock_axis_device(mock_device):
"""Prepare mock axis device."""
def mock_constructor(loop, host, username, password, port, web_proto):
def mock_constructor(host, username, password, port, web_proto):
"""Fake the controller constructor."""
mock_device.loop = loop
mock_device.host = host
mock_device.username = username
mock_device.password = password
@ -30,7 +38,7 @@ def setup_mock_axis_device(mock_device):
async def test_flow_manual_configuration(hass):
"""Test that config flow works."""
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN, context={"source": "user"}
AXIS_DOMAIN, context={"source": "user"}
)
assert result["type"] == "form"
@ -43,23 +51,23 @@ async def test_flow_manual_configuration(hass):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
config_flow.CONF_HOST: "1.2.3.4",
config_flow.CONF_USERNAME: "user",
config_flow.CONF_PASSWORD: "pass",
config_flow.CONF_PORT: 80,
CONF_HOST: "1.2.3.4",
CONF_USERNAME: "user",
CONF_PASSWORD: "pass",
CONF_PORT: 80,
},
)
assert result["type"] == "create_entry"
assert result["title"] == f"prodnbr - {MAC}"
assert result["data"] == {
config_flow.CONF_HOST: "1.2.3.4",
config_flow.CONF_USERNAME: "user",
config_flow.CONF_PASSWORD: "pass",
config_flow.CONF_PORT: 80,
config_flow.CONF_MAC: MAC,
config_flow.CONF_MODEL: "prodnbr",
config_flow.CONF_NAME: "prodnbr 0",
CONF_HOST: "1.2.3.4",
CONF_USERNAME: "user",
CONF_PASSWORD: "pass",
CONF_PORT: 80,
CONF_MAC: MAC,
CONF_MODEL: "prodnbr",
CONF_NAME: "prodnbr 0",
}
@ -68,7 +76,7 @@ async def test_manual_configuration_update_configuration(hass):
device = await setup_axis_integration(hass)
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN, context={"source": "user"}
AXIS_DOMAIN, context={"source": "user"}
)
assert result["type"] == "form"
@ -84,16 +92,16 @@ async def test_manual_configuration_update_configuration(hass):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
config_flow.CONF_HOST: "2.3.4.5",
config_flow.CONF_USERNAME: "user",
config_flow.CONF_PASSWORD: "pass",
config_flow.CONF_PORT: 80,
CONF_HOST: "2.3.4.5",
CONF_USERNAME: "user",
CONF_PASSWORD: "pass",
CONF_PORT: 80,
},
)
assert result["type"] == "abort"
assert result["reason"] == "already_configured"
assert device.config_entry.data[config_flow.CONF_HOST] == "2.3.4.5"
assert device.host == "2.3.4.5"
async def test_flow_fails_already_configured(hass):
@ -101,7 +109,7 @@ async def test_flow_fails_already_configured(hass):
await setup_axis_integration(hass)
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN, context={"source": "user"}
AXIS_DOMAIN, context={"source": "user"}
)
assert result["type"] == "form"
@ -117,10 +125,10 @@ async def test_flow_fails_already_configured(hass):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
config_flow.CONF_HOST: "1.2.3.4",
config_flow.CONF_USERNAME: "user",
config_flow.CONF_PASSWORD: "pass",
config_flow.CONF_PORT: 80,
CONF_HOST: "1.2.3.4",
CONF_USERNAME: "user",
CONF_PASSWORD: "pass",
CONF_PORT: 80,
},
)
@ -131,7 +139,7 @@ async def test_flow_fails_already_configured(hass):
async def test_flow_fails_faulty_credentials(hass):
"""Test that config flow fails on faulty credentials."""
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN, context={"source": "user"}
AXIS_DOMAIN, context={"source": "user"}
)
assert result["type"] == "form"
@ -144,10 +152,10 @@ async def test_flow_fails_faulty_credentials(hass):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
config_flow.CONF_HOST: "1.2.3.4",
config_flow.CONF_USERNAME: "user",
config_flow.CONF_PASSWORD: "pass",
config_flow.CONF_PORT: 80,
CONF_HOST: "1.2.3.4",
CONF_USERNAME: "user",
CONF_PASSWORD: "pass",
CONF_PORT: 80,
},
)
@ -157,7 +165,7 @@ async def test_flow_fails_faulty_credentials(hass):
async def test_flow_fails_device_unavailable(hass):
"""Test that config flow fails on device unavailable."""
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN, context={"source": "user"}
AXIS_DOMAIN, context={"source": "user"}
)
assert result["type"] == "form"
@ -170,10 +178,10 @@ async def test_flow_fails_device_unavailable(hass):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
config_flow.CONF_HOST: "1.2.3.4",
config_flow.CONF_USERNAME: "user",
config_flow.CONF_PASSWORD: "pass",
config_flow.CONF_PORT: 80,
CONF_HOST: "1.2.3.4",
CONF_USERNAME: "user",
CONF_PASSWORD: "pass",
CONF_PORT: 80,
},
)
@ -183,18 +191,16 @@ async def test_flow_fails_device_unavailable(hass):
async def test_flow_create_entry_multiple_existing_entries_of_same_model(hass):
"""Test that create entry can generate a name with other entries."""
entry = MockConfigEntry(
domain=axis.DOMAIN,
data={config_flow.CONF_NAME: "prodnbr 0", config_flow.CONF_MODEL: "prodnbr"},
domain=AXIS_DOMAIN, data={CONF_NAME: "prodnbr 0", CONF_MODEL: "prodnbr"},
)
entry.add_to_hass(hass)
entry2 = MockConfigEntry(
domain=axis.DOMAIN,
data={config_flow.CONF_NAME: "prodnbr 1", config_flow.CONF_MODEL: "prodnbr"},
domain=AXIS_DOMAIN, data={CONF_NAME: "prodnbr 1", CONF_MODEL: "prodnbr"},
)
entry2.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN, context={"source": "user"}
AXIS_DOMAIN, context={"source": "user"}
)
assert result["type"] == "form"
@ -207,36 +213,36 @@ async def test_flow_create_entry_multiple_existing_entries_of_same_model(hass):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
config_flow.CONF_HOST: "1.2.3.4",
config_flow.CONF_USERNAME: "user",
config_flow.CONF_PASSWORD: "pass",
config_flow.CONF_PORT: 80,
CONF_HOST: "1.2.3.4",
CONF_USERNAME: "user",
CONF_PASSWORD: "pass",
CONF_PORT: 80,
},
)
assert result["type"] == "create_entry"
assert result["title"] == f"prodnbr - {MAC}"
assert result["data"] == {
config_flow.CONF_HOST: "1.2.3.4",
config_flow.CONF_USERNAME: "user",
config_flow.CONF_PASSWORD: "pass",
config_flow.CONF_PORT: 80,
config_flow.CONF_MAC: MAC,
config_flow.CONF_MODEL: "prodnbr",
config_flow.CONF_NAME: "prodnbr 2",
CONF_HOST: "1.2.3.4",
CONF_USERNAME: "user",
CONF_PASSWORD: "pass",
CONF_PORT: 80,
CONF_MAC: MAC,
CONF_MODEL: "prodnbr",
CONF_NAME: "prodnbr 2",
}
assert result["data"][config_flow.CONF_NAME] == "prodnbr 2"
assert result["data"][CONF_NAME] == "prodnbr 2"
async def test_zeroconf_flow(hass):
"""Test that zeroconf discovery for new devices work."""
with patch.object(axis, "get_device", return_value=Mock()):
with patch.object(axis.device, "get_device", return_value=Mock()):
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN,
AXIS_DOMAIN,
data={
config_flow.CONF_HOST: "1.2.3.4",
config_flow.CONF_PORT: 80,
CONF_HOST: "1.2.3.4",
CONF_PORT: 80,
"hostname": "name",
"properties": {"macaddress": MAC},
},
@ -253,26 +259,26 @@ async def test_zeroconf_flow(hass):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
config_flow.CONF_HOST: "1.2.3.4",
config_flow.CONF_USERNAME: "user",
config_flow.CONF_PASSWORD: "pass",
config_flow.CONF_PORT: 80,
CONF_HOST: "1.2.3.4",
CONF_USERNAME: "user",
CONF_PASSWORD: "pass",
CONF_PORT: 80,
},
)
assert result["type"] == "create_entry"
assert result["title"] == f"prodnbr - {MAC}"
assert result["data"] == {
config_flow.CONF_HOST: "1.2.3.4",
config_flow.CONF_USERNAME: "user",
config_flow.CONF_PASSWORD: "pass",
config_flow.CONF_PORT: 80,
config_flow.CONF_MAC: MAC,
config_flow.CONF_MODEL: "prodnbr",
config_flow.CONF_NAME: "prodnbr 0",
CONF_HOST: "1.2.3.4",
CONF_USERNAME: "user",
CONF_PASSWORD: "pass",
CONF_PORT: 80,
CONF_MAC: MAC,
CONF_MODEL: "prodnbr",
CONF_NAME: "prodnbr 0",
}
assert result["data"][config_flow.CONF_NAME] == "prodnbr 0"
assert result["data"][CONF_NAME] == "prodnbr 0"
async def test_zeroconf_flow_already_configured(hass):
@ -281,10 +287,10 @@ async def test_zeroconf_flow_already_configured(hass):
assert device.host == "1.2.3.4"
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN,
AXIS_DOMAIN,
data={
config_flow.CONF_HOST: "1.2.3.4",
config_flow.CONF_PORT: 80,
CONF_HOST: "1.2.3.4",
CONF_PORT: 80,
"hostname": "name",
"properties": {"macaddress": MAC},
},
@ -301,20 +307,20 @@ async def test_zeroconf_flow_updated_configuration(hass):
device = await setup_axis_integration(hass)
assert device.host == "1.2.3.4"
assert device.config_entry.data == {
config_flow.CONF_HOST: "1.2.3.4",
config_flow.CONF_PORT: 80,
config_flow.CONF_USERNAME: "username",
config_flow.CONF_PASSWORD: "password",
config_flow.CONF_MAC: MAC,
config_flow.CONF_MODEL: MODEL,
config_flow.CONF_NAME: NAME,
CONF_HOST: "1.2.3.4",
CONF_PORT: 80,
CONF_USERNAME: "username",
CONF_PASSWORD: "password",
CONF_MAC: MAC,
CONF_MODEL: MODEL,
CONF_NAME: NAME,
}
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN,
AXIS_DOMAIN,
data={
config_flow.CONF_HOST: "2.3.4.5",
config_flow.CONF_PORT: 8080,
CONF_HOST: "2.3.4.5",
CONF_PORT: 8080,
"hostname": "name",
"properties": {"macaddress": MAC},
},
@ -324,24 +330,21 @@ async def test_zeroconf_flow_updated_configuration(hass):
assert result["type"] == "abort"
assert result["reason"] == "already_configured"
assert device.config_entry.data == {
config_flow.CONF_HOST: "2.3.4.5",
config_flow.CONF_PORT: 8080,
config_flow.CONF_USERNAME: "username",
config_flow.CONF_PASSWORD: "password",
config_flow.CONF_MAC: MAC,
config_flow.CONF_MODEL: MODEL,
config_flow.CONF_NAME: NAME,
CONF_HOST: "2.3.4.5",
CONF_PORT: 8080,
CONF_USERNAME: "username",
CONF_PASSWORD: "password",
CONF_MAC: MAC,
CONF_MODEL: MODEL,
CONF_NAME: NAME,
}
async def test_zeroconf_flow_ignore_non_axis_device(hass):
"""Test that zeroconf doesn't setup devices with link local addresses."""
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN,
data={
config_flow.CONF_HOST: "169.254.3.4",
"properties": {"macaddress": "01234567890"},
},
AXIS_DOMAIN,
data={CONF_HOST: "169.254.3.4", "properties": {"macaddress": "01234567890"}},
context={"source": "zeroconf"},
)
@ -352,8 +355,8 @@ async def test_zeroconf_flow_ignore_non_axis_device(hass):
async def test_zeroconf_flow_ignore_link_local_address(hass):
"""Test that zeroconf doesn't setup devices with link local addresses."""
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN,
data={config_flow.CONF_HOST: "169.254.3.4", "properties": {"macaddress": MAC}},
AXIS_DOMAIN,
data={CONF_HOST: "169.254.3.4", "properties": {"macaddress": MAC}},
context={"source": "zeroconf"},
)

51
tests/components/axis/test_device.py
View File

@ -2,10 +2,25 @@
from copy import deepcopy
import axis as axislib
from axis.event_stream import OPERATION_INITIALIZED
import pytest
from homeassistant import config_entries
from homeassistant.components import axis
from homeassistant.components.axis.const import (
CONF_CAMERA,
CONF_EVENTS,
CONF_MODEL,
DOMAIN as AXIS_DOMAIN,
)
from homeassistant.const import (
CONF_HOST,
CONF_MAC,
CONF_NAME,
CONF_PASSWORD,
CONF_PORT,
CONF_USERNAME,
)
from tests.async_mock import Mock, patch
from tests.common import MockConfigEntry
@ -14,16 +29,16 @@ MAC = "00408C12345"
MODEL = "model"
NAME = "name"
ENTRY_OPTIONS = {axis.CONF_CAMERA: True, axis.CONF_EVENTS: True}
ENTRY_OPTIONS = {CONF_CAMERA: True, CONF_EVENTS: True}
ENTRY_CONFIG = {
axis.CONF_HOST: "1.2.3.4",
axis.CONF_USERNAME: "username",
axis.CONF_PASSWORD: "password",
axis.CONF_PORT: 80,
axis.CONF_MAC: MAC,
axis.device.CONF_MODEL: MODEL,
axis.device.CONF_NAME: NAME,
CONF_HOST: "1.2.3.4",
CONF_USERNAME: "username",
CONF_PASSWORD: "password",
CONF_PORT: 80,
CONF_MAC: MAC,
CONF_MODEL: MODEL,
CONF_NAME: NAME,
}
DEFAULT_BRAND = """root.Brand.Brand=AXIS
@ -67,7 +82,7 @@ async def setup_axis_integration(
):
"""Create the Axis device."""
config_entry = MockConfigEntry(
domain=axis.DOMAIN,
domain=AXIS_DOMAIN,
data=deepcopy(config),
connection_class=config_entries.CONN_CLASS_LOCAL_PUSH,
options=deepcopy(options),
@ -95,7 +110,7 @@ async def setup_axis_integration(
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
return hass.data[axis.DOMAIN].get(config[axis.CONF_MAC])
return hass.data[AXIS_DOMAIN].get(config[CONF_MAC])
async def test_device_setup(hass):
@ -113,10 +128,10 @@ async def test_device_setup(hass):
assert forward_entry_setup.mock_calls[1][1] == (entry, "binary_sensor")
assert forward_entry_setup.mock_calls[2][1] == (entry, "switch")
assert device.host == ENTRY_CONFIG[axis.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.CONF_MAC]
assert device.host == ENTRY_CONFIG[CONF_HOST]
assert device.model == ENTRY_CONFIG[CONF_MODEL]
assert device.name == ENTRY_CONFIG[CONF_NAME]
assert device.serial == ENTRY_CONFIG[CONF_MAC]
async def test_update_address(hass):
@ -125,7 +140,7 @@ async def test_update_address(hass):
assert device.api.config.host == "1.2.3.4"
await hass.config_entries.flow.async_init(
axis.DOMAIN,
AXIS_DOMAIN,
data={
"host": "2.3.4.5",
"port": 80,
@ -157,14 +172,14 @@ async def test_device_not_accessible(hass):
"""Failed setup schedules a retry of setup."""
with patch.object(axis.device, "get_device", side_effect=axis.errors.CannotConnect):
await setup_axis_integration(hass)
assert hass.data[axis.DOMAIN] == {}
assert hass.data[AXIS_DOMAIN] == {}
async def test_device_unknown_error(hass):
"""Unknown errors are handled."""
with patch.object(axis.device, "get_device", side_effect=Exception):
await setup_axis_integration(hass)
assert hass.data[axis.DOMAIN] == {}
assert hass.data[AXIS_DOMAIN] == {}
async def test_new_event_sends_signal(hass):
@ -175,7 +190,7 @@ async def test_new_event_sends_signal(hass):
axis_device = axis.device.AxisNetworkDevice(hass, entry)
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=OPERATION_INITIALIZED, event_id="event")
await hass.async_block_till_done()
assert len(mock_dispatch_send.mock_calls) == 1

79
tests/components/axis/test_init.py
View File

@ -1,5 +1,15 @@
"""Test Axis component setup process."""
from homeassistant.components import axis
from homeassistant.components.axis.const import CONF_MODEL, DOMAIN as AXIS_DOMAIN
from homeassistant.const import (
CONF_DEVICE,
CONF_HOST,
CONF_MAC,
CONF_NAME,
CONF_PASSWORD,
CONF_PORT,
CONF_USERNAME,
)
from homeassistant.setup import async_setup_component
from .test_device import MAC, setup_axis_integration
@ -10,21 +20,21 @@ from tests.common import MockConfigEntry
async def test_setup_no_config(hass):
"""Test setup without configuration."""
assert await async_setup_component(hass, axis.DOMAIN, {})
assert axis.DOMAIN not in hass.data
assert await async_setup_component(hass, AXIS_DOMAIN, {})
assert AXIS_DOMAIN not in hass.data
async def test_setup_entry(hass):
"""Test successful setup of entry."""
await setup_axis_integration(hass)
assert len(hass.data[axis.DOMAIN]) == 1
assert MAC in hass.data[axis.DOMAIN]
assert len(hass.data[AXIS_DOMAIN]) == 1
assert MAC in hass.data[AXIS_DOMAIN]
async def test_setup_entry_fails(hass):
"""Test successful setup of entry."""
config_entry = MockConfigEntry(
domain=axis.DOMAIN, data={axis.CONF_MAC: "0123"}, version=2
domain=AXIS_DOMAIN, data={CONF_MAC: "0123"}, version=2
)
config_entry.add_to_hass(hass)
@ -36,43 +46,32 @@ async def test_setup_entry_fails(hass):
assert not await hass.config_entries.async_setup(config_entry.entry_id)
assert not hass.data[axis.DOMAIN]
assert not hass.data[AXIS_DOMAIN]
async def test_unload_entry(hass):
"""Test successful unload of entry."""
device = await setup_axis_integration(hass)
assert hass.data[axis.DOMAIN]
assert hass.data[AXIS_DOMAIN]
assert await hass.config_entries.async_unload(device.config_entry.entry_id)
assert not hass.data[axis.DOMAIN]
async def test_populate_options(hass):
"""Test successful populate options."""
device = await setup_axis_integration(hass, options=None)
assert device.config_entry.options == {
axis.CONF_CAMERA: True,
axis.CONF_EVENTS: True,
axis.CONF_TRIGGER_TIME: axis.DEFAULT_TRIGGER_TIME,
}
assert not hass.data[AXIS_DOMAIN]
async def test_migrate_entry(hass):
"""Test successful migration of entry data."""
legacy_config = {
axis.CONF_DEVICE: {
axis.CONF_HOST: "1.2.3.4",
axis.CONF_USERNAME: "username",
axis.CONF_PASSWORD: "password",
axis.CONF_PORT: 80,
CONF_DEVICE: {
CONF_HOST: "1.2.3.4",
CONF_USERNAME: "username",
CONF_PASSWORD: "password",
CONF_PORT: 80,
},
axis.CONF_MAC: "mac",
axis.device.CONF_MODEL: "model",
axis.device.CONF_NAME: "name",
CONF_MAC: "mac",
CONF_MODEL: "model",
CONF_NAME: "name",
}
entry = MockConfigEntry(domain=axis.DOMAIN, data=legacy_config)
entry = MockConfigEntry(domain=AXIS_DOMAIN, data=legacy_config)
assert entry.data == legacy_config
assert entry.version == 1
@ -80,18 +79,18 @@ async def test_migrate_entry(hass):
await entry.async_migrate(hass)
assert entry.data == {
axis.CONF_DEVICE: {
axis.CONF_HOST: "1.2.3.4",
axis.CONF_USERNAME: "username",
axis.CONF_PASSWORD: "password",
axis.CONF_PORT: 80,
CONF_DEVICE: {
CONF_HOST: "1.2.3.4",
CONF_USERNAME: "username",
CONF_PASSWORD: "password",
CONF_PORT: 80,
},
axis.CONF_HOST: "1.2.3.4",
axis.CONF_USERNAME: "username",
axis.CONF_PASSWORD: "password",
axis.CONF_PORT: 80,
axis.CONF_MAC: "mac",
axis.device.CONF_MODEL: "model",
axis.device.CONF_NAME: "name",
CONF_HOST: "1.2.3.4",
CONF_USERNAME: "username",
CONF_PASSWORD: "password",
CONF_PORT: 80,
CONF_MAC: "mac",
CONF_MODEL: "model",
CONF_NAME: "name",
}
assert entry.version == 2

30
tests/components/axis/test_switch.py
View File

@ -1,7 +1,9 @@
"""Axis switch platform tests."""
from homeassistant.components import axis
import homeassistant.components.switch as switch
from axis.port_cgi import ACTION_HIGH, ACTION_LOW
from homeassistant.components.axis.const import DOMAIN as AXIS_DOMAIN
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.setup import async_setup_component
from .test_device import NAME, setup_axis_integration
@ -31,17 +33,17 @@ EVENTS = [
async def test_platform_manually_configured(hass):
"""Test that nothing happens when platform is manually configured."""
assert await async_setup_component(
hass, switch.DOMAIN, {"switch": {"platform": axis.DOMAIN}}
hass, SWITCH_DOMAIN, {SWITCH_DOMAIN: {"platform": AXIS_DOMAIN}}
)
assert axis.DOMAIN not in hass.data
assert AXIS_DOMAIN not in hass.data
async def test_no_switches(hass):
"""Test that no output events in Axis results in no switch entities."""
await setup_axis_integration(hass)
assert not hass.states.async_entity_ids("switch")
assert not hass.states.async_entity_ids(SWITCH_DOMAIN)
async def test_switches(hass):
@ -53,10 +55,10 @@ async def test_switches(hass):
device.api.vapix.ports["1"].name = ""
for event in EVENTS:
device.api.stream.event.manage_event(event)
device.api.event.process_event(event)
await hass.async_block_till_done()
assert len(hass.states.async_entity_ids("switch")) == 2
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 2
relay_0 = hass.states.get(f"switch.{NAME}_doorbell")
assert relay_0.state == "off"
@ -69,14 +71,20 @@ async def test_switches(hass):
device.api.vapix.ports["0"].action = Mock()
await hass.services.async_call(
"switch", "turn_on", {"entity_id": f"switch.{NAME}_doorbell"}, blocking=True
SWITCH_DOMAIN,
"turn_on",
{"entity_id": f"switch.{NAME}_doorbell"},
blocking=True,
)
await hass.services.async_call(
"switch", "turn_off", {"entity_id": f"switch.{NAME}_doorbell"}, blocking=True
SWITCH_DOMAIN,
"turn_off",
{"entity_id": f"switch.{NAME}_doorbell"},
blocking=True,
)
assert device.api.vapix.ports["0"].action.call_args_list == [
mock_call("/"),
mock_call("\\"),
mock_call(ACTION_HIGH),
mock_call(ACTION_LOW),
]