Add Dyson Pure Cool Link support (#7795)

* Add Dyson Pure Cool Link support

* Code review

* Improve auto/night mode

* Move night_mode to Dyson fan component

* Code review

* fix asynchrone/sync

* Create dyson.py
pull/8028/head
Charles Blonde 2017-06-14 13:56:03 +02:00 committed by Pascal Vizeli
parent bf2fe60cb5
commit 8c0967a190
10 changed files with 924 additions and 1 deletions

View File

@ -0,0 +1,98 @@
"""Parent component for Dyson Pure Cool Link devices."""
import logging
import voluptuous as vol
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers import discovery
from homeassistant.const import CONF_USERNAME, CONF_PASSWORD, CONF_TIMEOUT, \
CONF_DEVICES
REQUIREMENTS = ['libpurecoollink==0.1.5']
_LOGGER = logging.getLogger(__name__)
CONF_LANGUAGE = "language"
CONF_RETRY = "retry"
DEFAULT_TIMEOUT = 5
DEFAULT_RETRY = 10
DOMAIN = "dyson"
CONFIG_SCHEMA = vol.Schema({
DOMAIN: vol.Schema({
vol.Required(CONF_USERNAME): cv.string,
vol.Required(CONF_PASSWORD): cv.string,
vol.Required(CONF_LANGUAGE): cv.string,
vol.Optional(CONF_TIMEOUT, default=DEFAULT_TIMEOUT): cv.positive_int,
vol.Optional(CONF_RETRY, default=DEFAULT_RETRY): cv.positive_int,
vol.Optional(CONF_DEVICES, default=[]):
vol.All(cv.ensure_list, [dict]),
})
}, extra=vol.ALLOW_EXTRA)
DYSON_DEVICES = "dyson_devices"
def setup(hass, config):
"""Set up the Dyson parent component."""
_LOGGER.info("Creating new Dyson component")
if DYSON_DEVICES not in hass.data:
hass.data[DYSON_DEVICES] = []
from libpurecoollink.dyson import DysonAccount
dyson_account = DysonAccount(config[DOMAIN].get(CONF_USERNAME),
config[DOMAIN].get(CONF_PASSWORD),
config[DOMAIN].get(CONF_LANGUAGE))
logged = dyson_account.login()
timeout = config[DOMAIN].get(CONF_TIMEOUT)
retry = config[DOMAIN].get(CONF_RETRY)
if not logged:
_LOGGER.error("Not connected to Dyson account. Unable to add devices")
return False
_LOGGER.info("Connected to Dyson account")
dyson_devices = dyson_account.devices()
if CONF_DEVICES in config[DOMAIN] and config[DOMAIN].get(CONF_DEVICES):
configured_devices = config[DOMAIN].get(CONF_DEVICES)
for device in configured_devices:
dyson_device = next((d for d in dyson_devices if
d.serial == device["device_id"]), None)
if dyson_device:
connected = dyson_device.connect(None, device["device_ip"],
timeout, retry)
if connected:
_LOGGER.info("Connected to device %s", dyson_device)
hass.data[DYSON_DEVICES].append(dyson_device)
else:
_LOGGER.warning("Unable to connect to device %s",
dyson_device)
else:
_LOGGER.warning(
"Unable to find device %s in Dyson account",
device["device_id"])
else:
# Not yet reliable
for device in dyson_devices:
_LOGGER.info("Trying to connect to device %s with timeout=%i "
"and retry=%i", device, timeout, retry)
connected = device.connect(None, None, timeout, retry)
if connected:
_LOGGER.info("Connected to device %s", device)
hass.data[DYSON_DEVICES].append(device)
else:
_LOGGER.warning("Unable to connect to device %s", device)
# Start fan/sensors components
if hass.data[DYSON_DEVICES]:
_LOGGER.debug("Starting sensor/fan components")
discovery.load_platform(hass, "sensor", DOMAIN, {}, config)
discovery.load_platform(hass, "fan", DOMAIN, {}, config)
return True

View File

@ -0,0 +1,218 @@
"""Support for Dyson Pure Cool link fan."""
import logging
import asyncio
from os import path
import voluptuous as vol
import homeassistant.helpers.config_validation as cv
from homeassistant.components.fan import (FanEntity, SUPPORT_OSCILLATE,
SUPPORT_SET_SPEED,
DOMAIN)
from homeassistant.helpers.entity import ToggleEntity
from homeassistant.components.dyson import DYSON_DEVICES
from homeassistant.config import load_yaml_config_file
DEPENDENCIES = ['dyson']
_LOGGER = logging.getLogger(__name__)
DYSON_FAN_DEVICES = "dyson_fan_devices"
SERVICE_SET_NIGHT_MODE = 'dyson_set_night_mode'
DYSON_SET_NIGHT_MODE_SCHEMA = vol.Schema({
vol.Required('entity_id'): cv.entity_id,
vol.Required('night_mode'): cv.boolean
})
def setup_platform(hass, config, add_devices, discovery_info=None):
"""Setup the Dyson fan components."""
_LOGGER.info("Creating new Dyson fans")
if DYSON_FAN_DEVICES not in hass.data:
hass.data[DYSON_FAN_DEVICES] = []
# Get Dyson Devices from parent component
for device in hass.data[DYSON_DEVICES]:
dyson_entity = DysonPureCoolLinkDevice(hass, device)
hass.data[DYSON_FAN_DEVICES].append(dyson_entity)
add_devices(hass.data[DYSON_FAN_DEVICES])
descriptions = load_yaml_config_file(
path.join(path.dirname(__file__), 'services.yaml'))
def service_handle(service):
"""Handle dyson services."""
entity_id = service.data.get('entity_id')
night_mode = service.data.get('night_mode')
fan_device = next([fan for fan in hass.data[DYSON_FAN_DEVICES] if
fan.entity_id == entity_id].__iter__(), None)
if fan_device is None:
_LOGGER.warning("Unable to find Dyson fan device %s",
str(entity_id))
return
if service.service == SERVICE_SET_NIGHT_MODE:
fan_device.night_mode(night_mode)
# Register dyson service(s)
hass.services.register(DOMAIN, SERVICE_SET_NIGHT_MODE,
service_handle,
descriptions.get(SERVICE_SET_NIGHT_MODE),
schema=DYSON_SET_NIGHT_MODE_SCHEMA)
class DysonPureCoolLinkDevice(FanEntity):
"""Representation of a Dyson fan."""
def __init__(self, hass, device):
"""Initialize the fan."""
_LOGGER.info("Creating device %s", device.name)
self.hass = hass
self._device = device
@asyncio.coroutine
def async_added_to_hass(self):
"""Callback when entity is added to hass."""
self.hass.async_add_job(
self._device.add_message_listener(self.on_message))
def on_message(self, message):
"""Called when new messages received from the fan."""
_LOGGER.debug(
"Message received for fan device %s : %s", self.name, message)
self.schedule_update_ha_state()
@property
def should_poll(self):
"""No polling needed."""
return False
@property
def name(self):
"""Return the display name of this fan."""
return self._device.name
def set_speed(self: ToggleEntity, speed: str) -> None:
"""Set the speed of the fan. Never called ??."""
_LOGGER.debug("Set fan speed to: " + speed)
from libpurecoollink.const import FanSpeed, FanMode
if speed == FanSpeed.FAN_SPEED_AUTO.value:
self._device.set_configuration(fan_mode=FanMode.AUTO)
else:
fan_speed = FanSpeed('{0:04d}'.format(int(speed)))
self._device.set_configuration(fan_mode=FanMode.FAN,
fan_speed=fan_speed)
def turn_on(self: ToggleEntity, speed: str=None, **kwargs) -> None:
"""Turn on the fan."""
_LOGGER.debug("Turn on fan %s with speed %s", self.name, speed)
from libpurecoollink.const import FanSpeed, FanMode
if speed:
if speed == FanSpeed.FAN_SPEED_AUTO.value:
self._device.set_configuration(fan_mode=FanMode.AUTO)
else:
fan_speed = FanSpeed('{0:04d}'.format(int(speed)))
self._device.set_configuration(fan_mode=FanMode.FAN,
fan_speed=fan_speed)
else:
# Speed not set, just turn on
self._device.set_configuration(fan_mode=FanMode.FAN)
def turn_off(self: ToggleEntity, **kwargs) -> None:
"""Turn off the fan."""
_LOGGER.debug("Turn off fan %s", self.name)
from libpurecoollink.const import FanMode
self._device.set_configuration(fan_mode=FanMode.OFF)
def oscillate(self: ToggleEntity, oscillating: bool) -> None:
"""Turn on/off oscillating."""
_LOGGER.debug("Turn oscillation %s for device %s", oscillating,
self.name)
from libpurecoollink.const import Oscillation
if oscillating:
self._device.set_configuration(
oscillation=Oscillation.OSCILLATION_ON)
else:
self._device.set_configuration(
oscillation=Oscillation.OSCILLATION_OFF)
@property
def oscillating(self):
"""Return the oscillation state."""
return self._device.state and self._device.state.oscillation == "ON"
@property
def is_on(self):
"""Return true if the entity is on."""
if self._device.state:
return self._device.state.fan_state == "FAN"
return False
@property
def speed(self) -> str:
"""Return the current speed."""
if self._device.state:
from libpurecoollink.const import FanSpeed
if self._device.state.speed == FanSpeed.FAN_SPEED_AUTO.value:
return self._device.state.speed
else:
return int(self._device.state.speed)
return None
@property
def current_direction(self):
"""Return direction of the fan [forward, reverse]."""
return None
@property
def is_night_mode(self):
"""Return Night mode."""
return self._device.state.night_mode == "ON"
def night_mode(self: ToggleEntity, night_mode: bool) -> None:
"""Turn fan in night mode."""
_LOGGER.debug("Set %s night mode %s", self.name, night_mode)
from libpurecoollink.const import NightMode
if night_mode:
self._device.set_configuration(night_mode=NightMode.NIGHT_MODE_ON)
else:
self._device.set_configuration(night_mode=NightMode.NIGHT_MODE_OFF)
@property
def is_auto_mode(self):
"""Return auto mode."""
return self._device.state.fan_mode == "AUTO"
def auto_mode(self: ToggleEntity, auto_mode: bool) -> None:
"""Turn fan in auto mode."""
_LOGGER.debug("Set %s auto mode %s", self.name, auto_mode)
from libpurecoollink.const import FanMode
if auto_mode:
self._device.set_configuration(fan_mode=FanMode.AUTO)
else:
self._device.set_configuration(fan_mode=FanMode.FAN)
@property
def speed_list(self: ToggleEntity) -> list:
"""Get the list of available speeds."""
from libpurecoollink.const import FanSpeed
supported_speeds = [FanSpeed.FAN_SPEED_AUTO.value,
int(FanSpeed.FAN_SPEED_1.value),
int(FanSpeed.FAN_SPEED_2.value),
int(FanSpeed.FAN_SPEED_3.value),
int(FanSpeed.FAN_SPEED_4.value),
int(FanSpeed.FAN_SPEED_5.value),
int(FanSpeed.FAN_SPEED_6.value),
int(FanSpeed.FAN_SPEED_7.value),
int(FanSpeed.FAN_SPEED_8.value),
int(FanSpeed.FAN_SPEED_9.value),
int(FanSpeed.FAN_SPEED_10.value)]
return supported_speeds
@property
def supported_features(self: ToggleEntity) -> int:
"""Flag supported features."""
return SUPPORT_OSCILLATE | SUPPORT_SET_SPEED

View File

@ -58,7 +58,18 @@ set_direction:
fields:
entity_id:
description: Name(s) of the entities to toggle
exampl: 'fan.living_room'
example: 'fan.living_room'
direction:
description: The direction to rotate
example: 'left'
dyson_set_night_mode:
description: Set the fan in night mode
fields:
entity_id:
description: Name(s) of the entities to enable/disable night mode
example: 'fan.living_room'
night_mode:
description: Night mode status
example: true

View File

@ -0,0 +1,73 @@
"""Support for Dyson Pure Cool Link Sensors."""
import logging
import asyncio
from homeassistant.const import STATE_UNKNOWN
from homeassistant.components.dyson import DYSON_DEVICES
from homeassistant.helpers.entity import Entity
DEPENDENCIES = ['dyson']
SENSOR_UNITS = {'filter_life': 'hours'}
_LOGGER = logging.getLogger(__name__)
def setup_platform(hass, config, add_devices, discovery_info=None):
"""Set up the Dyson Sensors."""
_LOGGER.info("Creating new Dyson fans")
devices = []
# Get Dyson Devices from parent component
for device in hass.data[DYSON_DEVICES]:
devices.append(DysonFilterLifeSensor(hass, device))
add_devices(devices)
class DysonFilterLifeSensor(Entity):
"""Representation of Dyson filter life sensor (in hours)."""
def __init__(self, hass, device):
"""Create a new Dyson filter life sensor."""
self.hass = hass
self._device = device
self._name = "{} filter life".format(self._device.name)
self._old_value = None
@asyncio.coroutine
def async_added_to_hass(self):
"""Callback when entity is added to hass."""
self.hass.async_add_job(
self._device.add_message_listener(self.on_message))
def on_message(self, message):
"""Called when new messages received from the fan."""
_LOGGER.debug(
"Message received for %s device: %s", self.name, message)
# Prevent refreshing if not needed
if self._old_value is None or self._old_value != self.state:
self._old_value = self.state
self.schedule_update_ha_state()
@property
def should_poll(self):
"""No polling needed."""
return False
@property
def state(self):
"""Return filter life in hours.."""
if self._device.state:
return self._device.state.filter_life
else:
return STATE_UNKNOWN
@property
def name(self):
"""Return the name of the dyson sensor name."""
return self._name
@property
def unit_of_measurement(self):
"""Return the unit the value is expressed in."""
return SENSOR_UNITS['filter_life']

View File

@ -341,6 +341,9 @@ knxip==0.3.3
# homeassistant.components.device_tracker.owntracks
libnacl==1.5.0
# homeassistant.components.dyson
libpurecoollink==0.1.5
# homeassistant.components.device_tracker.mikrotik
librouteros==1.0.2

View File

@ -61,6 +61,9 @@ holidays==0.8.1
# homeassistant.components.sensor.influxdb
influxdb==3.0.0
# homeassistant.components.dyson
libpurecoollink==0.1.5
# homeassistant.components.media_player.soundtouch
libsoundtouch==0.3.0

View File

@ -40,6 +40,7 @@ TEST_REQUIREMENTS = (
'aioautomatic',
'SoCo',
'libsoundtouch',
'libpurecoollink',
'rxv',
'apns2',
'sqlalchemy',

View File

@ -0,0 +1,279 @@
"""Test the Dyson fan component."""
import unittest
from unittest import mock
from homeassistant.components.dyson import DYSON_DEVICES
from homeassistant.components.fan import dyson
from tests.common import get_test_home_assistant
from libpurecoollink.const import FanSpeed, FanMode, NightMode, Oscillation
def _get_device_with_no_state():
"""Return a device with no state."""
device = mock.Mock()
device.name = "Device_name"
device.state = None
return device
def _get_device_off():
"""Return a device with state off."""
device = mock.Mock()
device.name = "Device_name"
device.state = mock.Mock()
device.state.fan_mode = "OFF"
device.state.night_mode = "ON"
device.state.speed = "0004"
return device
def _get_device_auto():
"""Return a device with state auto."""
device = mock.Mock()
device.name = "Device_name"
device.state = mock.Mock()
device.state.fan_mode = "AUTO"
device.state.night_mode = "ON"
device.state.speed = "AUTO"
return device
def _get_device_on():
"""Return a valid state on."""
device = mock.Mock()
device.name = "Device_name"
device.state = mock.Mock()
device.state.fan_mode = "FAN"
device.state.fan_state = "FAN"
device.state.oscillation = "ON"
device.state.night_mode = "OFF"
device.state.speed = "0001"
return device
class DysonTest(unittest.TestCase):
"""Dyson Sensor component test class."""
def setUp(self): # pylint: disable=invalid-name
"""Setup things to be run when tests are started."""
self.hass = get_test_home_assistant()
def tearDown(self): # pylint: disable=invalid-name
"""Stop everything that was started."""
self.hass.stop()
def test_setup_component_with_no_devices(self):
"""Test setup component with no devices."""
self.hass.data[dyson.DYSON_DEVICES] = []
add_devices = mock.MagicMock()
dyson.setup_platform(self.hass, None, add_devices)
add_devices.assert_called_with([])
def test_setup_component(self):
"""Test setup component with devices."""
def _add_device(devices):
assert len(devices) == 1
assert devices[0].name == "Device_name"
device = _get_device_on()
self.hass.data[dyson.DYSON_DEVICES] = [device]
dyson.setup_platform(self.hass, None, _add_device)
def test_dyson_set_speed(self):
"""Test set fan speed."""
device = _get_device_on()
component = dyson.DysonPureCoolLinkDevice(self.hass, device)
self.assertFalse(component.should_poll)
component.set_speed("1")
set_config = device.set_configuration
set_config.assert_called_with(fan_mode=FanMode.FAN,
fan_speed=FanSpeed.FAN_SPEED_1)
component.set_speed("AUTO")
set_config = device.set_configuration
set_config.assert_called_with(fan_mode=FanMode.AUTO)
def test_dyson_turn_on(self):
"""Test turn on fan."""
device = _get_device_on()
component = dyson.DysonPureCoolLinkDevice(self.hass, device)
self.assertFalse(component.should_poll)
component.turn_on()
set_config = device.set_configuration
set_config.assert_called_with(fan_mode=FanMode.FAN)
def test_dyson_turn_night_mode(self):
"""Test turn on fan with night mode."""
device = _get_device_on()
component = dyson.DysonPureCoolLinkDevice(self.hass, device)
self.assertFalse(component.should_poll)
component.night_mode(True)
set_config = device.set_configuration
set_config.assert_called_with(night_mode=NightMode.NIGHT_MODE_ON)
component.night_mode(False)
set_config = device.set_configuration
set_config.assert_called_with(night_mode=NightMode.NIGHT_MODE_OFF)
def test_is_night_mode(self):
"""Test night mode."""
device = _get_device_on()
component = dyson.DysonPureCoolLinkDevice(self.hass, device)
self.assertFalse(component.is_night_mode)
device = _get_device_off()
component = dyson.DysonPureCoolLinkDevice(self.hass, device)
self.assertTrue(component.is_night_mode)
def test_dyson_turn_auto_mode(self):
"""Test turn on/off fan with auto mode."""
device = _get_device_on()
component = dyson.DysonPureCoolLinkDevice(self.hass, device)
self.assertFalse(component.should_poll)
component.auto_mode(True)
set_config = device.set_configuration
set_config.assert_called_with(fan_mode=FanMode.AUTO)
component.auto_mode(False)
set_config = device.set_configuration
set_config.assert_called_with(fan_mode=FanMode.FAN)
def test_is_auto_mode(self):
"""Test auto mode."""
device = _get_device_on()
component = dyson.DysonPureCoolLinkDevice(self.hass, device)
self.assertFalse(component.is_auto_mode)
device = _get_device_auto()
component = dyson.DysonPureCoolLinkDevice(self.hass, device)
self.assertTrue(component.is_auto_mode)
def test_dyson_turn_on_speed(self):
"""Test turn on fan with specified speed."""
device = _get_device_on()
component = dyson.DysonPureCoolLinkDevice(self.hass, device)
self.assertFalse(component.should_poll)
component.turn_on("1")
set_config = device.set_configuration
set_config.assert_called_with(fan_mode=FanMode.FAN,
fan_speed=FanSpeed.FAN_SPEED_1)
component.turn_on("AUTO")
set_config = device.set_configuration
set_config.assert_called_with(fan_mode=FanMode.AUTO)
def test_dyson_turn_off(self):
"""Test turn off fan."""
device = _get_device_on()
component = dyson.DysonPureCoolLinkDevice(self.hass, device)
self.assertFalse(component.should_poll)
component.turn_off()
set_config = device.set_configuration
set_config.assert_called_with(fan_mode=FanMode.OFF)
def test_dyson_oscillate_off(self):
"""Test turn off oscillation."""
device = _get_device_on()
component = dyson.DysonPureCoolLinkDevice(self.hass, device)
component.oscillate(False)
set_config = device.set_configuration
set_config.assert_called_with(oscillation=Oscillation.OSCILLATION_OFF)
def test_dyson_oscillate_on(self):
"""Test turn on oscillation."""
device = _get_device_on()
component = dyson.DysonPureCoolLinkDevice(self.hass, device)
component.oscillate(True)
set_config = device.set_configuration
set_config.assert_called_with(oscillation=Oscillation.OSCILLATION_ON)
def test_dyson_oscillate_value_on(self):
"""Test get oscillation value on."""
device = _get_device_on()
component = dyson.DysonPureCoolLinkDevice(self.hass, device)
self.assertTrue(component.oscillating)
def test_dyson_oscillate_value_off(self):
"""Test get oscillation value off."""
device = _get_device_off()
component = dyson.DysonPureCoolLinkDevice(self.hass, device)
self.assertFalse(component.oscillating)
def test_dyson_on(self):
"""Test device is on."""
device = _get_device_on()
component = dyson.DysonPureCoolLinkDevice(self.hass, device)
self.assertTrue(component.is_on)
def test_dyson_off(self):
"""Test device is off."""
device = _get_device_off()
component = dyson.DysonPureCoolLinkDevice(self.hass, device)
self.assertFalse(component.is_on)
device = _get_device_with_no_state()
component = dyson.DysonPureCoolLinkDevice(self.hass, device)
self.assertFalse(component.is_on)
def test_dyson_get_speed(self):
"""Test get device speed."""
device = _get_device_on()
component = dyson.DysonPureCoolLinkDevice(self.hass, device)
self.assertEqual(component.speed, 1)
device = _get_device_off()
component = dyson.DysonPureCoolLinkDevice(self.hass, device)
self.assertEqual(component.speed, 4)
device = _get_device_with_no_state()
component = dyson.DysonPureCoolLinkDevice(self.hass, device)
self.assertIsNone(component.speed)
device = _get_device_auto()
component = dyson.DysonPureCoolLinkDevice(self.hass, device)
self.assertEqual(component.speed, "AUTO")
def test_dyson_get_direction(self):
"""Test get device direction."""
device = _get_device_on()
component = dyson.DysonPureCoolLinkDevice(self.hass, device)
self.assertIsNone(component.current_direction)
def test_dyson_get_speed_list(self):
"""Test get speeds list."""
device = _get_device_on()
component = dyson.DysonPureCoolLinkDevice(self.hass, device)
self.assertEqual(len(component.speed_list), 11)
def test_dyson_supported_features(self):
"""Test supported features."""
device = _get_device_on()
component = dyson.DysonPureCoolLinkDevice(self.hass, device)
self.assertEqual(component.supported_features, 3)
def test_on_message(self):
"""Test when message is received."""
device = _get_device_on()
component = dyson.DysonPureCoolLinkDevice(self.hass, device)
component.entity_id = "entity_id"
component.schedule_update_ha_state = mock.Mock()
component.on_message("Message")
component.schedule_update_ha_state.assert_called_with()
def test_service_set_night_mode(self):
"""Test set night mode service."""
dyson_device = mock.MagicMock()
self.hass.data[DYSON_DEVICES] = []
dyson_device.entity_id = 'fan.living_room'
self.hass.data[dyson.DYSON_FAN_DEVICES] = [dyson_device]
dyson.setup_platform(self.hass, None, mock.MagicMock())
self.hass.services.call(dyson.DOMAIN, dyson.SERVICE_SET_NIGHT_MODE,
{"entity_id": "fan.bed_room",
"night_mode": True}, True)
assert not dyson_device.night_mode.called
self.hass.services.call(dyson.DOMAIN, dyson.SERVICE_SET_NIGHT_MODE,
{"entity_id": "fan.living_room",
"night_mode": True}, True)
dyson_device.night_mode.assert_called_with(True)

View File

@ -0,0 +1,76 @@
"""Test the Dyson sensor(s) component."""
import unittest
from unittest import mock
from homeassistant.const import STATE_UNKNOWN
from homeassistant.components.sensor import dyson
from tests.common import get_test_home_assistant
def _get_device_without_state():
"""Return a valid device provide by Dyson web services."""
device = mock.Mock()
device.name = "Device_name"
device.state = None
return device
def _get_with_state():
"""Return a valid device with state values."""
device = mock.Mock()
device.name = "Device_name"
device.state = mock.Mock()
device.state.filter_life = 100
return device
class DysonTest(unittest.TestCase):
"""Dyson Sensor component test class."""
def setUp(self): # pylint: disable=invalid-name
"""Setup things to be run when tests are started."""
self.hass = get_test_home_assistant()
def tearDown(self): # pylint: disable=invalid-name
"""Stop everything that was started."""
self.hass.stop()
def test_setup_component_with_no_devices(self):
"""Test setup component with no devices."""
self.hass.data[dyson.DYSON_DEVICES] = []
add_devices = mock.MagicMock()
dyson.setup_platform(self.hass, None, add_devices)
add_devices.assert_called_with([])
def test_setup_component(self):
"""Test setup component with devices."""
def _add_device(devices):
assert len(devices) == 1
assert devices[0].name == "Device_name filter life"
device = _get_device_without_state()
self.hass.data[dyson.DYSON_DEVICES] = [device]
dyson.setup_platform(self.hass, None, _add_device)
def test_dyson_filter_life_sensor(self):
"""Test sensor with no value."""
sensor = dyson.DysonFilterLifeSensor(self.hass,
_get_device_without_state())
sensor.entity_id = "sensor.dyson_1"
self.assertFalse(sensor.should_poll)
self.assertEqual(sensor.state, STATE_UNKNOWN)
self.assertEqual(sensor.unit_of_measurement, "hours")
self.assertEqual(sensor.name, "Device_name filter life")
self.assertEqual(sensor.entity_id, "sensor.dyson_1")
sensor.on_message('message')
def test_dyson_filter_life_sensor_with_values(self):
"""Test sensor with values."""
sensor = dyson.DysonFilterLifeSensor(self.hass, _get_with_state())
sensor.entity_id = "sensor.dyson_1"
self.assertFalse(sensor.should_poll)
self.assertEqual(sensor.state, 100)
self.assertEqual(sensor.unit_of_measurement, "hours")
self.assertEqual(sensor.name, "Device_name filter life")
self.assertEqual(sensor.entity_id, "sensor.dyson_1")
sensor.on_message('message')

View File

@ -0,0 +1,161 @@
"""Test the parent Dyson component."""
import unittest
from unittest import mock
from homeassistant.components import dyson
from tests.common import get_test_home_assistant
def _get_dyson_account_device_available():
"""Return a valid device provide by Dyson web services."""
device = mock.Mock()
device.serial = "XX-XXXXX-XX"
device.connect = mock.Mock(return_value=True)
return device
def _get_dyson_account_device_not_available():
"""Return an invalid device provide by Dyson web services."""
device = mock.Mock()
device.serial = "XX-XXXXX-XX"
device.connect = mock.Mock(return_value=False)
return device
class DysonTest(unittest.TestCase):
"""Dyson parent component test class."""
def setUp(self): # pylint: disable=invalid-name
"""Setup things to be run when tests are started."""
self.hass = get_test_home_assistant()
def tearDown(self): # pylint: disable=invalid-name
"""Stop everything that was started."""
self.hass.stop()
@mock.patch('libpurecoollink.dyson.DysonAccount.login', return_value=False)
def test_dyson_login_failed(self, mocked_login):
"""Test if Dyson connection failed."""
dyson.setup(self.hass, {dyson.DOMAIN: {
dyson.CONF_USERNAME: "email",
dyson.CONF_PASSWORD: "password",
dyson.CONF_LANGUAGE: "FR"
}})
self.assertEqual(mocked_login.call_count, 1)
@mock.patch('libpurecoollink.dyson.DysonAccount.devices', return_value=[])
@mock.patch('libpurecoollink.dyson.DysonAccount.login', return_value=True)
def test_dyson_login(self, mocked_login, mocked_devices):
"""Test valid connection to dyson web service."""
dyson.setup(self.hass, {dyson.DOMAIN: {
dyson.CONF_USERNAME: "email",
dyson.CONF_PASSWORD: "password",
dyson.CONF_LANGUAGE: "FR"
}})
self.assertEqual(mocked_login.call_count, 1)
self.assertEqual(mocked_devices.call_count, 1)
self.assertEqual(len(self.hass.data[dyson.DYSON_DEVICES]), 0)
@mock.patch('homeassistant.helpers.discovery.load_platform')
@mock.patch('libpurecoollink.dyson.DysonAccount.devices',
return_value=[_get_dyson_account_device_available()])
@mock.patch('libpurecoollink.dyson.DysonAccount.login', return_value=True)
def test_dyson_custom_conf(self, mocked_login, mocked_devices,
mocked_discovery):
"""Test device connection using custom configuration."""
dyson.setup(self.hass, {dyson.DOMAIN: {
dyson.CONF_USERNAME: "email",
dyson.CONF_PASSWORD: "password",
dyson.CONF_LANGUAGE: "FR",
dyson.CONF_DEVICES: [
{
"device_id": "XX-XXXXX-XX",
"device_ip": "192.168.0.1"
}
]
}})
self.assertEqual(mocked_login.call_count, 1)
self.assertEqual(mocked_devices.call_count, 1)
self.assertEqual(len(self.hass.data[dyson.DYSON_DEVICES]), 1)
self.assertEqual(mocked_discovery.call_count, 2)
@mock.patch('libpurecoollink.dyson.DysonAccount.devices',
return_value=[_get_dyson_account_device_not_available()])
@mock.patch('libpurecoollink.dyson.DysonAccount.login', return_value=True)
def test_dyson_custom_conf_device_not_available(self, mocked_login,
mocked_devices):
"""Test device connection with an invalid device."""
dyson.setup(self.hass, {dyson.DOMAIN: {
dyson.CONF_USERNAME: "email",
dyson.CONF_PASSWORD: "password",
dyson.CONF_LANGUAGE: "FR",
dyson.CONF_DEVICES: [
{
"device_id": "XX-XXXXX-XX",
"device_ip": "192.168.0.1"
}
]
}})
self.assertEqual(mocked_login.call_count, 1)
self.assertEqual(mocked_devices.call_count, 1)
self.assertEqual(len(self.hass.data[dyson.DYSON_DEVICES]), 0)
@mock.patch('homeassistant.helpers.discovery.load_platform')
@mock.patch('libpurecoollink.dyson.DysonAccount.devices',
return_value=[_get_dyson_account_device_available()])
@mock.patch('libpurecoollink.dyson.DysonAccount.login', return_value=True)
def test_dyson_custom_conf_with_unknown_device(self, mocked_login,
mocked_devices,
mocked_discovery):
"""Test device connection with custom conf and unknown device."""
dyson.setup(self.hass, {dyson.DOMAIN: {
dyson.CONF_USERNAME: "email",
dyson.CONF_PASSWORD: "password",
dyson.CONF_LANGUAGE: "FR",
dyson.CONF_DEVICES: [
{
"device_id": "XX-XXXXX-XY",
"device_ip": "192.168.0.1"
}
]
}})
self.assertEqual(mocked_login.call_count, 1)
self.assertEqual(mocked_devices.call_count, 1)
self.assertEqual(len(self.hass.data[dyson.DYSON_DEVICES]), 0)
self.assertEqual(mocked_discovery.call_count, 0)
@mock.patch('homeassistant.helpers.discovery.load_platform')
@mock.patch('libpurecoollink.dyson.DysonAccount.devices',
return_value=[_get_dyson_account_device_available()])
@mock.patch('libpurecoollink.dyson.DysonAccount.login', return_value=True)
def test_dyson_discovery(self, mocked_login, mocked_devices,
mocked_discovery):
"""Test device connection using discovery."""
dyson.setup(self.hass, {dyson.DOMAIN: {
dyson.CONF_USERNAME: "email",
dyson.CONF_PASSWORD: "password",
dyson.CONF_LANGUAGE: "FR",
dyson.CONF_TIMEOUT: 5,
dyson.CONF_RETRY: 2
}})
self.assertEqual(mocked_login.call_count, 1)
self.assertEqual(mocked_devices.call_count, 1)
self.assertEqual(len(self.hass.data[dyson.DYSON_DEVICES]), 1)
self.assertEqual(mocked_discovery.call_count, 2)
@mock.patch('libpurecoollink.dyson.DysonAccount.devices',
return_value=[_get_dyson_account_device_not_available()])
@mock.patch('libpurecoollink.dyson.DysonAccount.login', return_value=True)
def test_dyson_discovery_device_not_available(self, mocked_login,
mocked_devices):
"""Test device connection with discovery and invalid device."""
dyson.setup(self.hass, {dyson.DOMAIN: {
dyson.CONF_USERNAME: "email",
dyson.CONF_PASSWORD: "password",
dyson.CONF_LANGUAGE: "FR",
dyson.CONF_TIMEOUT: 5,
dyson.CONF_RETRY: 2
}})
self.assertEqual(mocked_login.call_count, 1)
self.assertEqual(mocked_devices.call_count, 1)
self.assertEqual(len(self.hass.data[dyson.DYSON_DEVICES]), 0)