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

Axis config flow (#18543) 

* Initial draft

* Add tests for init
Fix hound comments

* Add tests for device
Change parameter handling to make device easier to test

* Remove superfluous functionality per Martins request

* Fix hound comments

* Embedded platforms

* Fix device import

* Config flow retry

* Options default values will be set automatically to options in config entry before component can be used

* Clean up init
Add populate options
Fix small issues in config flow
Add tests covering init

* Improve device tests

* Add config flow tests

* Fix hound comments

* Rebase miss

* Initial tests for binary sensors

* Clean up
More binary sensor tests

* Hound comments

* Add camera tests

* Fix initial state of sensors

* Bump dependency to v17

* Fix pylint and flake8

* Fix comments
pull/22353/head
Robert Svensson 2019-03-24 16:16:50 +01:00 committed by GitHub
parent 9214934d47
commit 6988fe783c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
22 changed files with 1288 additions and 324 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.coveragerc
View File

@ -36,7 +36,6 @@ omit =
homeassistant/components/arlo/*
homeassistant/components/asterisk_mbox/*
homeassistant/components/august/*
homeassistant/components/axis/*
homeassistant/components/bbb_gpio/*
homeassistant/components/arest/binary_sensor.py
homeassistant/components/concord232/binary_sensor.py

26
homeassistant/components/axis/.translations/en.json Normal file
View File

@ -0,0 +1,26 @@
{
"config": {
"title": "Axis device",
"step": {
"user": {
"title": "Set up Axis device",
"data": {
"host": "Host",
"username": "Username",
"password": "Password",
"port": "Port"
}
}
},
"error": {
"already_configured": "Device is already configured",
"device_unavailable": "Device is not available",
"faulty_credentials": "Bad user credentials"
},
"abort": {
"already_configured": "Device is already configured",
"bad_config_file": "Bad data from config file",
"link_local_address": "Link local addresses are not supported"
}
}
}

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

@ -1,262 +1,76 @@
"""Support for Axis devices."""
import logging
import voluptuous as vol
from homeassistant.components.discovery import SERVICE_AXIS
from homeassistant import config_entries
from homeassistant.const import (
ATTR_LOCATION, CONF_EVENT, CONF_HOST, CONF_INCLUDE,
CONF_NAME, CONF_PASSWORD, CONF_PORT, CONF_TRIGGER_TIME, CONF_USERNAME,
CONF_DEVICE, CONF_HOST, CONF_NAME, CONF_TRIGGER_TIME,
EVENT_HOMEASSISTANT_STOP)
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers import discovery
from homeassistant.helpers.dispatcher import dispatcher_send
from homeassistant.util.json import load_json, save_json
REQUIREMENTS = ['axis==16']
from .config_flow import configured_devices, DEVICE_SCHEMA
from .const import CONF_CAMERA, CONF_EVENTS, DEFAULT_TRIGGER_TIME, DOMAIN
from .device import AxisNetworkDevice, get_device
_LOGGER = logging.getLogger(__name__)
DOMAIN = 'axis'
CONFIG_FILE = 'axis.conf'
EVENT_TYPES = ['motion', 'vmd3', 'pir', 'sound',
'daynight', 'tampering', 'input']
PLATFORMS = ['camera']
AXIS_INCLUDE = EVENT_TYPES + PLATFORMS
AXIS_DEFAULT_HOST = '192.168.0.90'
AXIS_DEFAULT_USERNAME = 'root'
AXIS_DEFAULT_PASSWORD = 'pass'
DEFAULT_PORT = 80
DEVICE_SCHEMA = vol.Schema({
vol.Required(CONF_INCLUDE):
vol.All(cv.ensure_list, [vol.In(AXIS_INCLUDE)]),
vol.Optional(CONF_NAME): cv.string,
vol.Optional(CONF_HOST, default=AXIS_DEFAULT_HOST): cv.string,
vol.Optional(CONF_USERNAME, default=AXIS_DEFAULT_USERNAME): cv.string,
vol.Optional(CONF_PASSWORD, default=AXIS_DEFAULT_PASSWORD): cv.string,
vol.Optional(CONF_TRIGGER_TIME, default=0): cv.positive_int,
vol.Optional(CONF_PORT, default=DEFAULT_PORT): cv.port,
vol.Optional(ATTR_LOCATION, default=''): cv.string,
})
REQUIREMENTS = ['axis==17']
CONFIG_SCHEMA = vol.Schema({
DOMAIN: cv.schema_with_slug_keys(DEVICE_SCHEMA),
}, extra=vol.ALLOW_EXTRA)
SERVICE_VAPIX_CALL = 'vapix_call'
SERVICE_VAPIX_CALL_RESPONSE = 'vapix_call_response'
SERVICE_CGI = 'cgi'
SERVICE_ACTION = 'action'
SERVICE_PARAM = 'param'
SERVICE_DEFAULT_CGI = 'param.cgi'
SERVICE_DEFAULT_ACTION = 'update'
SERVICE_SCHEMA = vol.Schema({
vol.Required(CONF_NAME): cv.string,
vol.Required(SERVICE_PARAM): cv.string,
vol.Optional(SERVICE_CGI, default=SERVICE_DEFAULT_CGI): cv.string,
vol.Optional(SERVICE_ACTION, default=SERVICE_DEFAULT_ACTION): cv.string,
})
def request_configuration(hass, config, name, host, serialnumber):
"""Request configuration steps from the user."""
configurator = hass.components.configurator
def configuration_callback(callback_data):
"""Call when configuration is submitted."""
if CONF_INCLUDE not in callback_data:
configurator.notify_errors(
request_id, "Functionality mandatory.")
return False
callback_data[CONF_INCLUDE] = callback_data[CONF_INCLUDE].split()
callback_data[CONF_HOST] = host
if CONF_NAME not in callback_data:
callback_data[CONF_NAME] = name
try:
device_config = DEVICE_SCHEMA(callback_data)
except vol.Invalid:
configurator.notify_errors(
request_id, "Bad input, please check spelling.")
return False
if setup_device(hass, config, device_config):
config_file = load_json(hass.config.path(CONFIG_FILE))
config_file[serialnumber] = dict(device_config)
save_json(hass.config.path(CONFIG_FILE), config_file)
configurator.request_done(request_id)
else:
configurator.notify_errors(
request_id, "Failed to register, please try again.")
return False
title = '{} ({})'.format(name, host)
request_id = configurator.request_config(
title, configuration_callback,
description='Functionality: ' + str(AXIS_INCLUDE),
entity_picture="/static/images/logo_axis.png",
link_name='Axis platform documentation',
link_url='https://home-assistant.io/components/axis/',
submit_caption="Confirm",
fields=[
{'id': CONF_NAME,
'name': "Device name",
'type': 'text'},
{'id': CONF_USERNAME,
'name': "User name",
'type': 'text'},
{'id': CONF_PASSWORD,
'name': 'Password',
'type': 'password'},
{'id': CONF_INCLUDE,
'name': "Device functionality (space separated list)",
'type': 'text'},
{'id': ATTR_LOCATION,
'name': "Physical location of device (optional)",
'type': 'text'},
{'id': CONF_PORT,
'name': "HTTP port (default=80)",
'type': 'number'},
{'id': CONF_TRIGGER_TIME,
'name': "Sensor update interval (optional)",
'type': 'number'},
]
)
def setup(hass, config):
async def async_setup(hass, config):
"""Set up for Axis devices."""
hass.data[DOMAIN] = {}
def _shutdown(call):
"""Stop the event stream on shutdown."""
for serialnumber, device in hass.data[DOMAIN].items():
_LOGGER.info("Stopping event stream for %s.", serialnumber)
device.stop()
hass.bus.listen_once(EVENT_HOMEASSISTANT_STOP, _shutdown)
def axis_device_discovered(service, discovery_info):
"""Call when axis devices has been found."""
host = discovery_info[CONF_HOST]
name = discovery_info['hostname']
serialnumber = discovery_info['properties']['macaddress']
if serialnumber not in hass.data[DOMAIN]:
config_file = load_json(hass.config.path(CONFIG_FILE))
if serialnumber in config_file:
# Device config previously saved to file
try:
device_config = DEVICE_SCHEMA(config_file[serialnumber])
device_config[CONF_HOST] = host
except vol.Invalid as err:
_LOGGER.error("Bad data from %s. %s", CONFIG_FILE, err)
return False
if not setup_device(hass, config, device_config):
_LOGGER.error(
"Couldn't set up %s", device_config[CONF_NAME])
else:
# New device, create configuration request for UI
request_configuration(hass, config, name, host, serialnumber)
else:
# Device already registered, but on a different IP
device = hass.data[DOMAIN][serialnumber]
device.config.host = host
dispatcher_send(hass, DOMAIN + '_' + device.name + '_new_ip', host)
# Register discovery service
discovery.listen(hass, SERVICE_AXIS, axis_device_discovered)
if DOMAIN in config:
for device in config[DOMAIN]:
device_config = config[DOMAIN][device]
for device_name, device_config in config[DOMAIN].items():
if CONF_NAME not in device_config:
device_config[CONF_NAME] = device
if not setup_device(hass, config, device_config):
_LOGGER.error("Couldn't set up %s", device_config[CONF_NAME])
device_config[CONF_NAME] = device_name
def vapix_service(call):
"""Service to send a message."""
for device in hass.data[DOMAIN].values():
if device.name == call.data[CONF_NAME]:
response = device.vapix.do_request(
call.data[SERVICE_CGI],
call.data[SERVICE_ACTION],
call.data[SERVICE_PARAM])
hass.bus.fire(SERVICE_VAPIX_CALL_RESPONSE, response)
return True
_LOGGER.info("Couldn't find device %s", call.data[CONF_NAME])
return False
if device_config[CONF_HOST] not in configured_devices(hass):
hass.async_create_task(hass.config_entries.flow.async_init(
DOMAIN, context={'source': config_entries.SOURCE_IMPORT},
data=device_config
))
# Register service with Home Assistant.
hass.services.register(
DOMAIN, SERVICE_VAPIX_CALL, vapix_service, schema=SERVICE_SCHEMA)
return True
def setup_device(hass, config, device_config):
"""Set up an Axis device."""
import axis
async def async_setup_entry(hass, config_entry):
"""Set up the Axis component."""
if DOMAIN not in hass.data:
hass.data[DOMAIN] = {}
def signal_callback(action, event):
"""Call to configure events when initialized on event stream."""
if action == 'add':
event_config = {
CONF_EVENT: event,
CONF_NAME: device_config[CONF_NAME],
ATTR_LOCATION: device_config[ATTR_LOCATION],
CONF_TRIGGER_TIME: device_config[CONF_TRIGGER_TIME]
}
component = event.event_platform
discovery.load_platform(
hass, component, DOMAIN, event_config, config)
if not config_entry.options:
await async_populate_options(hass, config_entry)
event_types = [
event
for event in device_config[CONF_INCLUDE]
if event in EVENT_TYPES
]
device = AxisNetworkDevice(hass, config_entry)
device = axis.AxisDevice(
loop=hass.loop, host=device_config[CONF_HOST],
username=device_config[CONF_USERNAME],
password=device_config[CONF_PASSWORD],
port=device_config[CONF_PORT], web_proto='http',
event_types=event_types, signal=signal_callback)
try:
hass.data[DOMAIN][device.vapix.serial_number] = device
except axis.Unauthorized:
_LOGGER.error("Credentials for %s are faulty",
device_config[CONF_HOST])
if not await device.async_setup():
return False
except axis.RequestError:
return False
hass.data[DOMAIN][device.serial] = device
device.name = device_config[CONF_NAME]
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, device.shutdown)
for component in device_config[CONF_INCLUDE]:
if component == 'camera':
camera_config = {
CONF_NAME: device_config[CONF_NAME],
CONF_HOST: device_config[CONF_HOST],
CONF_PORT: device_config[CONF_PORT],
CONF_USERNAME: device_config[CONF_USERNAME],
CONF_PASSWORD: device_config[CONF_PASSWORD]
}
discovery.load_platform(
hass, component, DOMAIN, camera_config, config)
if event_types:
hass.add_job(device.start)
return True
async def async_populate_options(hass, config_entry):
"""Populate default options for device."""
from axis.vapix import VAPIX_IMAGE_FORMAT
device = await get_device(hass, config_entry.data[CONF_DEVICE])
supported_formats = device.vapix.get_param(VAPIX_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)

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

@ -1,86 +1,87 @@
"""Support for Axis binary sensors."""
from datetime import timedelta
import logging
from homeassistant.components.binary_sensor import BinarySensorDevice
from homeassistant.const import (
ATTR_LOCATION, CONF_EVENT, CONF_NAME, CONF_TRIGGER_TIME)
from homeassistant.const import CONF_MAC, 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
from homeassistant.util.dt import utcnow
DEPENDENCIES = ['axis']
from .const import DOMAIN as AXIS_DOMAIN, LOGGER
_LOGGER = logging.getLogger(__name__)
DEPENDENCIES = [AXIS_DOMAIN]
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the Axis binary devices."""
add_entities([AxisBinarySensor(discovery_info)], True)
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up a Axis binary sensor."""
serial_number = config_entry.data[CONF_MAC]
device = hass.data[AXIS_DOMAIN][serial_number]
@callback
def async_add_sensor(event):
"""Add binary sensor from Axis device."""
async_add_entities([AxisBinarySensor(event, device)], True)
device.listeners.append(
async_dispatcher_connect(hass, 'axis_add_sensor', async_add_sensor))
class AxisBinarySensor(BinarySensorDevice):
"""Representation of a binary Axis event."""
def __init__(self, event_config):
def __init__(self, event, device):
"""Initialize the Axis binary sensor."""
self.axis_event = event_config[CONF_EVENT]
self.device_name = event_config[CONF_NAME]
self.location = event_config[ATTR_LOCATION]
self.delay = event_config[CONF_TRIGGER_TIME]
self.event = event
self.device = device
self.delay = device.config_entry.options[CONF_TRIGGER_TIME]
self.remove_timer = None
async def async_added_to_hass(self):
"""Subscribe sensors events."""
self.axis_event.callback = self._update_callback
self.event.register_callback(self.update_callback)
def _update_callback(self):
def update_callback(self):
"""Update the sensor's state, if needed."""
delay = self.device.config_entry.options[CONF_TRIGGER_TIME]
if self.remove_timer is not None:
self.remove_timer()
self.remove_timer = None
if self.delay == 0 or self.is_on:
if delay == 0 or self.is_on:
self.schedule_update_ha_state()
else: # Run timer to delay updating the state
@callback
def _delay_update(now):
"""Timer callback for sensor update."""
_LOGGER.debug("%s called delayed (%s sec) update",
self.name, self.delay)
self.async_schedule_update_ha_state()
self.remove_timer = None
return
self.remove_timer = async_track_point_in_utc_time(
self.hass, _delay_update,
utcnow() + timedelta(seconds=self.delay))
@callback
def _delay_update(now):
"""Timer callback for sensor update."""
LOGGER.debug("%s called delayed (%s sec) update", self.name, delay)
self.async_schedule_update_ha_state()
self.remove_timer = None
self.remove_timer = async_track_point_in_utc_time(
self.hass, _delay_update,
utcnow() + timedelta(seconds=delay))
@property
def is_on(self):
"""Return true if event is active."""
return self.axis_event.is_tripped
return self.event.is_tripped
@property
def name(self):
"""Return the name of the event."""
return '{}_{}_{}'.format(
self.device_name, self.axis_event.event_type, self.axis_event.id)
return '{} {} {}'.format(
self.device.name, self.event.event_type, self.event.id)
@property
def device_class(self):
"""Return the class of the event."""
return self.axis_event.event_class
return self.event.event_class
@property
def should_poll(self):
"""No polling needed."""
return False
@property
def device_state_attributes(self):
"""Return the state attributes of the event."""
attr = {}
attr[ATTR_LOCATION] = self.location
return attr

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

@ -1,58 +1,59 @@
"""Support for Axis camera streaming."""
import logging
from homeassistant.components.mjpeg.camera import (
CONF_MJPEG_URL, CONF_STILL_IMAGE_URL, MjpegCamera, filter_urllib3_logging)
from homeassistant.const import (
CONF_AUTHENTICATION, CONF_HOST, CONF_NAME, CONF_PASSWORD, CONF_PORT,
CONF_USERNAME, HTTP_DIGEST_AUTHENTICATION)
from homeassistant.helpers.dispatcher import dispatcher_connect
CONF_AUTHENTICATION, CONF_DEVICE, CONF_HOST, CONF_MAC, CONF_NAME,
CONF_PASSWORD, CONF_PORT, CONF_USERNAME, HTTP_DIGEST_AUTHENTICATION)
from homeassistant.helpers.dispatcher import async_dispatcher_connect
_LOGGER = logging.getLogger(__name__)
from .const import DOMAIN as AXIS_DOMAIN
DOMAIN = 'axis'
DEPENDENCIES = [DOMAIN]
DEPENDENCIES = [AXIS_DOMAIN]
AXIS_IMAGE = 'http://{}:{}/axis-cgi/jpg/image.cgi'
AXIS_VIDEO = 'http://{}:{}/axis-cgi/mjpg/video.cgi'
def _get_image_url(host, port, mode):
"""Set the URL to get the image."""
if mode == 'mjpeg':
return 'http://{}:{}/axis-cgi/mjpg/video.cgi'.format(host, port)
if mode == 'single':
return 'http://{}:{}/axis-cgi/jpg/image.cgi'.format(host, port)
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the Axis camera."""
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up the Axis camera video stream."""
filter_urllib3_logging()
camera_config = {
CONF_NAME: discovery_info[CONF_NAME],
CONF_USERNAME: discovery_info[CONF_USERNAME],
CONF_PASSWORD: discovery_info[CONF_PASSWORD],
CONF_MJPEG_URL: _get_image_url(
discovery_info[CONF_HOST], str(discovery_info[CONF_PORT]),
'mjpeg'),
CONF_STILL_IMAGE_URL: _get_image_url(
discovery_info[CONF_HOST], str(discovery_info[CONF_PORT]),
'single'),
serial_number = config_entry.data[CONF_MAC]
device = hass.data[AXIS_DOMAIN][serial_number]
config = {
CONF_NAME: config_entry.data[CONF_NAME],
CONF_USERNAME: config_entry.data[CONF_DEVICE][CONF_USERNAME],
CONF_PASSWORD: config_entry.data[CONF_DEVICE][CONF_PASSWORD],
CONF_MJPEG_URL: AXIS_VIDEO.format(
config_entry.data[CONF_DEVICE][CONF_HOST],
config_entry.data[CONF_DEVICE][CONF_PORT]),
CONF_STILL_IMAGE_URL: AXIS_IMAGE.format(
config_entry.data[CONF_DEVICE][CONF_HOST],
config_entry.data[CONF_DEVICE][CONF_PORT]),
CONF_AUTHENTICATION: HTTP_DIGEST_AUTHENTICATION,
}
add_entities([AxisCamera(
hass, camera_config, str(discovery_info[CONF_PORT]))])
async_add_entities([AxisCamera(config, device)])
class AxisCamera(MjpegCamera):
"""Representation of a Axis camera."""
def __init__(self, hass, config, port):
def __init__(self, config, device):
"""Initialize Axis Communications camera component."""
super().__init__(config)
self.port = port
dispatcher_connect(
hass, DOMAIN + '_' + config[CONF_NAME] + '_new_ip', self._new_ip)
self.device_config = config
self.device = device
self.port = device.config_entry.data[CONF_DEVICE][CONF_PORT]
self.unsub_dispatcher = None
async def async_added_to_hass(self):
"""Subscribe camera events."""
self.unsub_dispatcher = async_dispatcher_connect(
self.hass, 'axis_{}_new_ip'.format(self.device.name), self._new_ip)
def _new_ip(self, host):
"""Set new IP for video stream."""
self._mjpeg_url = _get_image_url(host, self.port, 'mjpeg')
self._still_image_url = _get_image_url(host, self.port, 'single')
self._mjpeg_url = AXIS_VIDEO.format(host, self.port)
self._still_image_url = AXIS_IMAGE.format(host, self.port)

202
homeassistant/components/axis/config_flow.py Normal file
View File

@ -0,0 +1,202 @@
"""Config flow to configure Axis devices."""
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import (
CONF_DEVICE, CONF_HOST, CONF_MAC, CONF_NAME, CONF_PASSWORD, CONF_PORT,
CONF_USERNAME)
from homeassistant.core import callback
from homeassistant.helpers import config_validation as cv
from homeassistant.util.json import load_json
from .const import CONF_MODEL, DOMAIN
from .device import get_device
from .errors import AlreadyConfigured, AuthenticationRequired, CannotConnect
CONFIG_FILE = 'axis.conf'
EVENT_TYPES = ['motion', 'vmd3', 'pir', 'sound',
'daynight', 'tampering', 'input']
PLATFORMS = ['camera']
AXIS_INCLUDE = EVENT_TYPES + PLATFORMS
AXIS_DEFAULT_HOST = '192.168.0.90'
AXIS_DEFAULT_USERNAME = 'root'
AXIS_DEFAULT_PASSWORD = 'pass'
DEFAULT_PORT = 80
DEVICE_SCHEMA = vol.Schema({
vol.Optional(CONF_NAME): cv.string,
vol.Optional(CONF_HOST, default=AXIS_DEFAULT_HOST): cv.string,
vol.Optional(CONF_USERNAME, default=AXIS_DEFAULT_USERNAME): cv.string,
vol.Optional(CONF_PASSWORD, default=AXIS_DEFAULT_PASSWORD): cv.string,
vol.Optional(CONF_PORT, default=DEFAULT_PORT): cv.port,
}, extra=vol.ALLOW_EXTRA)
@callback
def configured_devices(hass):
"""Return a set of the configured devices."""
return set(entry.data[CONF_DEVICE][CONF_HOST] for entry
in hass.config_entries.async_entries(DOMAIN))
@config_entries.HANDLERS.register(DOMAIN)
class AxisFlowHandler(config_entries.ConfigFlow):
"""Handle a Axis config flow."""
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_PUSH
def __init__(self):
"""Initialize the Axis config flow."""
self.device_config = {}
self.model = None
self.name = None
self.serial_number = None
self.discovery_schema = {}
self.import_schema = {}
async def async_step_user(self, user_input=None):
"""Handle a Axis config flow start.
Manage device specific parameters.
"""
from axis.vapix import VAPIX_MODEL_ID, VAPIX_SERIAL_NUMBER
errors = {}
if user_input is not None:
try:
if user_input[CONF_HOST] in configured_devices(self.hass):
raise AlreadyConfigured
self.device_config = {
CONF_HOST: user_input[CONF_HOST],
CONF_PORT: user_input[CONF_PORT],
CONF_USERNAME: user_input[CONF_USERNAME],
CONF_PASSWORD: user_input[CONF_PASSWORD]
}
device = await get_device(self.hass, self.device_config)
self.serial_number = device.vapix.get_param(
VAPIX_SERIAL_NUMBER)
self.model = device.vapix.get_param(VAPIX_MODEL_ID)
return await self._create_entry()
except AlreadyConfigured:
errors['base'] = 'already_configured'
except AuthenticationRequired:
errors['base'] = 'faulty_credentials'
except CannotConnect:
errors['base'] = 'device_unavailable'
data = self.import_schema or self.discovery_schema or {
vol.Required(CONF_HOST): str,
vol.Required(CONF_USERNAME): str,
vol.Required(CONF_PASSWORD): str,
vol.Required(CONF_PORT, default=DEFAULT_PORT): int
}
return self.async_show_form(
step_id='user',
description_placeholders=self.device_config,
data_schema=vol.Schema(data),
errors=errors
)
async def _create_entry(self):
"""Create entry for device.
Generate a name to be used as a prefix for device entities.
"""
if self.name is None:
same_model = [
entry.data[CONF_NAME] for entry
in self.hass.config_entries.async_entries(DOMAIN)
if entry.data[CONF_MODEL] == self.model
]
self.name = "{}".format(self.model)
for idx in range(len(same_model) + 1):
self.name = "{} {}".format(self.model, idx)
if self.name not in same_model:
break
data = {
CONF_DEVICE: self.device_config,
CONF_NAME: self.name,
CONF_MAC: self.serial_number,
CONF_MODEL: self.model,
}
title = "{} - {}".format(self.model, self.serial_number)
return self.async_create_entry(
title=title,
data=data
)
async def async_step_discovery(self, discovery_info):
"""Prepare configuration for a discovered Axis device.
This flow is triggered by the discovery component.
"""
if discovery_info[CONF_HOST] in configured_devices(self.hass):
return self.async_abort(reason='already_configured')
if discovery_info[CONF_HOST].startswith('169.254'):
return self.async_abort(reason='link_local_address')
config_file = await self.hass.async_add_executor_job(
load_json, self.hass.config.path(CONFIG_FILE))
serialnumber = discovery_info['properties']['macaddress']
if serialnumber not in config_file:
self.discovery_schema = {
vol.Required(
CONF_HOST, default=discovery_info[CONF_HOST]): str,
vol.Required(CONF_USERNAME): str,
vol.Required(CONF_PASSWORD): str,
vol.Required(CONF_PORT, default=discovery_info[CONF_PORT]): int
}
return await self.async_step_user()
try:
device_config = DEVICE_SCHEMA(config_file[serialnumber])
device_config[CONF_HOST] = discovery_info[CONF_HOST]
if CONF_NAME not in device_config:
device_config[CONF_NAME] = discovery_info['hostname']
except vol.Invalid:
return self.async_abort(reason='bad_config_file')
return await self.async_step_import(device_config)
async def async_step_import(self, import_config):
"""Import a Axis device as a config entry.
This flow is triggered by `async_setup` for configured devices.
This flow is also triggered by `async_step_discovery`.
This will execute for any Axis device that contains a complete
configuration.
"""
self.name = import_config[CONF_NAME]
self.import_schema = {
vol.Required(CONF_HOST, default=import_config[CONF_HOST]): str,
vol.Required(
CONF_USERNAME, default=import_config[CONF_USERNAME]): str,
vol.Required(
CONF_PASSWORD, default=import_config[CONF_PASSWORD]): str,
vol.Required(CONF_PORT, default=import_config[CONF_PORT]): int
}
return await self.async_step_user(user_input=import_config)

12
homeassistant/components/axis/const.py Normal file
View File

@ -0,0 +1,12 @@
"""Constants for the Axis component."""
import logging
LOGGER = logging.getLogger('homeassistant.components.axis')
DOMAIN = 'axis'
CONF_CAMERA = 'camera'
CONF_EVENTS = 'events'
CONF_MODEL = 'model'
DEFAULT_TRIGGER_TIME = 0

127
homeassistant/components/axis/device.py Normal file
View File

@ -0,0 +1,127 @@
"""Axis network device abstraction."""
import asyncio
import async_timeout
from homeassistant.const import (
CONF_DEVICE, CONF_HOST, CONF_MAC, CONF_NAME, CONF_PASSWORD, CONF_PORT,
CONF_USERNAME)
from homeassistant.core import callback
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers.dispatcher import async_dispatcher_send
from .const import CONF_CAMERA, CONF_EVENTS, CONF_MODEL, LOGGER
from .errors import AuthenticationRequired, CannotConnect
class AxisNetworkDevice:
"""Manages a Axis device."""
def __init__(self, hass, config_entry):
"""Initialize the device."""
self.hass = hass
self.config_entry = config_entry
self.available = True
self.api = None
self.fw_version = None
self.product_type = None
self.listeners = []
@property
def host(self):
"""Return the host of this device."""
return self.config_entry.data[CONF_DEVICE][CONF_HOST]
@property
def model(self):
"""Return the model of this device."""
return self.config_entry.data[CONF_MODEL]
@property
def name(self):
"""Return the name of this device."""
return self.config_entry.data[CONF_NAME]
@property
def serial(self):
"""Return the mac of this device."""
return self.config_entry.data[CONF_MAC]
async def async_setup(self):
"""Set up the device."""
from axis.vapix import VAPIX_FW_VERSION, VAPIX_PROD_TYPE
hass = self.hass
try:
self.api = await get_device(
hass, self.config_entry.data[CONF_DEVICE],
event_types='on', signal_callback=self.async_signal_callback)
except CannotConnect:
raise ConfigEntryNotReady
except Exception: # pylint: disable=broad-except
LOGGER.error(
'Unknown error connecting with Axis device on %s', self.host)
return False
self.fw_version = self.api.vapix.get_param(VAPIX_FW_VERSION)
self.product_type = self.api.vapix.get_param(VAPIX_PROD_TYPE)
if self.config_entry.options[CONF_CAMERA]:
self.hass.async_create_task(
self.hass.config_entries.async_forward_entry_setup(
self.config_entry, 'camera'))
if self.config_entry.options[CONF_EVENTS]:
self.hass.async_create_task(
self.hass.config_entries.async_forward_entry_setup(
self.config_entry, 'binary_sensor'))
self.api.start()
return True
@callback
def async_signal_callback(self, action, event):
"""Call to configure events when initialized on event stream."""
if action == 'add':
async_dispatcher_send(self.hass, 'axis_add_sensor', event)
@callback
def shutdown(self, event):
"""Stop the event stream."""
self.api.stop()
async def get_device(hass, config, event_types=None, signal_callback=None):
"""Create a Axis device."""
import axis
device = axis.AxisDevice(
loop=hass.loop, host=config[CONF_HOST],
username=config[CONF_USERNAME],
password=config[CONF_PASSWORD],
port=config[CONF_PORT], web_proto='http',
event_types=event_types, signal=signal_callback)
try:
with async_timeout.timeout(15):
await hass.async_add_executor_job(device.vapix.load_params)
return device
except axis.Unauthorized:
LOGGER.warning("Connected to device at %s but not registered.",
config[CONF_HOST])
raise AuthenticationRequired
except (asyncio.TimeoutError, axis.RequestError):
LOGGER.error("Error connecting to the Axis device at %s",
config[CONF_HOST])
raise CannotConnect
except axis.AxisException:
LOGGER.exception('Unknown Axis communication error occurred')
raise AuthenticationRequired

22
homeassistant/components/axis/errors.py Normal file
View File

@ -0,0 +1,22 @@
"""Errors for the Axis component."""
from homeassistant.exceptions import HomeAssistantError
class AxisException(HomeAssistantError):
"""Base class for Axis exceptions."""
class AlreadyConfigured(AxisException):
"""Device is already configured."""
class AuthenticationRequired(AxisException):
"""Unknown error occurred."""
class CannotConnect(AxisException):
"""Unable to connect to the device."""
class UserLevel(AxisException):
"""User level too low."""

15
homeassistant/components/axis/services.yaml
View File

@ -1,15 +0,0 @@
vapix_call:
description: Configure device using Vapix parameter management.
fields:
name:
description: Name of device to Configure. [Required]
example: M1065-W
cgi:
description: Which cgi to call on device. [Optional] Default is 'param.cgi'
example: 'applications/control.cgi'
action:
description: What type of call. [Optional] Default is 'update'
example: 'start'
param:
description: What parameter to operate on. [Required]
example: 'package=VideoMotionDetection'

26
homeassistant/components/axis/strings.json Normal file
View File

@ -0,0 +1,26 @@
{
"config": {
"title": "Axis device",
"step": {
"user": {
"title": "Set up Axis device",
"data": {
"host": "Host",
"username": "Username",
"password": "Password",
"port": "Port"
}
}
},
"error": {
"already_configured": "Device is already configured",
"device_unavailable": "Device is not available",
"faulty_credentials": "Bad user credentials"
},
"abort": {
"already_configured": "Device is already configured",
"bad_config_file": "Bad data from config file",
"link_local_address": "Link local addresses are not supported"
}
}
}

2
homeassistant/components/discovery/__init__.py
View File

@ -52,6 +52,7 @@ SERVICE_WINK = 'wink'
SERVICE_XIAOMI_GW = 'xiaomi_gw'
CONFIG_ENTRY_HANDLERS = {
SERVICE_AXIS: 'axis',
SERVICE_DAIKIN: 'daikin',
SERVICE_DECONZ: 'deconz',
'esphome': 'esphome',
@ -69,7 +70,6 @@ SERVICE_HANDLERS = {
SERVICE_NETGEAR: ('device_tracker', None),
SERVICE_WEMO: ('wemo', None),
SERVICE_HASSIO: ('hassio', None),
SERVICE_AXIS: ('axis', None),
SERVICE_APPLE_TV: ('apple_tv', None),
SERVICE_ENIGMA2: ('media_player', 'enigma2'),
SERVICE_ROKU: ('roku', None),

1
homeassistant/config_entries.py
View File

@ -143,6 +143,7 @@ HANDLERS = Registry()
# Components that have config flows. In future we will auto-generate this list.
FLOWS = [
'ambient_station',
'axis',
'cast',
'daikin',
'deconz',

2
requirements_all.txt
View File

@ -186,7 +186,7 @@ av==6.1.2
# avion==0.10
# homeassistant.components.axis
axis==16
axis==17
# homeassistant.components.tts.baidu
baidu-aip==1.6.6

3
requirements_test_all.txt
View File

@ -56,6 +56,9 @@ apns2==0.3.0
# homeassistant.components.stream
av==6.1.2
# homeassistant.components.axis
axis==17
# homeassistant.components.zha
bellows-homeassistant==0.7.1

1
script/gen_requirements_all.py
View File

@ -46,6 +46,7 @@ TEST_REQUIREMENTS = (
'aiounifi',
'apns2',
'av',
'axis',
'caldav',
'coinmarketcap',
'defusedxml',

1
tests/components/axis/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Tests for the Axis component."""

102
tests/components/axis/test_binary_sensor.py Normal file
View File

@ -0,0 +1,102 @@
"""Axis binary sensor platform tests."""
from unittest.mock import Mock
from homeassistant import config_entries
from homeassistant.components import axis
from homeassistant.setup import async_setup_component
import homeassistant.components.binary_sensor as binary_sensor
EVENTS = [
{
'operation': 'Initialized',
'topic': 'tns1:Device/tnsaxis:Sensor/PIR',
'source': 'sensor',
'source_idx': '0',
'type': 'state',
'value': '0'
},
{
'operation': 'Initialized',
'topic': 'tnsaxis:CameraApplicationPlatform/VMD/Camera1Profile1',
'type': 'active',
'value': '1'
}
]
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, options=ENTRY_OPTIONS)
device = axis.AxisNetworkDevice(hass, config_entry)
device.api = AxisDevice(loop=loop, **config_entry.data[axis.CONF_DEVICE],
signal=device.async_signal_callback)
hass.data[axis.DOMAIN] = {device.serial: device}
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):
"""Test that nothing happens when platform is manually configured."""
assert await async_setup_component(hass, binary_sensor.DOMAIN, {
'binary_sensor': {
'platform': axis.DOMAIN
}
}) is True
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_device(hass)
assert len(hass.states.async_all()) == 0
async def test_binary_sensors(hass):
"""Test that sensors are loaded properly."""
device = await setup_device(hass)
for event in EVENTS:
device.api.stream.event.manage_event(event)
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 2
pir = hass.states.get('binary_sensor.model_0_pir_0')
assert pir.state == 'off'
assert pir.name == 'model 0 PIR 0'
vmd4 = hass.states.get('binary_sensor.model_0_vmd4_camera1profile1')
assert vmd4.state == 'on'
assert vmd4.name == 'model 0 VMD4 Camera1Profile1'

73
tests/components/axis/test_camera.py Normal file
View File

@ -0,0 +1,73 @@
"""Axis camera platform tests."""
from unittest.mock import Mock
from homeassistant import config_entries
from homeassistant.components import axis
from homeassistant.setup import async_setup_component
import homeassistant.components.camera as camera
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, options=ENTRY_OPTIONS)
device = axis.AxisNetworkDevice(hass, config_entry)
device.api = AxisDevice(loop=loop, **config_entry.data[axis.CONF_DEVICE],
signal=device.async_signal_callback)
hass.data[axis.DOMAIN] = {device.serial: device}
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):
"""Test that nothing happens when platform is manually configured."""
assert await async_setup_component(hass, camera.DOMAIN, {
'camera': {
'platform': axis.DOMAIN
}
}) is True
assert axis.DOMAIN not in hass.data
async def test_camera(hass):
"""Test that Axis camera platform is loaded properly."""
await setup_device(hass)
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 1
cam = hass.states.get('camera.model_0')
assert cam.state == 'idle'
assert cam.name == 'model 0'

319
tests/components/axis/test_config_flow.py Normal file
View File

@ -0,0 +1,319 @@
"""Test Axis config flow."""
from unittest.mock import Mock, patch
from homeassistant.components import axis
from homeassistant.components.axis import config_flow
from tests.common import mock_coro, MockConfigEntry
import axis as axis_lib
async def test_configured_devices(hass):
"""Test that configured devices works as expected."""
result = config_flow.configured_devices(hass)
assert not result
entry = MockConfigEntry(domain=axis.DOMAIN,
data={axis.CONF_DEVICE: {axis.CONF_HOST: ''}})
entry.add_to_hass(hass)
result = config_flow.configured_devices(hass)
assert len(result) == 1
async def test_flow_works(hass):
"""Test that config flow works."""
flow = config_flow.AxisFlowHandler()
flow.hass = hass
with patch('axis.AxisDevice') as mock_device:
def mock_constructor(
loop, host, username, password, port, web_proto, event_types,
signal):
"""Fake the controller constructor."""
mock_device.loop = loop
mock_device.host = host
mock_device.username = username
mock_device.password = password
mock_device.port = port
return mock_device
def mock_get_param(param):
"""Fake get param method."""
return param
mock_device.side_effect = mock_constructor
mock_device.vapix.load_params.return_value = Mock()
mock_device.vapix.get_param.side_effect = mock_get_param
result = await flow.async_step_user(user_input={
config_flow.CONF_HOST: '1.2.3.4',
config_flow.CONF_USERNAME: 'user',
config_flow.CONF_PASSWORD: 'pass',
config_flow.CONF_PORT: 81
})
assert result['type'] == 'create_entry'
assert result['title'] == '{} - {}'.format(
axis_lib.vapix.VAPIX_MODEL_ID, axis_lib.vapix.VAPIX_SERIAL_NUMBER)
assert result['data'] == {
axis.CONF_DEVICE: {
config_flow.CONF_HOST: '1.2.3.4',
config_flow.CONF_USERNAME: 'user',
config_flow.CONF_PASSWORD: 'pass',
config_flow.CONF_PORT: 81
},
config_flow.CONF_MAC: axis_lib.vapix.VAPIX_SERIAL_NUMBER,
config_flow.CONF_MODEL: axis_lib.vapix.VAPIX_MODEL_ID,
config_flow.CONF_NAME: 'Brand.ProdNbr 0'
}
async def test_flow_fails_already_configured(hass):
"""Test that config flow fails on already configured device."""
flow = config_flow.AxisFlowHandler()
flow.hass = hass
entry = MockConfigEntry(domain=axis.DOMAIN, data={axis.CONF_DEVICE: {
axis.CONF_HOST: '1.2.3.4'
}})
entry.add_to_hass(hass)
result = await flow.async_step_user(user_input={
config_flow.CONF_HOST: '1.2.3.4',
config_flow.CONF_USERNAME: 'user',
config_flow.CONF_PASSWORD: 'pass',
config_flow.CONF_PORT: 81
})
assert result['errors'] == {'base': 'already_configured'}
async def test_flow_fails_faulty_credentials(hass):
"""Test that config flow fails on faulty credentials."""
flow = config_flow.AxisFlowHandler()
flow.hass = hass
with patch('homeassistant.components.axis.config_flow.get_device',
side_effect=config_flow.AuthenticationRequired):
result = await flow.async_step_user(user_input={
config_flow.CONF_HOST: '1.2.3.4',
config_flow.CONF_USERNAME: 'user',
config_flow.CONF_PASSWORD: 'pass',
config_flow.CONF_PORT: 81
})
assert result['errors'] == {'base': 'faulty_credentials'}
async def test_flow_fails_device_unavailable(hass):
"""Test that config flow fails on device unavailable."""
flow = config_flow.AxisFlowHandler()
flow.hass = hass
with patch('homeassistant.components.axis.config_flow.get_device',
side_effect=config_flow.CannotConnect):
result = await flow.async_step_user(user_input={
config_flow.CONF_HOST: '1.2.3.4',
config_flow.CONF_USERNAME: 'user',
config_flow.CONF_PASSWORD: 'pass',
config_flow.CONF_PORT: 81
})
assert result['errors'] == {'base': 'device_unavailable'}
async def test_flow_create_entry(hass):
"""Test that create entry can generate a name without other entries."""
flow = config_flow.AxisFlowHandler()
flow.hass = hass
flow.model = 'model'
result = await flow._create_entry()
assert result['data'][config_flow.CONF_NAME] == 'model 0'
async def test_flow_create_entry_more_entries(hass):
"""Test that create entry can generate a name with other entries."""
entry = MockConfigEntry(
domain=axis.DOMAIN, data={config_flow.CONF_NAME: 'model 0',
config_flow.CONF_MODEL: 'model'})
entry.add_to_hass(hass)
entry2 = MockConfigEntry(
domain=axis.DOMAIN, data={config_flow.CONF_NAME: 'model 1',
config_flow.CONF_MODEL: 'model'})
entry2.add_to_hass(hass)
flow = config_flow.AxisFlowHandler()
flow.hass = hass
flow.model = 'model'
result = await flow._create_entry()
assert result['data'][config_flow.CONF_NAME] == 'model 2'
async def test_discovery_flow(hass):
"""Test that discovery for new devices work."""
flow = config_flow.AxisFlowHandler()
flow.hass = hass
with patch.object(axis, 'get_device', return_value=mock_coro(Mock())):
result = await flow.async_step_discovery(discovery_info={
config_flow.CONF_HOST: '1.2.3.4',
config_flow.CONF_PORT: 80,
'properties': {'macaddress': '1234'}
})
assert result['type'] == 'form'
assert result['step_id'] == 'user'
async def test_discovery_flow_known_device(hass):
"""Test that discovery for known devices work.
This is legacy support from devices registered with configurator.
"""
flow = config_flow.AxisFlowHandler()
flow.hass = hass
with patch('homeassistant.components.axis.config_flow.load_json',
return_value={'1234ABCD': {
config_flow.CONF_HOST: '2.3.4.5',
config_flow.CONF_USERNAME: 'user',
config_flow.CONF_PASSWORD: 'pass',
config_flow.CONF_PORT: 80}}), \
patch('axis.AxisDevice') as mock_device:
def mock_constructor(
loop, host, username, password, port, web_proto, event_types,
signal):
"""Fake the controller constructor."""
mock_device.loop = loop
mock_device.host = host
mock_device.username = username
mock_device.password = password
mock_device.port = port
return mock_device
def mock_get_param(param):
"""Fake get param method."""
return param
mock_device.side_effect = mock_constructor
mock_device.vapix.load_params.return_value = Mock()
mock_device.vapix.get_param.side_effect = mock_get_param
result = await flow.async_step_discovery(discovery_info={
config_flow.CONF_HOST: '1.2.3.4',
config_flow.CONF_PORT: 80,
'hostname': 'name',
'properties': {'macaddress': '1234ABCD'}
})
assert result['type'] == 'create_entry'
async def test_discovery_flow_already_configured(hass):
"""Test that discovery doesn't setup already configured devices."""
flow = config_flow.AxisFlowHandler()
flow.hass = hass
entry = MockConfigEntry(domain=axis.DOMAIN, data={axis.CONF_DEVICE: {
axis.CONF_HOST: '1.2.3.4'
}})
entry.add_to_hass(hass)
result = await flow.async_step_discovery(discovery_info={
config_flow.CONF_HOST: '1.2.3.4',
config_flow.CONF_USERNAME: 'user',
config_flow.CONF_PASSWORD: 'pass',
config_flow.CONF_PORT: 81
})
print(result)
assert result['type'] == 'abort'
async def test_discovery_flow_link_local_address(hass):
"""Test that discovery doesn't setup devices with link local addresses."""
flow = config_flow.AxisFlowHandler()
flow.hass = hass
result = await flow.async_step_discovery(discovery_info={
config_flow.CONF_HOST: '169.254.3.4'
})
assert result['type'] == 'abort'
async def test_discovery_flow_bad_config_file(hass):
"""Test that discovery with bad config files abort."""
flow = config_flow.AxisFlowHandler()
flow.hass = hass
with patch('homeassistant.components.axis.config_flow.load_json',
return_value={'1234ABCD': {
config_flow.CONF_HOST: '2.3.4.5',
config_flow.CONF_USERNAME: 'user',
config_flow.CONF_PASSWORD: 'pass',
config_flow.CONF_PORT: 80}}), \
patch('homeassistant.components.axis.config_flow.DEVICE_SCHEMA',
side_effect=config_flow.vol.Invalid('')):
result = await flow.async_step_discovery(discovery_info={
config_flow.CONF_HOST: '1.2.3.4',
'properties': {'macaddress': '1234ABCD'}
})
assert result['type'] == 'abort'
async def test_import_flow_works(hass):
"""Test that import flow works."""
flow = config_flow.AxisFlowHandler()
flow.hass = hass
with patch('axis.AxisDevice') as mock_device:
def mock_constructor(
loop, host, username, password, port, web_proto, event_types,
signal):
"""Fake the controller constructor."""
mock_device.loop = loop
mock_device.host = host
mock_device.username = username
mock_device.password = password
mock_device.port = port
return mock_device
def mock_get_param(param):
"""Fake get param method."""
return param
mock_device.side_effect = mock_constructor
mock_device.vapix.load_params.return_value = Mock()
mock_device.vapix.get_param.side_effect = mock_get_param
result = await flow.async_step_import(import_config={
config_flow.CONF_HOST: '1.2.3.4',
config_flow.CONF_USERNAME: 'user',
config_flow.CONF_PASSWORD: 'pass',
config_flow.CONF_PORT: 81,
config_flow.CONF_NAME: 'name'
})
assert result['type'] == 'create_entry'
assert result['title'] == '{} - {}'.format(
axis_lib.vapix.VAPIX_MODEL_ID, axis_lib.vapix.VAPIX_SERIAL_NUMBER)
assert result['data'] == {
axis.CONF_DEVICE: {
config_flow.CONF_HOST: '1.2.3.4',
config_flow.CONF_USERNAME: 'user',
config_flow.CONF_PASSWORD: 'pass',
config_flow.CONF_PORT: 81
},
config_flow.CONF_MAC: axis_lib.vapix.VAPIX_SERIAL_NUMBER,
config_flow.CONF_MODEL: axis_lib.vapix.VAPIX_MODEL_ID,
config_flow.CONF_NAME: 'name'
}

152
tests/components/axis/test_device.py Normal file
View File

@ -0,0 +1,152 @@
"""Test Axis device."""
from unittest.mock import Mock, patch
import pytest
from tests.common import mock_coro
from homeassistant.components.axis import device, errors
DEVICE_DATA = {
device.CONF_HOST: '1.2.3.4',
device.CONF_USERNAME: 'username',
device.CONF_PASSWORD: 'password',
device.CONF_PORT: 1234
}
ENTRY_OPTIONS = {
device.CONF_CAMERA: True,
device.CONF_EVENTS: ['pir'],
}
ENTRY_CONFIG = {
device.CONF_DEVICE: DEVICE_DATA,
device.CONF_MAC: 'mac',
device.CONF_MODEL: 'model',
device.CONF_NAME: 'name'
}
async def test_device_setup():
"""Successful setup."""
hass = Mock()
entry = Mock()
entry.data = ENTRY_CONFIG
entry.options = ENTRY_OPTIONS
api = Mock()
axis_device = device.AxisNetworkDevice(hass, entry)
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)):
assert await axis_device.async_setup() is True
assert axis_device.api is api
assert len(hass.config_entries.async_forward_entry_setup.mock_calls) == 2
assert hass.config_entries.async_forward_entry_setup.mock_calls[0][1] == \
(entry, 'camera')
assert hass.config_entries.async_forward_entry_setup.mock_calls[1][1] == \
(entry, 'binary_sensor')
async def test_device_not_accessible():
"""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(device, 'get_device',
side_effect=errors.CannotConnect), \
pytest.raises(device.ConfigEntryNotReady):
await axis_device.async_setup()
assert not hass.helpers.event.async_call_later.mock_calls
async def test_device_unknown_error():
"""Unknown errors are handled."""
hass = Mock()
entry = Mock()
entry.data = ENTRY_CONFIG
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):
"""Make sure that new event send signal."""
entry = Mock()
entry.data = ENTRY_CONFIG
axis_device = device.AxisNetworkDevice(hass, entry)
with patch.object(device, 'async_dispatcher_send') as mock_dispatch_send:
axis_device.async_signal_callback(action='add', event='event')
await hass.async_block_till_done()
assert len(mock_dispatch_send.mock_calls) == 1
assert len(mock_dispatch_send.mock_calls[0]) == 3
async def test_shutdown():
"""Successful shutdown."""
hass = Mock()
entry = Mock()
entry.data = ENTRY_CONFIG
axis_device = device.AxisNetworkDevice(hass, entry)
axis_device.api = Mock()
axis_device.shutdown(None)
assert len(axis_device.api.stop.mock_calls) == 1
async def test_get_device(hass):
"""Successful call."""
with patch('axis.vapix.Vapix.load_params',
return_value=mock_coro()):
assert await device.get_device(hass, DEVICE_DATA)
async def test_get_device_fails(hass):
"""Device unauthorized yields authentication required error."""
import axis
with patch('axis.vapix.Vapix.load_params',
side_effect=axis.Unauthorized), \
pytest.raises(errors.AuthenticationRequired):
await device.get_device(hass, DEVICE_DATA)
async def test_get_device_device_unavailable(hass):
"""Device unavailable yields cannot connect error."""
import axis
with patch('axis.vapix.Vapix.load_params',
side_effect=axis.RequestError), \
pytest.raises(errors.CannotConnect):
await device.get_device(hass, DEVICE_DATA)
async def test_get_device_unknown_error(hass):
"""Device yield unknown error."""
import axis
with patch('axis.vapix.Vapix.load_params',
side_effect=axis.AxisException), \
pytest.raises(errors.AuthenticationRequired):
await device.get_device(hass, DEVICE_DATA)

97
tests/components/axis/test_init.py Normal file
View File

@ -0,0 +1,97 @@
"""Test Axis component setup process."""
from unittest.mock import Mock, patch
from homeassistant.setup import async_setup_component
from homeassistant.components import axis
from tests.common import mock_coro, MockConfigEntry
async def test_setup(hass):
"""Test configured options for a device are loaded via config entry."""
with patch.object(hass, 'config_entries') as mock_config_entries, \
patch.object(axis, 'configured_devices', return_value={}):
assert await async_setup_component(hass, axis.DOMAIN, {
axis.DOMAIN: {
'device_name': {
axis.CONF_HOST: '1.2.3.4',
axis.config_flow.CONF_PORT: 80,
}
}
})
assert len(mock_config_entries.flow.mock_calls) == 1
async def test_setup_device_already_configured(hass):
"""Test already configured device does not configure a second."""
with patch.object(hass, 'config_entries') as mock_config_entries, \
patch.object(axis, 'configured_devices', return_value={'1.2.3.4'}):
assert await async_setup_component(hass, axis.DOMAIN, {
axis.DOMAIN: {
'device_name': {
axis.CONF_HOST: '1.2.3.4'
}
}
})
assert not mock_config_entries.flow.mock_calls
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
async def test_setup_entry(hass):
"""Test successful setup of entry."""
entry = MockConfigEntry(
domain=axis.DOMAIN, data={axis.device.CONF_MAC: '0123'})
mock_device = Mock()
mock_device.async_setup.return_value = mock_coro(True)
mock_device.serial.return_value = '1'
with patch.object(axis, 'AxisNetworkDevice') as mock_device_class, \
patch.object(
axis, 'async_populate_options', return_value=mock_coro(True)):
mock_device_class.return_value = mock_device
assert await axis.async_setup_entry(hass, entry)
assert len(hass.data[axis.DOMAIN]) == 1
async def test_setup_entry_fails(hass):
"""Test successful setup of entry."""
entry = MockConfigEntry(
domain=axis.DOMAIN, data={axis.device.CONF_MAC: '0123'}, options=True)
mock_device = Mock()
mock_device.async_setup.return_value = mock_coro(False)
with patch.object(axis, 'AxisNetworkDevice') as mock_device_class:
mock_device_class.return_value = mock_device
assert not await axis.async_setup_entry(hass, entry)
assert not hass.data[axis.DOMAIN]
async def test_populate_options(hass):
"""Test successful populate options."""
entry = MockConfigEntry(domain=axis.DOMAIN, data={'device': {}})
entry.add_to_hass(hass)
with patch.object(axis, 'get_device', return_value=mock_coro(Mock())):
await axis.async_populate_options(hass, entry)
assert entry.options == {
axis.CONF_CAMERA: True,
axis.CONF_EVENTS: True,
axis.CONF_TRIGGER_TIME: axis.DEFAULT_TRIGGER_TIME
}