jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-24 05:37:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

Rewritten HomeKit tests (#14377)

Browse Source 
* Use pytest fixtures and parametrize
* Use async
This commit is contained in:
cdce8p 2018-05-11 01:21:59 +02:00 committed by GitHub
parent 6843893d9f
commit 8fcf085829
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 1255 additions and 1591 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

220
tests/components/homekit/test_accessories.py
View File

@ -3,8 +3,7 @@
This includes tests for all mock object types.
"""
from datetime import datetime, timedelta
import unittest
from unittest.mock import call, patch, Mock
from unittest.mock import patch, Mock
from homeassistant.components.homekit.accessories import (
debounce, HomeAccessory, HomeBridge, HomeDriver)
@ -15,8 +14,6 @@ from homeassistant.components.homekit.const import (
from homeassistant.const import __version__, ATTR_NOW, EVENT_TIME_CHANGED
import homeassistant.util.dt as dt_util
from tests.common import get_test_home_assistant
def patch_debounce():
"""Return patch for debounce method."""
@ -24,141 +21,122 @@ def patch_debounce():
lambda f: lambda *args, **kwargs: f(*args, **kwargs))
class TestAccessories(unittest.TestCase):
"""Test pyhap adapter methods."""
async def test_debounce(hass):
"""Test add_timeout decorator function."""
def demo_func(*args):
nonlocal arguments, counter
counter += 1
arguments = args
def test_debounce(self):
"""Test add_timeout decorator function."""
def demo_func(*args):
nonlocal arguments, counter
counter += 1
arguments = args
arguments = None
counter = 0
mock = Mock(hass=hass)
arguments = None
counter = 0
hass = get_test_home_assistant()
mock = Mock(hass=hass)
debounce_demo = debounce(demo_func)
assert debounce_demo.__name__ == 'demo_func'
now = datetime(2018, 1, 1, 20, 0, 0, tzinfo=dt_util.UTC)
debounce_demo = debounce(demo_func)
self.assertEqual(debounce_demo.__name__, 'demo_func')
now = datetime(2018, 1, 1, 20, 0, 0, tzinfo=dt_util.UTC)
with patch('homeassistant.util.dt.utcnow', return_value=now):
await hass.async_add_job(debounce_demo, mock, 'value')
hass.bus.async_fire(
EVENT_TIME_CHANGED, {ATTR_NOW: now + timedelta(seconds=3)})
await hass.async_block_till_done()
assert counter == 1
assert len(arguments) == 2
with patch('homeassistant.util.dt.utcnow', return_value=now):
debounce_demo(mock, 'value')
hass.bus.fire(
EVENT_TIME_CHANGED, {ATTR_NOW: now + timedelta(seconds=3)})
hass.block_till_done()
assert counter == 1
assert len(arguments) == 2
with patch('homeassistant.util.dt.utcnow', return_value=now):
await hass.async_add_job(debounce_demo, mock, 'value')
await hass.async_add_job(debounce_demo, mock, 'value')
with patch('homeassistant.util.dt.utcnow', return_value=now):
debounce_demo(mock, 'value')
debounce_demo(mock, 'value')
hass.bus.async_fire(
EVENT_TIME_CHANGED, {ATTR_NOW: now + timedelta(seconds=3)})
await hass.async_block_till_done()
assert counter == 2
hass.bus.fire(
EVENT_TIME_CHANGED, {ATTR_NOW: now + timedelta(seconds=3)})
hass.block_till_done()
assert counter == 2
hass.stop()
async def test_home_accessory(hass):
"""Test HomeAccessory class."""
acc = HomeAccessory(hass, 'Home Accessory', 'homekit.accessory', 2)
assert acc.hass == hass
assert acc.display_name == 'Home Accessory'
assert acc.category == 1 # Category.OTHER
assert len(acc.services) == 1
serv = acc.services[0] # SERV_ACCESSORY_INFO
assert serv.display_name == SERV_ACCESSORY_INFO
assert serv.get_characteristic(CHAR_NAME).value == 'Home Accessory'
assert serv.get_characteristic(CHAR_MANUFACTURER).value == MANUFACTURER
assert serv.get_characteristic(CHAR_MODEL).value == 'Homekit'
assert serv.get_characteristic(CHAR_SERIAL_NUMBER).value == \
'homekit.accessory'
def test_home_accessory(self):
"""Test HomeAccessory class."""
hass = get_test_home_assistant()
hass.states.async_set('homekit.accessory', 'on')
await hass.async_block_till_done()
await hass.async_add_job(acc.run)
hass.states.async_set('homekit.accessory', 'off')
await hass.async_block_till_done()
acc = HomeAccessory(hass, 'Home Accessory', 'homekit.accessory', 2)
self.assertEqual(acc.hass, hass)
self.assertEqual(acc.display_name, 'Home Accessory')
self.assertEqual(acc.category, 1) # Category.OTHER
self.assertEqual(len(acc.services), 1)
serv = acc.services[0] # SERV_ACCESSORY_INFO
self.assertEqual(serv.display_name, SERV_ACCESSORY_INFO)
self.assertEqual(
serv.get_characteristic(CHAR_NAME).value, 'Home Accessory')
self.assertEqual(
serv.get_characteristic(CHAR_MANUFACTURER).value, MANUFACTURER)
self.assertEqual(
serv.get_characteristic(CHAR_MODEL).value, 'Homekit')
self.assertEqual(serv.get_characteristic(CHAR_SERIAL_NUMBER).value,
'homekit.accessory')
acc = HomeAccessory('hass', 'test_name', 'test_model.demo', 2)
assert acc.display_name == 'test_name'
assert acc.aid == 2
assert len(acc.services) == 1
serv = acc.services[0] # SERV_ACCESSORY_INFO
assert serv.get_characteristic(CHAR_MODEL).value == 'Test Model'
hass.states.set('homekit.accessory', 'on')
hass.block_till_done()
acc.run()
hass.states.set('homekit.accessory', 'off')
hass.block_till_done()
acc = HomeAccessory('hass', 'test_name', 'test_model.demo', 2)
self.assertEqual(acc.display_name, 'test_name')
self.assertEqual(acc.aid, 2)
self.assertEqual(len(acc.services), 1)
serv = acc.services[0] # SERV_ACCESSORY_INFO
self.assertEqual(
serv.get_characteristic(CHAR_MODEL).value, 'Test Model')
def test_home_bridge():
"""Test HomeBridge class."""
bridge = HomeBridge('hass')
assert bridge.hass == 'hass'
assert bridge.display_name == BRIDGE_NAME
assert bridge.category == 2 # Category.BRIDGE
assert len(bridge.services) == 1
serv = bridge.services[0] # SERV_ACCESSORY_INFO
assert serv.display_name == SERV_ACCESSORY_INFO
assert serv.get_characteristic(CHAR_NAME).value == BRIDGE_NAME
assert serv.get_characteristic(CHAR_FIRMWARE_REVISION).value == __version__
assert serv.get_characteristic(CHAR_MANUFACTURER).value == MANUFACTURER
assert serv.get_characteristic(CHAR_MODEL).value == BRIDGE_MODEL
assert serv.get_characteristic(CHAR_SERIAL_NUMBER).value == \
BRIDGE_SERIAL_NUMBER
hass.stop()
bridge = HomeBridge('hass', 'test_name')
assert bridge.display_name == 'test_name'
assert len(bridge.services) == 1
serv = bridge.services[0] # SERV_ACCESSORY_INFO
def test_home_bridge(self):
"""Test HomeBridge class."""
bridge = HomeBridge('hass')
self.assertEqual(bridge.hass, 'hass')
self.assertEqual(bridge.display_name, BRIDGE_NAME)
self.assertEqual(bridge.category, 2) # Category.BRIDGE
self.assertEqual(len(bridge.services), 1)
serv = bridge.services[0] # SERV_ACCESSORY_INFO
self.assertEqual(serv.display_name, SERV_ACCESSORY_INFO)
self.assertEqual(
serv.get_characteristic(CHAR_NAME).value, BRIDGE_NAME)
self.assertEqual(
serv.get_characteristic(CHAR_FIRMWARE_REVISION).value, __version__)
self.assertEqual(
serv.get_characteristic(CHAR_MANUFACTURER).value, MANUFACTURER)
self.assertEqual(
serv.get_characteristic(CHAR_MODEL).value, BRIDGE_MODEL)
self.assertEqual(
serv.get_characteristic(CHAR_SERIAL_NUMBER).value,
BRIDGE_SERIAL_NUMBER)
# setup_message
bridge.setup_message()
bridge = HomeBridge('hass', 'test_name')
self.assertEqual(bridge.display_name, 'test_name')
self.assertEqual(len(bridge.services), 1)
serv = bridge.services[0] # SERV_ACCESSORY_INFO
# add_paired_client
with patch('pyhap.accessory.Accessory.add_paired_client') \
as mock_add_paired_client, \
patch('homeassistant.components.homekit.accessories.'
'dismiss_setup_message') as mock_dissmiss_msg:
bridge.add_paired_client('client_uuid', 'client_public')
# setup_message
bridge.setup_message()
mock_add_paired_client.assert_called_with('client_uuid', 'client_public')
mock_dissmiss_msg.assert_called_with('hass')
# add_paired_client
with patch('pyhap.accessory.Accessory.add_paired_client') \
as mock_add_paired_client, \
patch('homeassistant.components.homekit.accessories.'
'dismiss_setup_message') as mock_dissmiss_msg:
bridge.add_paired_client('client_uuid', 'client_public')
# remove_paired_client
with patch('pyhap.accessory.Accessory.remove_paired_client') \
as mock_remove_paired_client, \
patch('homeassistant.components.homekit.accessories.'
'show_setup_message') as mock_show_msg:
bridge.remove_paired_client('client_uuid')
self.assertEqual(mock_add_paired_client.call_args,
call('client_uuid', 'client_public'))
self.assertEqual(mock_dissmiss_msg.call_args, call('hass'))
mock_remove_paired_client.assert_called_with('client_uuid')
mock_show_msg.assert_called_with('hass', bridge)
# remove_paired_client
with patch('pyhap.accessory.Accessory.remove_paired_client') \
as mock_remove_paired_client, \
patch('homeassistant.components.homekit.accessories.'
'show_setup_message') as mock_show_msg:
bridge.remove_paired_client('client_uuid')
self.assertEqual(
mock_remove_paired_client.call_args, call('client_uuid'))
self.assertEqual(mock_show_msg.call_args, call('hass', bridge))
def test_home_driver():
"""Test HomeDriver class."""
bridge = HomeBridge('hass')
ip_address = '127.0.0.1'
port = 51826
path = '.homekit.state'
def test_home_driver(self):
"""Test HomeDriver class."""
bridge = HomeBridge('hass')
ip_address = '127.0.0.1'
port = 51826
path = '.homekit.state'
with patch('pyhap.accessory_driver.AccessoryDriver.__init__') \
as mock_driver:
HomeDriver(bridge, ip_address, port, path)
with patch('pyhap.accessory_driver.AccessoryDriver.__init__') \
as mock_driver:
HomeDriver(bridge, ip_address, port, path)
self.assertEqual(
mock_driver.call_args, call(bridge, ip_address, port, path))
mock_driver.assert_called_with(bridge, ip_address, port, path)

253
tests/components/homekit/test_get_accessories.py
View File

@ -1,22 +1,20 @@
"""Package to test the get_accessory method."""
import logging
import unittest
from unittest.mock import patch, Mock
import pytest
from homeassistant.core import State
from homeassistant.components.cover import (
SUPPORT_OPEN, SUPPORT_CLOSE)
from homeassistant.components.cover import SUPPORT_OPEN, SUPPORT_CLOSE
from homeassistant.components.climate import (
SUPPORT_TARGET_TEMPERATURE_HIGH, SUPPORT_TARGET_TEMPERATURE_LOW)
from homeassistant.components.homekit import get_accessory, TYPES
from homeassistant.const import (
ATTR_CODE, ATTR_UNIT_OF_MEASUREMENT, ATTR_SUPPORTED_FEATURES,
TEMP_CELSIUS, TEMP_FAHRENHEIT, ATTR_DEVICE_CLASS)
ATTR_CODE, ATTR_DEVICE_CLASS, ATTR_SUPPORTED_FEATURES,
ATTR_UNIT_OF_MEASUREMENT, TEMP_CELSIUS, TEMP_FAHRENHEIT)
_LOGGER = logging.getLogger(__name__)
CONFIG = {}
def test_get_accessory_invalid_aid(caplog):
"""Test with unsupported component."""
@ -32,182 +30,93 @@ def test_not_supported():
is None
class TestGetAccessories(unittest.TestCase):
"""Methods to test the get_accessory method."""
@pytest.mark.parametrize('type_name, entity_id, state, attrs, config', [
('Light', 'light.test', 'on', {}, None),
('Lock', 'lock.test', 'locked', {}, None),
def setUp(self):
"""Setup Mock type."""
self.mock_type = Mock()
('Thermostat', 'climate.test', 'auto', {}, None),
('Thermostat', 'climate.test', 'auto',
{ATTR_SUPPORTED_FEATURES: SUPPORT_TARGET_TEMPERATURE_LOW |
SUPPORT_TARGET_TEMPERATURE_HIGH}, None),
def tearDown(self):
"""Test if mock type was called."""
self.assertTrue(self.mock_type.called)
('SecuritySystem', 'alarm_control_panel.test', 'armed', {},
{ATTR_CODE: '1234'}),
])
def test_types(type_name, entity_id, state, attrs, config):
"""Test if types are associated correctly."""
mock_type = Mock()
with patch.dict(TYPES, {type_name: mock_type}):
entity_state = State(entity_id, state, attrs)
get_accessory(None, entity_state, 2, config)
assert mock_type.called
def test_sensor_temperature(self):
"""Test temperature sensor with device class temperature."""
with patch.dict(TYPES, {'TemperatureSensor': self.mock_type}):
state = State('sensor.temperature', '23',
{ATTR_DEVICE_CLASS: 'temperature'})
get_accessory(None, state, 2, {})
if config:
assert mock_type.call_args[1]['config'] == config
def test_sensor_temperature_celsius(self):
"""Test temperature sensor with Celsius as unit."""
with patch.dict(TYPES, {'TemperatureSensor': self.mock_type}):
state = State('sensor.temperature', '23',
{ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS})
get_accessory(None, state, 2, {})
def test_sensor_temperature_fahrenheit(self):
"""Test temperature sensor with Fahrenheit as unit."""
with patch.dict(TYPES, {'TemperatureSensor': self.mock_type}):
state = State('sensor.temperature', '74',
{ATTR_UNIT_OF_MEASUREMENT: TEMP_FAHRENHEIT})
get_accessory(None, state, 2, {})
@pytest.mark.parametrize('type_name, entity_id, state, attrs', [
('GarageDoorOpener', 'cover.garage_door', 'open',
{ATTR_DEVICE_CLASS: 'garage',
ATTR_SUPPORTED_FEATURES: SUPPORT_OPEN | SUPPORT_CLOSE}),
('WindowCovering', 'cover.set_position', 'open',
{ATTR_SUPPORTED_FEATURES: 4}),
('WindowCoveringBasic', 'cover.open_window', 'open',
{ATTR_SUPPORTED_FEATURES: 3}),
])
def test_type_covers(type_name, entity_id, state, attrs):
"""Test if cover types are associated correctly."""
mock_type = Mock()
with patch.dict(TYPES, {type_name: mock_type}):
entity_state = State(entity_id, state, attrs)
get_accessory(None, entity_state, 2, None)
assert mock_type.called
def test_sensor_humidity(self):
"""Test humidity sensor with device class humidity."""
with patch.dict(TYPES, {'HumiditySensor': self.mock_type}):
state = State('sensor.humidity', '20',
{ATTR_DEVICE_CLASS: 'humidity',
ATTR_UNIT_OF_MEASUREMENT: '%'})
get_accessory(None, state, 2, {})
def test_air_quality_sensor(self):
"""Test air quality sensor with pm25 class."""
with patch.dict(TYPES, {'AirQualitySensor': self.mock_type}):
state = State('sensor.air_quality', '40',
{ATTR_DEVICE_CLASS: 'pm25'})
get_accessory(None, state, 2, {})
@pytest.mark.parametrize('type_name, entity_id, state, attrs', [
('BinarySensor', 'binary_sensor.opening', 'on',
{ATTR_DEVICE_CLASS: 'opening'}),
('BinarySensor', 'device_tracker.someone', 'not_home', {}),
def test_air_quality_sensor_entity_id(self):
"""Test air quality sensor with entity_id contains pm25."""
with patch.dict(TYPES, {'AirQualitySensor': self.mock_type}):
state = State('sensor.air_quality_pm25', '40', {})
get_accessory(None, state, 2, {})
('AirQualitySensor', 'sensor.air_quality_pm25', '40', {}),
('AirQualitySensor', 'sensor.air_quality', '40',
{ATTR_DEVICE_CLASS: 'pm25'}),
def test_co2_sensor(self):
"""Test co2 sensor with device class co2."""
with patch.dict(TYPES, {'CarbonDioxideSensor': self.mock_type}):
state = State('sensor.airmeter', '500',
{ATTR_DEVICE_CLASS: 'co2'})
get_accessory(None, state, 2, {})
('CarbonDioxideSensor', 'sensor.airmeter_co2', '500', {}),
('CarbonDioxideSensor', 'sensor.airmeter', '500',
{ATTR_DEVICE_CLASS: 'co2'}),
def test_co2_sensor_entity_id(self):
"""Test co2 sensor with entity_id contains co2."""
with patch.dict(TYPES, {'CarbonDioxideSensor': self.mock_type}):
state = State('sensor.airmeter_co2', '500', {})
get_accessory(None, state, 2, {})
('HumiditySensor', 'sensor.humidity', '20',
{ATTR_DEVICE_CLASS: 'humidity', ATTR_UNIT_OF_MEASUREMENT: '%'}),
def test_light_sensor(self):
"""Test light sensor with device class illuminance."""
with patch.dict(TYPES, {'LightSensor': self.mock_type}):
state = State('sensor.light', '900',
{ATTR_DEVICE_CLASS: 'illuminance'})
get_accessory(None, state, 2, {})
('LightSensor', 'sensor.light', '900', {ATTR_DEVICE_CLASS: 'illuminance'}),
('LightSensor', 'sensor.light', '900', {ATTR_UNIT_OF_MEASUREMENT: 'lm'}),
('LightSensor', 'sensor.light', '900', {ATTR_UNIT_OF_MEASUREMENT: 'lx'}),
def test_light_sensor_unit_lm(self):
"""Test light sensor with lm as unit."""
with patch.dict(TYPES, {'LightSensor': self.mock_type}):
state = State('sensor.light', '900',
{ATTR_UNIT_OF_MEASUREMENT: 'lm'})
get_accessory(None, state, 2, {})
('TemperatureSensor', 'sensor.temperature', '23',
{ATTR_DEVICE_CLASS: 'temperature'}),
('TemperatureSensor', 'sensor.temperature', '23',
{ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS}),
('TemperatureSensor', 'sensor.temperature', '74',
{ATTR_UNIT_OF_MEASUREMENT: TEMP_FAHRENHEIT}),
])
def test_type_sensors(type_name, entity_id, state, attrs):
"""Test if sensor types are associated correctly."""
mock_type = Mock()
with patch.dict(TYPES, {type_name: mock_type}):
entity_state = State(entity_id, state, attrs)
get_accessory(None, entity_state, 2, None)
assert mock_type.called
def test_light_sensor_unit_lx(self):
"""Test light sensor with lx as unit."""
with patch.dict(TYPES, {'LightSensor': self.mock_type}):
state = State('sensor.light', '900',
{ATTR_UNIT_OF_MEASUREMENT: 'lx'})
get_accessory(None, state, 2, {})
def test_binary_sensor(self):
"""Test binary sensor with opening class."""
with patch.dict(TYPES, {'BinarySensor': self.mock_type}):
state = State('binary_sensor.opening', 'on',
{ATTR_DEVICE_CLASS: 'opening'})
get_accessory(None, state, 2, {})
def test_device_tracker(self):
"""Test binary sensor with opening class."""
with patch.dict(TYPES, {'BinarySensor': self.mock_type}):
state = State('device_tracker.someone', 'not_home', {})
get_accessory(None, state, 2, {})
def test_garage_door(self):
"""Test cover with device_class: 'garage' and required features."""
with patch.dict(TYPES, {'GarageDoorOpener': self.mock_type}):
state = State('cover.garage_door', 'open', {
ATTR_DEVICE_CLASS: 'garage',
ATTR_SUPPORTED_FEATURES:
SUPPORT_OPEN | SUPPORT_CLOSE})
get_accessory(None, state, 2, {})
def test_cover_set_position(self):
"""Test cover with support for set_cover_position."""
with patch.dict(TYPES, {'WindowCovering': self.mock_type}):
state = State('cover.set_position', 'open',
{ATTR_SUPPORTED_FEATURES: 4})
get_accessory(None, state, 2, {})
def test_cover_open_close(self):
"""Test cover with support for open and close."""
with patch.dict(TYPES, {'WindowCoveringBasic': self.mock_type}):
state = State('cover.open_window', 'open',
{ATTR_SUPPORTED_FEATURES: 3})
get_accessory(None, state, 2, {})
def test_alarm_control_panel(self):
"""Test alarm control panel."""
config = {ATTR_CODE: '1234'}
with patch.dict(TYPES, {'SecuritySystem': self.mock_type}):
state = State('alarm_control_panel.test', 'armed')
get_accessory(None, state, 2, config)
# pylint: disable=unsubscriptable-object
print(self.mock_type.call_args[1])
self.assertEqual(
self.mock_type.call_args[1]['config'][ATTR_CODE], '1234')
def test_climate(self):
"""Test climate devices."""
with patch.dict(TYPES, {'Thermostat': self.mock_type}):
state = State('climate.test', 'auto')
get_accessory(None, state, 2, {})
def test_light(self):
"""Test light devices."""
with patch.dict(TYPES, {'Light': self.mock_type}):
state = State('light.test', 'on')
get_accessory(None, state, 2, {})
def test_climate_support_auto(self):
"""Test climate devices with support for auto mode."""
with patch.dict(TYPES, {'Thermostat': self.mock_type}):
state = State('climate.test', 'auto', {
ATTR_SUPPORTED_FEATURES:
SUPPORT_TARGET_TEMPERATURE_LOW |
SUPPORT_TARGET_TEMPERATURE_HIGH})
get_accessory(None, state, 2, {})
def test_switch(self):
"""Test switch."""
with patch.dict(TYPES, {'Switch': self.mock_type}):
state = State('switch.test', 'on')
get_accessory(None, state, 2, {})
def test_remote(self):
"""Test remote."""
with patch.dict(TYPES, {'Switch': self.mock_type}):
state = State('remote.test', 'on')
get_accessory(None, state, 2, {})
def test_input_boolean(self):
"""Test input_boolean."""
with patch.dict(TYPES, {'Switch': self.mock_type}):
state = State('input_boolean.test', 'on')
get_accessory(None, state, 2, {})
def test_lock(self):
"""Test lock."""
with patch.dict(TYPES, {'Lock': self.mock_type}):
state = State('lock.test', 'locked')
get_accessory(None, state, 2, {})
@pytest.mark.parametrize('type_name, entity_id, state, attrs', [
('Switch', 'switch.test', 'on', {}),
('Switch', 'remote.test', 'on', {}),
('Switch', 'input_boolean.test', 'on', {}),
])
def test_type_switches(type_name, entity_id, state, attrs):
"""Test if switch types are associated correctly."""
mock_type = Mock()
with patch.dict(TYPES, {type_name: mock_type}):
entity_state = State(entity_id, state, attrs)
get_accessory(None, entity_state, 2, None)
assert mock_type.called

308
tests/components/homekit/test_homekit.py
View File

@ -1,6 +1,7 @@
"""Tests for the HomeKit component."""
import unittest
from unittest.mock import call, patch, ANY, Mock
from unittest.mock import patch, ANY, Mock
import pytest
from homeassistant import setup
from homeassistant.core import State
@ -16,208 +17,193 @@ from homeassistant.const import (
CONF_IP_ADDRESS, CONF_PORT,
EVENT_HOMEASSISTANT_START, EVENT_HOMEASSISTANT_STOP)
from tests.common import get_test_home_assistant
from tests.components.homekit.test_accessories import patch_debounce
IP_ADDRESS = '127.0.0.1'
PATH_HOMEKIT = 'homeassistant.components.homekit'
class TestHomeKit(unittest.TestCase):
"""Test setup of HomeKit component and HomeKit class."""
@pytest.fixture('module')
def debounce_patcher(request):
"""Patch debounce method."""
patcher = patch_debounce()
patcher.start()
request.addfinalizer(patcher.stop)
@classmethod
def setUpClass(cls):
"""Setup debounce patcher."""
cls.patcher = patch_debounce()
cls.patcher.start()
@classmethod
def tearDownClass(cls):
"""Stop debounce patcher."""
cls.patcher.stop()
def test_generate_aid():
"""Test generate aid method."""
aid = generate_aid('demo.entity')
assert isinstance(aid, int)
assert aid >= 2 and aid <= 18446744073709551615
def setUp(self):
"""Setup things to be run when tests are started."""
self.hass = get_test_home_assistant()
with patch(PATH_HOMEKIT + '.adler32') as mock_adler32:
mock_adler32.side_effect = [0, 1]
assert generate_aid('demo.entity') is None
def tearDown(self):
"""Stop down everything that was started."""
self.hass.stop()
def test_generate_aid(self):
"""Test generate aid method."""
aid = generate_aid('demo.entity')
self.assertIsInstance(aid, int)
self.assertTrue(aid >= 2 and aid <= 18446744073709551615)
async def test_setup_min(hass):
"""Test async_setup with min config options."""
with patch(PATH_HOMEKIT + '.HomeKit') as mock_homekit:
assert await setup.async_setup_component(
hass, DOMAIN, {DOMAIN: {}})
with patch(PATH_HOMEKIT + '.adler32') as mock_adler32:
mock_adler32.side_effect = [0, 1]
self.assertIsNone(generate_aid('demo.entity'))
mock_homekit.assert_any_call(hass, DEFAULT_PORT, None, ANY, {})
assert mock_homekit().setup.called is True
@patch(PATH_HOMEKIT + '.HomeKit')
def test_setup_min(self, mock_homekit):
"""Test async_setup with min config options."""
self.assertTrue(setup.setup_component(
self.hass, DOMAIN, {DOMAIN: {}}))
# Test auto start enabled
mock_homekit.reset_mock()
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
await hass.async_block_till_done()
self.assertEqual(mock_homekit.mock_calls, [
call(self.hass, DEFAULT_PORT, None, ANY, {}),
call().setup()])
mock_homekit().start.assert_called_with(ANY)
# Test auto start enabled
mock_homekit.reset_mock()
self.hass.bus.fire(EVENT_HOMEASSISTANT_START)
self.hass.block_till_done()
self.assertEqual(mock_homekit.mock_calls, [call().start(ANY)])
async def test_setup_auto_start_disabled(hass):
"""Test async_setup with auto start disabled and test service calls."""
config = {DOMAIN: {CONF_AUTO_START: False, CONF_PORT: 11111,
CONF_IP_ADDRESS: '172.0.0.0'}}
@patch(PATH_HOMEKIT + '.HomeKit')
def test_setup_auto_start_disabled(self, mock_homekit):
"""Test async_setup with auto start disabled and test service calls."""
with patch(PATH_HOMEKIT + '.HomeKit') as mock_homekit:
mock_homekit.return_value = homekit = Mock()
assert await setup.async_setup_component(
hass, DOMAIN, config)
config = {DOMAIN: {CONF_AUTO_START: False, CONF_PORT: 11111,
CONF_IP_ADDRESS: '172.0.0.0'}}
self.assertTrue(setup.setup_component(
self.hass, DOMAIN, config))
self.hass.block_till_done()
mock_homekit.assert_any_call(hass, 11111, '172.0.0.0', ANY, {})
assert mock_homekit().setup.called is True
self.assertEqual(mock_homekit.mock_calls, [
call(self.hass, 11111, '172.0.0.0', ANY, {}),
call().setup()])
# Test auto_start disabled
homekit.reset_mock()
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
await hass.async_block_till_done()
assert homekit.start.called is False
# Test auto_start disabled
homekit.reset_mock()
self.hass.bus.fire(EVENT_HOMEASSISTANT_START)
self.hass.block_till_done()
self.assertEqual(homekit.mock_calls, [])
# Test start call with driver is ready
homekit.reset_mock()
homekit.status = STATUS_READY
# Test start call with driver is ready
homekit.reset_mock()
homekit.status = STATUS_READY
await hass.services.async_call(
DOMAIN, SERVICE_HOMEKIT_START, blocking=True)
assert homekit.start.called is True
self.hass.services.call('homekit', 'start')
self.assertEqual(homekit.mock_calls, [call.start()])
# Test start call with driver started
homekit.reset_mock()
homekit.status = STATUS_STOPPED
# Test start call with driver started
homekit.reset_mock()
homekit.status = STATUS_STOPPED
await hass.services.async_call(
DOMAIN, SERVICE_HOMEKIT_START, blocking=True)
assert homekit.start.called is False
self.hass.services.call(DOMAIN, SERVICE_HOMEKIT_START)
self.assertEqual(homekit.mock_calls, [])
def test_homekit_setup(self):
"""Test setup of bridge and driver."""
homekit = HomeKit(self.hass, DEFAULT_PORT, None, {}, {})
async def test_homekit_setup(hass):
"""Test setup of bridge and driver."""
homekit = HomeKit(hass, DEFAULT_PORT, None, {}, {})
with patch(PATH_HOMEKIT + '.accessories.HomeDriver') as mock_driver, \
patch('homeassistant.util.get_local_ip') as mock_ip:
mock_ip.return_value = IP_ADDRESS
homekit.setup()
with patch(PATH_HOMEKIT + '.accessories.HomeDriver') as mock_driver, \
patch('homeassistant.util.get_local_ip') as mock_ip:
mock_ip.return_value = IP_ADDRESS
await hass.async_add_job(homekit.setup)
path = self.hass.config.path(HOMEKIT_FILE)
self.assertTrue(isinstance(homekit.bridge, HomeBridge))
self.assertEqual(mock_driver.mock_calls, [
call(homekit.bridge, DEFAULT_PORT, IP_ADDRESS, path)])
path = hass.config.path(HOMEKIT_FILE)
assert isinstance(homekit.bridge, HomeBridge)
mock_driver.assert_called_with(
homekit.bridge, DEFAULT_PORT, IP_ADDRESS, path)
# Test if stop listener is setup
self.assertEqual(
self.hass.bus.listeners.get(EVENT_HOMEASSISTANT_STOP), 1)
# Test if stop listener is setup
assert hass.bus.async_listeners().get(EVENT_HOMEASSISTANT_STOP) == 1
def test_homekit_setup_ip_address(self):
"""Test setup with given IP address."""
homekit = HomeKit(self.hass, DEFAULT_PORT, '172.0.0.0', {}, {})
with patch(PATH_HOMEKIT + '.accessories.HomeDriver') as mock_driver:
homekit.setup()
mock_driver.assert_called_with(ANY, DEFAULT_PORT, '172.0.0.0', ANY)
async def test_homekit_setup_ip_address(hass):
"""Test setup with given IP address."""
homekit = HomeKit(hass, DEFAULT_PORT, '172.0.0.0', {}, {})
def test_homekit_add_accessory(self):
"""Add accessory if config exists and get_acc returns an accessory."""
homekit = HomeKit(self.hass, None, None, lambda entity_id: True, {})
homekit.bridge = HomeBridge(self.hass)
with patch(PATH_HOMEKIT + '.accessories.HomeDriver') as mock_driver:
await hass.async_add_job(homekit.setup)
mock_driver.assert_called_with(ANY, DEFAULT_PORT, '172.0.0.0', ANY)
with patch(PATH_HOMEKIT + '.accessories.HomeBridge.add_accessory') \
as mock_add_acc, \
patch(PATH_HOMEKIT + '.get_accessory') as mock_get_acc:
mock_get_acc.side_effect = [None, 'acc', None]
homekit.add_bridge_accessory(State('light.demo', 'on'))
self.assertEqual(mock_get_acc.call_args,
call(self.hass, ANY, 363398124, {}))
self.assertFalse(mock_add_acc.called)
homekit.add_bridge_accessory(State('demo.test', 'on'))
self.assertEqual(mock_get_acc.call_args,
call(self.hass, ANY, 294192020, {}))
self.assertTrue(mock_add_acc.called)
homekit.add_bridge_accessory(State('demo.test_2', 'on'))
self.assertEqual(mock_get_acc.call_args,
call(self.hass, ANY, 429982757, {}))
self.assertEqual(mock_add_acc.mock_calls, [call('acc')])
def test_homekit_entity_filter(self):
"""Test the entity filter."""
entity_filter = generate_filter(['cover'], ['demo.test'], [], [])
homekit = HomeKit(self.hass, None, None, entity_filter, {})
async def test_homekit_add_accessory(hass):
"""Add accessory if config exists and get_acc returns an accessory."""
homekit = HomeKit(hass, None, None, lambda entity_id: True, {})
homekit.bridge = HomeBridge(hass)
with patch(PATH_HOMEKIT + '.get_accessory') as mock_get_acc:
mock_get_acc.return_value = None
with patch(PATH_HOMEKIT + '.accessories.HomeBridge.add_accessory') \
as mock_add_acc, \
patch(PATH_HOMEKIT + '.get_accessory') as mock_get_acc:
homekit.add_bridge_accessory(State('cover.test', 'open'))
self.assertTrue(mock_get_acc.called)
mock_get_acc.reset_mock()
mock_get_acc.side_effect = [None, 'acc', None]
homekit.add_bridge_accessory(State('light.demo', 'on'))
mock_get_acc.assert_called_with(hass, ANY, 363398124, {})
assert mock_add_acc.called is False
homekit.add_bridge_accessory(State('demo.test', 'on'))
self.assertTrue(mock_get_acc.called)
mock_get_acc.reset_mock()
homekit.add_bridge_accessory(State('demo.test', 'on'))
mock_get_acc.assert_called_with(hass, ANY, 294192020, {})
assert mock_add_acc.called is True
homekit.add_bridge_accessory(State('light.demo', 'light'))
self.assertFalse(mock_get_acc.called)
homekit.add_bridge_accessory(State('demo.test_2', 'on'))
mock_get_acc.assert_called_with(hass, ANY, 429982757, {})
mock_add_acc.assert_called_with('acc')
@patch(PATH_HOMEKIT + '.show_setup_message')
@patch(PATH_HOMEKIT + '.HomeKit.add_bridge_accessory')
def test_homekit_start(self, mock_add_bridge_acc, mock_show_setup_msg):
"""Test HomeKit start method."""
homekit = HomeKit(self.hass, None, None, {}, {'cover.demo': {}})
homekit.bridge = HomeBridge(self.hass)
homekit.driver = Mock()
self.hass.states.set('light.demo', 'on')
state = self.hass.states.all()[0]
async def test_homekit_entity_filter(hass):
"""Test the entity filter."""
entity_filter = generate_filter(['cover'], ['demo.test'], [], [])
homekit = HomeKit(hass, None, None, entity_filter, {})
homekit.start()
self.hass.block_till_done()
with patch(PATH_HOMEKIT + '.get_accessory') as mock_get_acc:
mock_get_acc.return_value = None
self.assertEqual(mock_add_bridge_acc.mock_calls, [call(state)])
self.assertEqual(mock_show_setup_msg.mock_calls, [
call(self.hass, homekit.bridge)])
self.assertEqual(homekit.driver.mock_calls, [call.start()])
self.assertEqual(homekit.status, STATUS_RUNNING)
homekit.add_bridge_accessory(State('cover.test', 'open'))
assert mock_get_acc.called is True
mock_get_acc.reset_mock()
# Test start() if already started
homekit.driver.reset_mock()
homekit.start()
self.hass.block_till_done()
self.assertEqual(homekit.driver.mock_calls, [])
homekit.add_bridge_accessory(State('demo.test', 'on'))
assert mock_get_acc.called is True
mock_get_acc.reset_mock()
def test_homekit_stop(self):
"""Test HomeKit stop method."""
homekit = HomeKit(self.hass, None, None, None, None)
homekit.driver = Mock()
homekit.add_bridge_accessory(State('light.demo', 'light'))
assert mock_get_acc.called is False
self.assertEqual(homekit.status, STATUS_READY)
homekit.stop()
self.hass.block_till_done()
homekit.status = STATUS_WAIT
homekit.stop()
self.hass.block_till_done()
homekit.status = STATUS_STOPPED
homekit.stop()
self.hass.block_till_done()
self.assertFalse(homekit.driver.stop.called)
# Test if driver is started
homekit.status = STATUS_RUNNING
homekit.stop()
self.hass.block_till_done()
self.assertTrue(homekit.driver.stop.called)
async def test_homekit_start(hass, debounce_patcher):
"""Test HomeKit start method."""
homekit = HomeKit(hass, None, None, {}, {'cover.demo': {}})
homekit.bridge = HomeBridge(hass)
homekit.driver = Mock()
hass.states.async_set('light.demo', 'on')
state = hass.states.async_all()[0]
with patch(PATH_HOMEKIT + '.HomeKit.add_bridge_accessory') as \
mock_add_acc, \
patch(PATH_HOMEKIT + '.show_setup_message') as mock_setup_msg:
await hass.async_add_job(homekit.start)
mock_add_acc.assert_called_with(state)
mock_setup_msg.assert_called_with(hass, homekit.bridge)
assert homekit.driver.start.called is True
assert homekit.status == STATUS_RUNNING
# Test start() if already started
homekit.driver.reset_mock()
await hass.async_add_job(homekit.start)
assert homekit.driver.start.called is False
async def test_homekit_stop(hass):
"""Test HomeKit stop method."""
homekit = HomeKit(hass, None, None, None, None)
homekit.driver = Mock()
assert homekit.status == STATUS_READY
await hass.async_add_job(homekit.stop)
homekit.status = STATUS_WAIT
await hass.async_add_job(homekit.stop)
homekit.status = STATUS_STOPPED
await hass.async_add_job(homekit.stop)
assert homekit.driver.stop.called is False
# Test if driver is started
homekit.status = STATUS_RUNNING
await hass.async_add_job(homekit.stop)
assert homekit.driver.stop.called is True

396
tests/components/homekit/test_type_covers.py
View File

@ -1,265 +1,231 @@
"""Test different accessory types: Covers."""
import unittest
from collections import namedtuple
import pytest
from homeassistant.core import callback
from homeassistant.components.cover import (
ATTR_POSITION, ATTR_CURRENT_POSITION, SUPPORT_STOP)
DOMAIN, ATTR_CURRENT_POSITION, ATTR_POSITION, SUPPORT_STOP)
from homeassistant.const import (
STATE_CLOSED, STATE_UNAVAILABLE, STATE_UNKNOWN, STATE_OPEN,
ATTR_SERVICE, ATTR_SERVICE_DATA, EVENT_CALL_SERVICE,
ATTR_SUPPORTED_FEATURES)
ATTR_ENTITY_ID, ATTR_SUPPORTED_FEATURES,
STATE_CLOSED, STATE_OPEN, STATE_UNAVAILABLE, STATE_UNKNOWN)
from tests.common import get_test_home_assistant
from tests.common import async_mock_service
from tests.components.homekit.test_accessories import patch_debounce
class TestHomekitCovers(unittest.TestCase):
"""Test class for all accessory types regarding covers."""
@pytest.fixture(scope='module')
def cls(request):
"""Patch debounce decorator during import of type_covers."""
patcher = patch_debounce()
patcher.start()
_import = __import__('homeassistant.components.homekit.type_covers',
fromlist=['GarageDoorOpener', 'WindowCovering,',
'WindowCoveringBasic'])
request.addfinalizer(patcher.stop)
patcher_tuple = namedtuple('Cls', ['window', 'window_basic', 'garage'])
return patcher_tuple(window=_import.WindowCovering,
window_basic=_import.WindowCoveringBasic,
garage=_import.GarageDoorOpener)
@classmethod
def setUpClass(cls):
"""Setup Light class import and debounce patcher."""
cls.patcher = patch_debounce()
cls.patcher.start()
_import = __import__('homeassistant.components.homekit.type_covers',
fromlist=['GarageDoorOpener', 'WindowCovering,',
'WindowCoveringBasic'])
cls.garage_cls = _import.GarageDoorOpener
cls.window_cls = _import.WindowCovering
cls.window_basic_cls = _import.WindowCoveringBasic
@classmethod
def tearDownClass(cls):
"""Stop debounce patcher."""
cls.patcher.stop()
async def test_garage_door_open_close(hass, cls):
"""Test if accessory and HA are updated accordingly."""
entity_id = 'cover.garage_door'
def setUp(self):
"""Setup things to be run when tests are started."""
self.hass = get_test_home_assistant()
self.events = []
acc = cls.garage(hass, 'Garage Door', entity_id, 2, config=None)
await hass.async_add_job(acc.run)
@callback
def record_event(event):
"""Track called event."""
self.events.append(event)
assert acc.aid == 2
assert acc.category == 4 # GarageDoorOpener
self.hass.bus.listen(EVENT_CALL_SERVICE, record_event)
assert acc.char_current_state.value == 0
assert acc.char_target_state.value == 0
def tearDown(self):
"""Stop down everything that was started."""
self.hass.stop()
hass.states.async_set(entity_id, STATE_CLOSED)
await hass.async_block_till_done()
assert acc.char_current_state.value == 1
assert acc.char_target_state.value == 1
def test_garage_door_open_close(self):
"""Test if accessory and HA are updated accordingly."""
garage_door = 'cover.garage_door'
hass.states.async_set(entity_id, STATE_OPEN)
await hass.async_block_till_done()
assert acc.char_current_state.value == 0
assert acc.char_target_state.value == 0
acc = self.garage_cls(self.hass, 'Cover', garage_door, 2, config=None)
acc.run()
hass.states.async_set(entity_id, STATE_UNAVAILABLE)
await hass.async_block_till_done()
assert acc.char_current_state.value == 0
assert acc.char_target_state.value == 0
self.assertEqual(acc.aid, 2)
self.assertEqual(acc.category, 4) # GarageDoorOpener
hass.states.async_set(entity_id, STATE_UNKNOWN)
await hass.async_block_till_done()
assert acc.char_current_state.value == 0
assert acc.char_target_state.value == 0
self.assertEqual(acc.char_current_state.value, 0)
self.assertEqual(acc.char_target_state.value, 0)
# Set from HomeKit
call_close_cover = async_mock_service(hass, DOMAIN, 'close_cover')
call_open_cover = async_mock_service(hass, DOMAIN, 'open_cover')
self.hass.states.set(garage_door, STATE_CLOSED)
self.hass.block_till_done()
await hass.async_add_job(acc.char_target_state.client_update_value, 1)
await hass.async_block_till_done()
assert call_close_cover
assert call_close_cover[0].data[ATTR_ENTITY_ID] == entity_id
assert acc.char_current_state.value == 2
assert acc.char_target_state.value == 1
self.assertEqual(acc.char_current_state.value, 1)
self.assertEqual(acc.char_target_state.value, 1)
hass.states.async_set(entity_id, STATE_CLOSED)
await hass.async_block_till_done()
self.hass.states.set(garage_door, STATE_OPEN)
self.hass.block_till_done()
await hass.async_add_job(acc.char_target_state.client_update_value, 0)
await hass.async_block_till_done()
assert call_open_cover
assert call_open_cover[0].data[ATTR_ENTITY_ID] == entity_id
assert acc.char_current_state.value == 3
assert acc.char_target_state.value == 0
self.assertEqual(acc.char_current_state.value, 0)
self.assertEqual(acc.char_target_state.value, 0)
self.hass.states.set(garage_door, STATE_UNAVAILABLE)
self.hass.block_till_done()
async def test_window_set_cover_position(hass, cls):
"""Test if accessory and HA are updated accordingly."""
entity_id = 'cover.window'
self.assertEqual(acc.char_current_state.value, 0)
self.assertEqual(acc.char_target_state.value, 0)
acc = cls.window(hass, 'Cover', entity_id, 2, config=None)
await hass.async_add_job(acc.run)
self.hass.states.set(garage_door, STATE_UNKNOWN)
self.hass.block_till_done()
assert acc.aid == 2
assert acc.category == 14 # WindowCovering
self.assertEqual(acc.char_current_state.value, 0)
self.assertEqual(acc.char_target_state.value, 0)
assert acc.char_current_position.value == 0
assert acc.char_target_position.value == 0
# Set closed from HomeKit
acc.char_target_state.client_update_value(1)
self.hass.block_till_done()
hass.states.async_set(entity_id, STATE_UNKNOWN,
{ATTR_CURRENT_POSITION: None})
await hass.async_block_till_done()
assert acc.char_current_position.value == 0
assert acc.char_target_position.value == 0
self.assertEqual(acc.char_current_state.value, 2)
self.assertEqual(acc.char_target_state.value, 1)
self.assertEqual(
self.events[0].data[ATTR_SERVICE], 'close_cover')
hass.states.async_set(entity_id, STATE_OPEN,
{ATTR_CURRENT_POSITION: 50})
await hass.async_block_till_done()
assert acc.char_current_position.value == 50
assert acc.char_target_position.value == 50
self.hass.states.set(garage_door, STATE_CLOSED)
self.hass.block_till_done()
# Set from HomeKit
call_set_cover_position = async_mock_service(hass, DOMAIN,
'set_cover_position')
# Set open from HomeKit
acc.char_target_state.client_update_value(0)
self.hass.block_till_done()
await hass.async_add_job(acc.char_target_position.client_update_value, 25)
await hass.async_block_till_done()
assert call_set_cover_position[0]
assert call_set_cover_position[0].data[ATTR_ENTITY_ID] == entity_id
assert call_set_cover_position[0].data[ATTR_POSITION] == 25
assert acc.char_current_position.value == 50
assert acc.char_target_position.value == 25
self.assertEqual(acc.char_current_state.value, 3)
self.assertEqual(acc.char_target_state.value, 0)
self.assertEqual(
self.events[1].data[ATTR_SERVICE], 'open_cover')
await hass.async_add_job(acc.char_target_position.client_update_value, 75)
await hass.async_block_till_done()
assert call_set_cover_position[1]
assert call_set_cover_position[1].data[ATTR_ENTITY_ID] == entity_id
assert call_set_cover_position[1].data[ATTR_POSITION] == 75
assert acc.char_current_position.value == 50
assert acc.char_target_position.value == 75
def test_window_set_cover_position(self):
"""Test if accessory and HA are updated accordingly."""
window_cover = 'cover.window'
acc = self.window_cls(self.hass, 'Cover', window_cover, 2, config=None)
acc.run()
async def test_window_open_close(hass, cls):
"""Test if accessory and HA are updated accordingly."""
entity_id = 'cover.window'
self.assertEqual(acc.aid, 2)
self.assertEqual(acc.category, 14) # WindowCovering
hass.states.async_set(entity_id, STATE_UNKNOWN,
{ATTR_SUPPORTED_FEATURES: 0})
acc = cls.window_basic(hass, 'Cover', entity_id, 2, config=None)
await hass.async_add_job(acc.run)
self.assertEqual(acc.char_current_position.value, 0)
self.assertEqual(acc.char_target_position.value, 0)
assert acc.aid == 2
assert acc.category == 14 # WindowCovering
self.hass.states.set(window_cover, STATE_UNKNOWN,
{ATTR_CURRENT_POSITION: None})
self.hass.block_till_done()
assert acc.char_current_position.value == 0
assert acc.char_target_position.value == 0
assert acc.char_position_state.value == 2
self.assertEqual(acc.char_current_position.value, 0)
self.assertEqual(acc.char_target_position.value, 0)
hass.states.async_set(entity_id, STATE_UNKNOWN)
await hass.async_block_till_done()
assert acc.char_current_position.value == 0
assert acc.char_target_position.value == 0
assert acc.char_position_state.value == 2
self.hass.states.set(window_cover, STATE_OPEN,
{ATTR_CURRENT_POSITION: 50})
self.hass.block_till_done()
hass.states.async_set(entity_id, STATE_OPEN)
await hass.async_block_till_done()
assert acc.char_current_position.value == 100
assert acc.char_target_position.value == 100
assert acc.char_position_state.value == 2
self.assertEqual(acc.char_current_position.value, 50)
self.assertEqual(acc.char_target_position.value, 50)
hass.states.async_set(entity_id, STATE_CLOSED)
await hass.async_block_till_done()
assert acc.char_current_position.value == 0
assert acc.char_target_position.value == 0
assert acc.char_position_state.value == 2
# Set from HomeKit
acc.char_target_position.client_update_value(25)
self.hass.block_till_done()
self.assertEqual(
self.events[0].data[ATTR_SERVICE], 'set_cover_position')
self.assertEqual(
self.events[0].data[ATTR_SERVICE_DATA][ATTR_POSITION], 25)
# Set from HomeKit
call_close_cover = async_mock_service(hass, DOMAIN, 'close_cover')
call_open_cover = async_mock_service(hass, DOMAIN, 'open_cover')
self.assertEqual(acc.char_current_position.value, 50)
self.assertEqual(acc.char_target_position.value, 25)
await hass.async_add_job(acc.char_target_position.client_update_value, 25)
await hass.async_block_till_done()
assert call_close_cover
assert call_close_cover[0].data[ATTR_ENTITY_ID] == entity_id
assert acc.char_current_position.value == 0
assert acc.char_target_position.value == 0
assert acc.char_position_state.value == 2
# Set from HomeKit
acc.char_target_position.client_update_value(75)
self.hass.block_till_done()
self.assertEqual(
self.events[1].data[ATTR_SERVICE], 'set_cover_position')
self.assertEqual(
self.events[1].data[ATTR_SERVICE_DATA][ATTR_POSITION], 75)
await hass.async_add_job(acc.char_target_position.client_update_value, 90)
await hass.async_block_till_done()
assert call_open_cover[0]
assert call_open_cover[0].data[ATTR_ENTITY_ID] == entity_id
assert acc.char_current_position.value == 100
assert acc.char_target_position.value == 100
assert acc.char_position_state.value == 2
self.assertEqual(acc.char_current_position.value, 50)
self.assertEqual(acc.char_target_position.value, 75)
await hass.async_add_job(acc.char_target_position.client_update_value, 55)
await hass.async_block_till_done()
assert call_open_cover[1]
assert call_open_cover[1].data[ATTR_ENTITY_ID] == entity_id
assert acc.char_current_position.value == 100
assert acc.char_target_position.value == 100
assert acc.char_position_state.value == 2
def test_window_open_close(self):
"""Test if accessory and HA are updated accordingly."""
window_cover = 'cover.window'
self.hass.states.set(window_cover, STATE_UNKNOWN,
{ATTR_SUPPORTED_FEATURES: 0})
acc = self.window_basic_cls(self.hass, 'Cover', window_cover, 2,
config=None)
acc.run()
async def test_window_open_close_stop(hass, cls):
"""Test if accessory and HA are updated accordingly."""
entity_id = 'cover.window'
self.assertEqual(acc.aid, 2)
self.assertEqual(acc.category, 14) # WindowCovering
hass.states.async_set(entity_id, STATE_UNKNOWN,
{ATTR_SUPPORTED_FEATURES: SUPPORT_STOP})
acc = cls.window_basic(hass, 'Cover', entity_id, 2, config=None)
await hass.async_add_job(acc.run)
self.assertEqual(acc.char_current_position.value, 0)
self.assertEqual(acc.char_target_position.value, 0)
self.assertEqual(acc.char_position_state.value, 2)
# Set from HomeKit
call_close_cover = async_mock_service(hass, DOMAIN, 'close_cover')
call_open_cover = async_mock_service(hass, DOMAIN, 'open_cover')
call_stop_cover = async_mock_service(hass, DOMAIN, 'stop_cover')
self.hass.states.set(window_cover, STATE_UNKNOWN)
self.hass.block_till_done()
await hass.async_add_job(acc.char_target_position.client_update_value, 25)
await hass.async_block_till_done()
assert call_close_cover
assert call_close_cover[0].data[ATTR_ENTITY_ID] == entity_id
assert acc.char_current_position.value == 0
assert acc.char_target_position.value == 0
assert acc.char_position_state.value == 2
self.assertEqual(acc.char_current_position.value, 0)
self.assertEqual(acc.char_target_position.value, 0)
self.assertEqual(acc.char_position_state.value, 2)
await hass.async_add_job(acc.char_target_position.client_update_value, 90)
await hass.async_block_till_done()
assert call_open_cover
assert call_open_cover[0].data[ATTR_ENTITY_ID] == entity_id
assert acc.char_current_position.value == 100
assert acc.char_target_position.value == 100
assert acc.char_position_state.value == 2
self.hass.states.set(window_cover, STATE_OPEN)
self.hass.block_till_done()
self.assertEqual(acc.char_current_position.value, 100)
self.assertEqual(acc.char_target_position.value, 100)
self.assertEqual(acc.char_position_state.value, 2)
self.hass.states.set(window_cover, STATE_CLOSED)
self.hass.block_till_done()
self.assertEqual(acc.char_current_position.value, 0)
self.assertEqual(acc.char_target_position.value, 0)
self.assertEqual(acc.char_position_state.value, 2)
# Set from HomeKit
acc.char_target_position.client_update_value(25)
self.hass.block_till_done()
self.assertEqual(
self.events[0].data[ATTR_SERVICE], 'close_cover')
self.assertEqual(acc.char_current_position.value, 0)
self.assertEqual(acc.char_target_position.value, 0)
self.assertEqual(acc.char_position_state.value, 2)
# Set from HomeKit
acc.char_target_position.client_update_value(90)
self.hass.block_till_done()
self.assertEqual(
self.events[1].data[ATTR_SERVICE], 'open_cover')
self.assertEqual(acc.char_current_position.value, 100)
self.assertEqual(acc.char_target_position.value, 100)
self.assertEqual(acc.char_position_state.value, 2)
# Set from HomeKit
acc.char_target_position.client_update_value(55)
self.hass.block_till_done()
self.assertEqual(
self.events[2].data[ATTR_SERVICE], 'open_cover')
self.assertEqual(acc.char_current_position.value, 100)
self.assertEqual(acc.char_target_position.value, 100)
self.assertEqual(acc.char_position_state.value, 2)
def test_window_open_close_stop(self):
"""Test if accessory and HA are updated accordingly."""
window_cover = 'cover.window'
self.hass.states.set(window_cover, STATE_UNKNOWN,
{ATTR_SUPPORTED_FEATURES: SUPPORT_STOP})
acc = self.window_basic_cls(self.hass, 'Cover', window_cover, 2,
config=None)
acc.run()
# Set from HomeKit
acc.char_target_position.client_update_value(25)
self.hass.block_till_done()
self.assertEqual(
self.events[0].data[ATTR_SERVICE], 'close_cover')
self.assertEqual(acc.char_current_position.value, 0)
self.assertEqual(acc.char_target_position.value, 0)
self.assertEqual(acc.char_position_state.value, 2)
# Set from HomeKit
acc.char_target_position.client_update_value(90)
self.hass.block_till_done()
self.assertEqual(
self.events[1].data[ATTR_SERVICE], 'open_cover')
self.assertEqual(acc.char_current_position.value, 100)
self.assertEqual(acc.char_target_position.value, 100)
self.assertEqual(acc.char_position_state.value, 2)
# Set from HomeKit
acc.char_target_position.client_update_value(55)
self.hass.block_till_done()
self.assertEqual(
self.events[2].data[ATTR_SERVICE], 'stop_cover')
self.assertEqual(acc.char_current_position.value, 50)
self.assertEqual(acc.char_target_position.value, 50)
self.assertEqual(acc.char_position_state.value, 2)
await hass.async_add_job(acc.char_target_position.client_update_value, 55)
await hass.async_block_till_done()
assert call_stop_cover
assert call_stop_cover[0].data[ATTR_ENTITY_ID] == entity_id
assert acc.char_current_position.value == 50
assert acc.char_target_position.value == 50
assert acc.char_position_state.value == 2

280
tests/components/homekit/test_type_lights.py
View File

@ -1,188 +1,174 @@
"""Test different accessory types: Lights."""
import unittest
from collections import namedtuple
import pytest
from homeassistant.core import callback
from homeassistant.components.light import (
DOMAIN, ATTR_BRIGHTNESS, ATTR_BRIGHTNESS_PCT, ATTR_COLOR_TEMP,
ATTR_HS_COLOR, SUPPORT_BRIGHTNESS, SUPPORT_COLOR_TEMP, SUPPORT_COLOR)
from homeassistant.const import (
ATTR_DOMAIN, ATTR_ENTITY_ID, ATTR_SERVICE, ATTR_SERVICE_DATA,
ATTR_SUPPORTED_FEATURES, EVENT_CALL_SERVICE, SERVICE_TURN_ON,
SERVICE_TURN_OFF, STATE_ON, STATE_OFF, STATE_UNKNOWN)
ATTR_ENTITY_ID, ATTR_SUPPORTED_FEATURES,
STATE_ON, STATE_OFF, STATE_UNKNOWN)
from tests.common import get_test_home_assistant
from tests.common import async_mock_service
from tests.components.homekit.test_accessories import patch_debounce
class TestHomekitLights(unittest.TestCase):
"""Test class for all accessory types regarding lights."""
@pytest.fixture(scope='module')
def cls(request):
"""Patch debounce decorator during import of type_lights."""
patcher = patch_debounce()
patcher.start()
_import = __import__('homeassistant.components.homekit.type_lights',
fromlist=['Light'])
request.addfinalizer(patcher.stop)
patcher_tuple = namedtuple('Cls', ['light'])
return patcher_tuple(light=_import.Light)
@classmethod
def setUpClass(cls):
"""Setup Light class import and debounce patcher."""
cls.patcher = patch_debounce()
cls.patcher.start()
_import = __import__('homeassistant.components.homekit.type_lights',
fromlist=['Light'])
cls.light_cls = _import.Light
@classmethod
def tearDownClass(cls):
"""Stop debounce patcher."""
cls.patcher.stop()
async def test_light_basic(hass, cls):
"""Test light with char state."""
entity_id = 'light.demo'
def setUp(self):
"""Setup things to be run when tests are started."""
self.hass = get_test_home_assistant()
self.events = []
hass.states.async_set(entity_id, STATE_ON,
{ATTR_SUPPORTED_FEATURES: 0})
await hass.async_block_till_done()
acc = cls.light(hass, 'Light', entity_id, 2, config=None)
@callback
def record_event(event):
"""Track called event."""
self.events.append(event)
assert acc.aid == 2
assert acc.category == 5 # Lightbulb
assert acc.char_on.value == 0
self.hass.bus.listen(EVENT_CALL_SERVICE, record_event)
await hass.async_add_job(acc.run)
await hass.async_block_till_done()
assert acc.char_on.value == 1
def tearDown(self):
"""Stop down everything that was started."""
self.hass.stop()
hass.states.async_set(entity_id, STATE_OFF,
{ATTR_SUPPORTED_FEATURES: 0})
await hass.async_block_till_done()
assert acc.char_on.value == 0
def test_light_basic(self):
"""Test light with char state."""
entity_id = 'light.demo'
hass.states.async_set(entity_id, STATE_UNKNOWN)
await hass.async_block_till_done()
assert acc.char_on.value == 0
self.hass.states.set(entity_id, STATE_ON,
{ATTR_SUPPORTED_FEATURES: 0})
self.hass.block_till_done()
acc = self.light_cls(self.hass, 'Light', entity_id, 2, config=None)
self.assertEqual(acc.aid, 2)
self.assertEqual(acc.category, 5) # Lightbulb
self.assertEqual(acc.char_on.value, 0)
hass.states.async_remove(entity_id)
await hass.async_block_till_done()
assert acc.char_on.value == 0
acc.run()
self.hass.block_till_done()
self.assertEqual(acc.char_on.value, 1)
# Set from HomeKit
call_turn_on = async_mock_service(hass, DOMAIN, 'turn_on')
call_turn_off = async_mock_service(hass, DOMAIN, 'turn_off')
self.hass.states.set(entity_id, STATE_OFF,
{ATTR_SUPPORTED_FEATURES: 0})
self.hass.block_till_done()
self.assertEqual(acc.char_on.value, 0)
await hass.async_add_job(acc.char_on.client_update_value, 1)
await hass.async_block_till_done()
assert call_turn_on
assert call_turn_on[0].data[ATTR_ENTITY_ID] == entity_id
self.hass.states.set(entity_id, STATE_UNKNOWN)
self.hass.block_till_done()
self.assertEqual(acc.char_on.value, 0)
hass.states.async_set(entity_id, STATE_ON)
await hass.async_block_till_done()
# Set from HomeKit
acc.char_on.client_update_value(1)
self.hass.block_till_done()
self.assertEqual(self.events[0].data[ATTR_DOMAIN], DOMAIN)
self.assertEqual(self.events[0].data[ATTR_SERVICE], SERVICE_TURN_ON)
await hass.async_add_job(acc.char_on.client_update_value, 0)
await hass.async_block_till_done()
assert call_turn_off
assert call_turn_off[0].data[ATTR_ENTITY_ID] == entity_id
self.hass.states.set(entity_id, STATE_ON)
self.hass.block_till_done()
acc.char_on.client_update_value(0)
self.hass.block_till_done()
self.assertEqual(self.events[1].data[ATTR_DOMAIN], DOMAIN)
self.assertEqual(self.events[1].data[ATTR_SERVICE], SERVICE_TURN_OFF)
async def test_light_brightness(hass, cls):
"""Test light with brightness."""
entity_id = 'light.demo'
self.hass.states.set(entity_id, STATE_OFF)
self.hass.block_till_done()
hass.states.async_set(entity_id, STATE_ON, {
ATTR_SUPPORTED_FEATURES: SUPPORT_BRIGHTNESS, ATTR_BRIGHTNESS: 255})
await hass.async_block_till_done()
acc = cls.light(hass, 'Light', entity_id, 2, config=None)
# Remove entity
self.hass.states.remove(entity_id)
self.hass.block_till_done()
assert acc.char_brightness.value == 0
def test_light_brightness(self):
"""Test light with brightness."""
entity_id = 'light.demo'
await hass.async_add_job(acc.run)
await hass.async_block_till_done()
assert acc.char_brightness.value == 100
self.hass.states.set(entity_id, STATE_ON, {
ATTR_SUPPORTED_FEATURES: SUPPORT_BRIGHTNESS, ATTR_BRIGHTNESS: 255})
self.hass.block_till_done()
acc = self.light_cls(self.hass, 'Light', entity_id, 2, config=None)
self.assertEqual(acc.char_brightness.value, 0)
hass.states.async_set(entity_id, STATE_ON, {ATTR_BRIGHTNESS: 102})
await hass.async_block_till_done()
assert acc.char_brightness.value == 40
acc.run()
self.hass.block_till_done()
self.assertEqual(acc.char_brightness.value, 100)
# Set from HomeKit
call_turn_on = async_mock_service(hass, DOMAIN, 'turn_on')
call_turn_off = async_mock_service(hass, DOMAIN, 'turn_off')
self.hass.states.set(entity_id, STATE_ON, {ATTR_BRIGHTNESS: 102})
self.hass.block_till_done()
self.assertEqual(acc.char_brightness.value, 40)
await hass.async_add_job(acc.char_brightness.client_update_value, 20)
await hass.async_add_job(acc.char_on.client_update_value, 1)
await hass.async_block_till_done()
assert call_turn_on[0]
assert call_turn_on[0].data[ATTR_ENTITY_ID] == entity_id
assert call_turn_on[0].data[ATTR_BRIGHTNESS_PCT] == 20
# Set from HomeKit
acc.char_brightness.client_update_value(20)
acc.char_on.client_update_value(1)
self.hass.block_till_done()
self.assertEqual(self.events[0].data[ATTR_DOMAIN], DOMAIN)
self.assertEqual(self.events[0].data[ATTR_SERVICE], SERVICE_TURN_ON)
self.assertEqual(
self.events[0].data[ATTR_SERVICE_DATA], {
ATTR_ENTITY_ID: entity_id, ATTR_BRIGHTNESS_PCT: 20})
await hass.async_add_job(acc.char_on.client_update_value, 1)
await hass.async_add_job(acc.char_brightness.client_update_value, 40)
await hass.async_block_till_done()
assert call_turn_on[1]
assert call_turn_on[1].data[ATTR_ENTITY_ID] == entity_id
assert call_turn_on[1].data[ATTR_BRIGHTNESS_PCT] == 40
acc.char_on.client_update_value(1)
acc.char_brightness.client_update_value(40)
self.hass.block_till_done()
self.assertEqual(self.events[1].data[ATTR_DOMAIN], DOMAIN)
self.assertEqual(self.events[1].data[ATTR_SERVICE], SERVICE_TURN_ON)
self.assertEqual(
self.events[1].data[ATTR_SERVICE_DATA], {
ATTR_ENTITY_ID: entity_id, ATTR_BRIGHTNESS_PCT: 40})
await hass.async_add_job(acc.char_on.client_update_value, 1)
await hass.async_add_job(acc.char_brightness.client_update_value, 0)
await hass.async_block_till_done()
assert call_turn_off
assert call_turn_off[0].data[ATTR_ENTITY_ID] == entity_id
acc.char_on.client_update_value(1)
acc.char_brightness.client_update_value(0)
self.hass.block_till_done()
self.assertEqual(self.events[2].data[ATTR_DOMAIN], DOMAIN)
self.assertEqual(self.events[2].data[ATTR_SERVICE], SERVICE_TURN_OFF)
def test_light_color_temperature(self):
"""Test light with color temperature."""
entity_id = 'light.demo'
async def test_light_color_temperature(hass, cls):
"""Test light with color temperature."""
entity_id = 'light.demo'
self.hass.states.set(entity_id, STATE_ON, {
ATTR_SUPPORTED_FEATURES: SUPPORT_COLOR_TEMP,
ATTR_COLOR_TEMP: 190})
self.hass.block_till_done()
acc = self.light_cls(self.hass, 'Light', entity_id, 2, config=None)
self.assertEqual(acc.char_color_temperature.value, 153)
hass.states.async_set(entity_id, STATE_ON, {
ATTR_SUPPORTED_FEATURES: SUPPORT_COLOR_TEMP,
ATTR_COLOR_TEMP: 190})
await hass.async_block_till_done()
acc = cls.light(hass, 'Light', entity_id, 2, config=None)
acc.run()
self.hass.block_till_done()
self.assertEqual(acc.char_color_temperature.value, 190)
assert acc.char_color_temperature.value == 153
# Set from HomeKit
acc.char_color_temperature.client_update_value(250)
self.hass.block_till_done()
self.assertEqual(self.events[0].data[ATTR_DOMAIN], DOMAIN)
self.assertEqual(self.events[0].data[ATTR_SERVICE], SERVICE_TURN_ON)
self.assertEqual(
self.events[0].data[ATTR_SERVICE_DATA], {
ATTR_ENTITY_ID: entity_id, ATTR_COLOR_TEMP: 250})
await hass.async_add_job(acc.run)
await hass.async_block_till_done()
assert acc.char_color_temperature.value == 190
def test_light_rgb_color(self):
"""Test light with rgb_color."""
entity_id = 'light.demo'
# Set from HomeKit
call_turn_on = async_mock_service(hass, DOMAIN, 'turn_on')
self.hass.states.set(entity_id, STATE_ON, {
ATTR_SUPPORTED_FEATURES: SUPPORT_COLOR,
ATTR_HS_COLOR: (260, 90)})
self.hass.block_till_done()
acc = self.light_cls(self.hass, 'Light', entity_id, 2, config=None)
self.assertEqual(acc.char_hue.value, 0)
self.assertEqual(acc.char_saturation.value, 75)
await hass.async_add_job(
acc.char_color_temperature.client_update_value, 250)
await hass.async_block_till_done()
assert call_turn_on
assert call_turn_on[0].data[ATTR_ENTITY_ID] == entity_id
assert call_turn_on[0].data[ATTR_COLOR_TEMP] == 250
acc.run()
self.hass.block_till_done()
self.assertEqual(acc.char_hue.value, 260)
self.assertEqual(acc.char_saturation.value, 90)
# Set from HomeKit
acc.char_hue.client_update_value(145)
acc.char_saturation.client_update_value(75)
self.hass.block_till_done()
self.assertEqual(self.events[0].data[ATTR_DOMAIN], DOMAIN)
self.assertEqual(self.events[0].data[ATTR_SERVICE], SERVICE_TURN_ON)
self.assertEqual(
self.events[0].data[ATTR_SERVICE_DATA], {
ATTR_ENTITY_ID: entity_id, ATTR_HS_COLOR: (145, 75)})
async def test_light_rgb_color(hass, cls):
"""Test light with rgb_color."""
entity_id = 'light.demo'
hass.states.async_set(entity_id, STATE_ON, {
ATTR_SUPPORTED_FEATURES: SUPPORT_COLOR,
ATTR_HS_COLOR: (260, 90)})
await hass.async_block_till_done()
acc = cls.light(hass, 'Light', entity_id, 2, config=None)
assert acc.char_hue.value == 0
assert acc.char_saturation.value == 75
await hass.async_add_job(acc.run)
await hass.async_block_till_done()
assert acc.char_hue.value == 260
assert acc.char_saturation.value == 90
# Set from HomeKit
call_turn_on = async_mock_service(hass, DOMAIN, 'turn_on')
await hass.async_add_job(acc.char_hue.client_update_value, 145)
await hass.async_add_job(acc.char_saturation.client_update_value, 75)
await hass.async_block_till_done()
assert call_turn_on
assert call_turn_on[0].data[ATTR_ENTITY_ID] == entity_id
assert call_turn_on[0].data[ATTR_HS_COLOR] == (145, 75)

102
tests/components/homekit/test_type_locks.py
View File

@ -1,77 +1,57 @@
"""Test different accessory types: Locks."""
import unittest
from homeassistant.core import callback
from homeassistant.components.homekit.type_locks import Lock
from homeassistant.components.lock import DOMAIN
from homeassistant.const import (
STATE_UNKNOWN, STATE_UNLOCKED, STATE_LOCKED,
ATTR_SERVICE, EVENT_CALL_SERVICE)
ATTR_ENTITY_ID, STATE_UNKNOWN, STATE_UNLOCKED, STATE_LOCKED)
from tests.common import get_test_home_assistant
from tests.common import async_mock_service
class TestHomekitSensors(unittest.TestCase):
"""Test class for all accessory types regarding covers."""
async def test_lock_unlock(hass):
"""Test if accessory and HA are updated accordingly."""
entity_id = 'lock.kitchen_door'
def setUp(self):
"""Setup things to be run when tests are started."""
self.hass = get_test_home_assistant()
self.events = []
acc = Lock(hass, 'Lock', entity_id, 2, config=None)
await hass.async_add_job(acc.run)
@callback
def record_event(event):
"""Track called event."""
self.events.append(event)
assert acc.aid == 2
assert acc.category == 6 # DoorLock
self.hass.bus.listen(EVENT_CALL_SERVICE, record_event)
assert acc.char_current_state.value == 3
assert acc.char_target_state.value == 1
def tearDown(self):
"""Stop down everything that was started."""
self.hass.stop()
hass.states.async_set(entity_id, STATE_LOCKED)
await hass.async_block_till_done()
assert acc.char_current_state.value == 1
assert acc.char_target_state.value == 1
def test_lock_unlock(self):
"""Test if accessory and HA are updated accordingly."""
kitchen_lock = 'lock.kitchen_door'
hass.states.async_set(entity_id, STATE_UNLOCKED)
await hass.async_block_till_done()
assert acc.char_current_state.value == 0
assert acc.char_target_state.value == 0
acc = Lock(self.hass, 'Lock', kitchen_lock, 2, config=None)
acc.run()
hass.states.async_set(entity_id, STATE_UNKNOWN)
await hass.async_block_till_done()
assert acc.char_current_state.value == 3
assert acc.char_target_state.value == 0
self.assertEqual(acc.aid, 2)
self.assertEqual(acc.category, 6) # DoorLock
hass.states.async_remove(entity_id)
await hass.async_block_till_done()
assert acc.char_current_state.value == 3
assert acc.char_target_state.value == 0
self.assertEqual(acc.char_current_state.value, 3)
self.assertEqual(acc.char_target_state.value, 1)
# Set from HomeKit
call_lock = async_mock_service(hass, DOMAIN, 'lock')
call_unlock = async_mock_service(hass, DOMAIN, 'unlock')
self.hass.states.set(kitchen_lock, STATE_LOCKED)
self.hass.block_till_done()
await hass.async_add_job(acc.char_target_state.client_update_value, 1)
await hass.async_block_till_done()
assert call_lock
assert call_lock[0].data[ATTR_ENTITY_ID] == entity_id
assert acc.char_target_state.value == 1
self.assertEqual(acc.char_current_state.value, 1)
self.assertEqual(acc.char_target_state.value, 1)
self.hass.states.set(kitchen_lock, STATE_UNLOCKED)
self.hass.block_till_done()
self.assertEqual(acc.char_current_state.value, 0)
self.assertEqual(acc.char_target_state.value, 0)
self.hass.states.set(kitchen_lock, STATE_UNKNOWN)
self.hass.block_till_done()
self.assertEqual(acc.char_current_state.value, 3)
self.assertEqual(acc.char_target_state.value, 0)
# Set from HomeKit
acc.char_target_state.client_update_value(1)
self.hass.block_till_done()
self.assertEqual(
self.events[0].data[ATTR_SERVICE], 'lock')
self.assertEqual(acc.char_target_state.value, 1)
acc.char_target_state.client_update_value(0)
self.hass.block_till_done()
self.assertEqual(
self.events[1].data[ATTR_SERVICE], 'unlock')
self.assertEqual(acc.char_target_state.value, 0)
self.hass.states.remove(kitchen_lock)
self.hass.block_till_done()
await hass.async_add_job(acc.char_target_state.client_update_value, 0)
await hass.async_block_till_done()
assert call_unlock
assert call_unlock[0].data[ATTR_ENTITY_ID] == entity_id
assert acc.char_target_state.value == 0

188
tests/components/homekit/test_type_security_systems.py
View File

@ -1,134 +1,110 @@
"""Test different accessory types: Security Systems."""
import unittest
import pytest
from homeassistant.core import callback
from homeassistant.components.alarm_control_panel import DOMAIN
from homeassistant.components.homekit.type_security_systems import (
SecuritySystem)
from homeassistant.const import (
ATTR_CODE, ATTR_SERVICE, ATTR_SERVICE_DATA, EVENT_CALL_SERVICE,
ATTR_CODE, ATTR_ENTITY_ID, STATE_UNKNOWN,
STATE_ALARM_ARMED_AWAY, STATE_ALARM_ARMED_HOME,
STATE_ALARM_ARMED_NIGHT, STATE_ALARM_DISARMED, STATE_ALARM_TRIGGERED,
STATE_UNKNOWN)
STATE_ALARM_ARMED_NIGHT, STATE_ALARM_DISARMED, STATE_ALARM_TRIGGERED)
from tests.common import get_test_home_assistant
from tests.common import async_mock_service
class TestHomekitSecuritySystems(unittest.TestCase):
"""Test class for all accessory types regarding security systems."""
async def test_switch_set_state(hass):
"""Test if accessory and HA are updated accordingly."""
code = '1234'
config = {ATTR_CODE: code}
entity_id = 'alarm_control_panel.test'
def setUp(self):
"""Setup things to be run when tests are started."""
self.hass = get_test_home_assistant()
self.events = []
acc = SecuritySystem(hass, 'SecuritySystem', entity_id, 2, config=config)
await hass.async_add_job(acc.run)
@callback
def record_event(event):
"""Track called event."""
self.events.append(event)
assert acc.aid == 2
assert acc.category == 11 # AlarmSystem
self.hass.bus.listen(EVENT_CALL_SERVICE, record_event)
assert acc.char_current_state.value == 3
assert acc.char_target_state.value == 3
def tearDown(self):
"""Stop down everything that was started."""
self.hass.stop()
hass.states.async_set(entity_id, STATE_ALARM_ARMED_AWAY)
await hass.async_block_till_done()
assert acc.char_target_state.value == 1
assert acc.char_current_state.value == 1
def test_switch_set_state(self):
"""Test if accessory and HA are updated accordingly."""
acp = 'alarm_control_panel.test'
hass.states.async_set(entity_id, STATE_ALARM_ARMED_HOME)
await hass.async_block_till_done()
assert acc.char_target_state.value == 0
assert acc.char_current_state.value == 0
acc = SecuritySystem(self.hass, 'SecuritySystem', acp,
2, config={ATTR_CODE: '1234'})
acc.run()
hass.states.async_set(entity_id, STATE_ALARM_ARMED_NIGHT)
await hass.async_block_till_done()
assert acc.char_target_state.value == 2
assert acc.char_current_state.value == 2
self.assertEqual(acc.aid, 2)
self.assertEqual(acc.category, 11) # AlarmSystem
hass.states.async_set(entity_id, STATE_ALARM_DISARMED)
await hass.async_block_till_done()
assert acc.char_target_state.value == 3
assert acc.char_current_state.value == 3
self.assertEqual(acc.char_current_state.value, 3)
self.assertEqual(acc.char_target_state.value, 3)
hass.states.async_set(entity_id, STATE_ALARM_TRIGGERED)
await hass.async_block_till_done()
assert acc.char_target_state.value == 3
assert acc.char_current_state.value == 4
self.hass.states.set(acp, STATE_ALARM_ARMED_AWAY)
self.hass.block_till_done()
self.assertEqual(acc.char_target_state.value, 1)
self.assertEqual(acc.char_current_state.value, 1)
hass.states.async_set(entity_id, STATE_UNKNOWN)
await hass.async_block_till_done()
assert acc.char_target_state.value == 3
assert acc.char_current_state.value == 4
self.hass.states.set(acp, STATE_ALARM_ARMED_HOME)
self.hass.block_till_done()
self.assertEqual(acc.char_target_state.value, 0)
self.assertEqual(acc.char_current_state.value, 0)
# Set from HomeKit
call_arm_home = async_mock_service(hass, DOMAIN, 'alarm_arm_home')
call_arm_away = async_mock_service(hass, DOMAIN, 'alarm_arm_away')
call_arm_night = async_mock_service(hass, DOMAIN, 'alarm_arm_night')
call_disarm = async_mock_service(hass, DOMAIN, 'alarm_disarm')
self.hass.states.set(acp, STATE_ALARM_ARMED_NIGHT)
self.hass.block_till_done()
self.assertEqual(acc.char_target_state.value, 2)
self.assertEqual(acc.char_current_state.value, 2)
await hass.async_add_job(acc.char_target_state.client_update_value, 0)
await hass.async_block_till_done()
assert call_arm_home
assert call_arm_home[0].data[ATTR_ENTITY_ID] == entity_id
assert call_arm_home[0].data[ATTR_CODE] == code
assert acc.char_target_state.value == 0
self.hass.states.set(acp, STATE_ALARM_DISARMED)
self.hass.block_till_done()
self.assertEqual(acc.char_target_state.value, 3)
self.assertEqual(acc.char_current_state.value, 3)
await hass.async_add_job(acc.char_target_state.client_update_value, 1)
await hass.async_block_till_done()
assert call_arm_away
assert call_arm_away[0].data[ATTR_ENTITY_ID] == entity_id
assert call_arm_away[0].data[ATTR_CODE] == code
assert acc.char_target_state.value == 1
self.hass.states.set(acp, STATE_ALARM_TRIGGERED)
self.hass.block_till_done()
self.assertEqual(acc.char_target_state.value, 3)
self.assertEqual(acc.char_current_state.value, 4)
await hass.async_add_job(acc.char_target_state.client_update_value, 2)
await hass.async_block_till_done()
assert call_arm_night
assert call_arm_night[0].data[ATTR_ENTITY_ID] == entity_id
assert call_arm_night[0].data[ATTR_CODE] == code
assert acc.char_target_state.value == 2
self.hass.states.set(acp, STATE_UNKNOWN)
self.hass.block_till_done()
self.assertEqual(acc.char_target_state.value, 3)
self.assertEqual(acc.char_current_state.value, 4)
await hass.async_add_job(acc.char_target_state.client_update_value, 3)
await hass.async_block_till_done()
assert call_disarm
assert call_disarm[0].data[ATTR_ENTITY_ID] == entity_id
assert call_disarm[0].data[ATTR_CODE] == code
assert acc.char_target_state.value == 3
# Set from HomeKit
acc.char_target_state.client_update_value(0)
self.hass.block_till_done()
self.assertEqual(
self.events[0].data[ATTR_SERVICE], 'alarm_arm_home')
self.assertEqual(
self.events[0].data[ATTR_SERVICE_DATA][ATTR_CODE], '1234')
self.assertEqual(acc.char_target_state.value, 0)
acc.char_target_state.client_update_value(1)
self.hass.block_till_done()
self.assertEqual(
self.events[1].data[ATTR_SERVICE], 'alarm_arm_away')
self.assertEqual(
self.events[0].data[ATTR_SERVICE_DATA][ATTR_CODE], '1234')
self.assertEqual(acc.char_target_state.value, 1)
@pytest.mark.parametrize('config', [{}, {ATTR_CODE: None}])
async def test_no_alarm_code(hass, config):
"""Test accessory if security_system doesn't require a alarm_code."""
entity_id = 'alarm_control_panel.test'
acc.char_target_state.client_update_value(2)
self.hass.block_till_done()
self.assertEqual(
self.events[2].data[ATTR_SERVICE], 'alarm_arm_night')
self.assertEqual(
self.events[0].data[ATTR_SERVICE_DATA][ATTR_CODE], '1234')
self.assertEqual(acc.char_target_state.value, 2)
acc = SecuritySystem(hass, 'SecuritySystem', entity_id, 2, config=config)
acc.char_target_state.client_update_value(3)
self.hass.block_till_done()
self.assertEqual(
self.events[3].data[ATTR_SERVICE], 'alarm_disarm')
self.assertEqual(
self.events[0].data[ATTR_SERVICE_DATA][ATTR_CODE], '1234')
self.assertEqual(acc.char_target_state.value, 3)
# Set from HomeKit
call_arm_home = async_mock_service(hass, DOMAIN, 'alarm_arm_home')
def test_no_alarm_code(self):
"""Test accessory if security_system doesn't require a alarm_code."""
acp = 'alarm_control_panel.test'
acc = SecuritySystem(self.hass, 'SecuritySystem', acp,
2, config={ATTR_CODE: None})
# Set from HomeKit
acc.char_target_state.client_update_value(0)
self.hass.block_till_done()
self.assertEqual(
self.events[0].data[ATTR_SERVICE], 'alarm_arm_home')
self.assertNotIn(ATTR_CODE, self.events[0].data[ATTR_SERVICE_DATA])
self.assertEqual(acc.char_target_state.value, 0)
acc = SecuritySystem(self.hass, 'SecuritySystem', acp,
2, config={})
# Set from HomeKit
acc.char_target_state.client_update_value(0)
self.hass.block_till_done()
self.assertEqual(
self.events[0].data[ATTR_SERVICE], 'alarm_arm_home')
self.assertNotIn(ATTR_CODE, self.events[0].data[ATTR_SERVICE_DATA])
self.assertEqual(acc.char_target_state.value, 0)
await hass.async_add_job(acc.char_target_state.client_update_value, 0)
await hass.async_block_till_done()
assert call_arm_home
assert call_arm_home[0].data[ATTR_ENTITY_ID] == entity_id
assert ATTR_CODE not in call_arm_home[0].data
assert acc.char_target_state.value == 0

298
tests/components/homekit/test_type_sensors.py
View File

@ -1,6 +1,4 @@
"""Test different accessory types: Sensors."""
import unittest
from homeassistant.components.homekit.const import PROP_CELSIUS
from homeassistant.components.homekit.type_sensors import (
TemperatureSensor, HumiditySensor, AirQualitySensor, CarbonDioxideSensor,
@ -9,201 +7,191 @@ from homeassistant.const import (
ATTR_UNIT_OF_MEASUREMENT, ATTR_DEVICE_CLASS, STATE_UNKNOWN, STATE_ON,
STATE_OFF, STATE_HOME, STATE_NOT_HOME, TEMP_CELSIUS, TEMP_FAHRENHEIT)
from tests.common import get_test_home_assistant
async def test_temperature(hass):
"""Test if accessory is updated after state change."""
entity_id = 'sensor.temperature'
acc = TemperatureSensor(hass, 'Temperature', entity_id, 2, config=None)
await hass.async_add_job(acc.run)
assert acc.aid == 2
assert acc.category == 10 # Sensor
assert acc.char_temp.value == 0.0
for key, value in PROP_CELSIUS.items():
assert acc.char_temp.properties[key] == value
hass.states.async_set(entity_id, STATE_UNKNOWN,
{ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS})
await hass.async_block_till_done()
assert acc.char_temp.value == 0.0
hass.states.async_set(entity_id, '20',
{ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS})
await hass.async_block_till_done()
assert acc.char_temp.value == 20
hass.states.async_set(entity_id, '75.2',
{ATTR_UNIT_OF_MEASUREMENT: TEMP_FAHRENHEIT})
await hass.async_block_till_done()
assert acc.char_temp.value == 24
class TestHomekitSensors(unittest.TestCase):
"""Test class for all accessory types regarding sensors."""
async def test_humidity(hass):
"""Test if accessory is updated after state change."""
entity_id = 'sensor.humidity'
def setUp(self):
"""Setup things to be run when tests are started."""
self.hass = get_test_home_assistant()
acc = HumiditySensor(hass, 'Humidity', entity_id, 2, config=None)
await hass.async_add_job(acc.run)
def tearDown(self):
"""Stop down everything that was started."""
self.hass.stop()
assert acc.aid == 2
assert acc.category == 10 # Sensor
def test_temperature(self):
"""Test if accessory is updated after state change."""
entity_id = 'sensor.temperature'
assert acc.char_humidity.value == 0
acc = TemperatureSensor(self.hass, 'Temperature', entity_id,
2, config=None)
acc.run()
hass.states.async_set(entity_id, STATE_UNKNOWN)
await hass.async_block_till_done()
assert acc.char_humidity.value == 0
self.assertEqual(acc.aid, 2)
self.assertEqual(acc.category, 10) # Sensor
hass.states.async_set(entity_id, '20')
await hass.async_block_till_done()
assert acc.char_humidity.value == 20
self.assertEqual(acc.char_temp.value, 0.0)
for key, value in PROP_CELSIUS.items():
self.assertEqual(acc.char_temp.properties[key], value)
self.hass.states.set(entity_id, STATE_UNKNOWN,
{ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS})
self.hass.block_till_done()
self.assertEqual(acc.char_temp.value, 0.0)
async def test_air_quality(hass):
"""Test if accessory is updated after state change."""
entity_id = 'sensor.air_quality'
self.hass.states.set(entity_id, '20',
{ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS})
self.hass.block_till_done()
self.assertEqual(acc.char_temp.value, 20)
acc = AirQualitySensor(hass, 'Air Quality', entity_id, 2, config=None)
await hass.async_add_job(acc.run)
self.hass.states.set(entity_id, '75.2',
{ATTR_UNIT_OF_MEASUREMENT: TEMP_FAHRENHEIT})
self.hass.block_till_done()
self.assertEqual(acc.char_temp.value, 24)
assert acc.aid == 2
assert acc.category == 10 # Sensor
def test_humidity(self):
"""Test if accessory is updated after state change."""
entity_id = 'sensor.humidity'
assert acc.char_density.value == 0
assert acc.char_quality.value == 0
acc = HumiditySensor(self.hass, 'Humidity', entity_id, 2, config=None)
acc.run()
hass.states.async_set(entity_id, STATE_UNKNOWN)
await hass.async_block_till_done()
assert acc.char_density.value == 0
assert acc.char_quality.value == 0
self.assertEqual(acc.aid, 2)
self.assertEqual(acc.category, 10) # Sensor
hass.states.async_set(entity_id, '34')
await hass.async_block_till_done()
assert acc.char_density.value == 34
assert acc.char_quality.value == 1
self.assertEqual(acc.char_humidity.value, 0)
hass.states.async_set(entity_id, '200')
await hass.async_block_till_done()
assert acc.char_density.value == 200
assert acc.char_quality.value == 5
self.hass.states.set(entity_id, STATE_UNKNOWN)
self.hass.block_till_done()
self.assertEqual(acc.char_humidity.value, 0)
self.hass.states.set(entity_id, '20')
self.hass.block_till_done()
self.assertEqual(acc.char_humidity.value, 20)
async def test_co2(hass):
"""Test if accessory is updated after state change."""
entity_id = 'sensor.co2'
def test_air_quality(self):
"""Test if accessory is updated after state change."""
entity_id = 'sensor.air_quality'
acc = CarbonDioxideSensor(hass, 'CO2', entity_id, 2, config=None)
await hass.async_add_job(acc.run)
acc = AirQualitySensor(self.hass, 'Air Quality', entity_id,
2, config=None)
acc.run()
assert acc.aid == 2
assert acc.category == 10 # Sensor
self.assertEqual(acc.aid, 2)
self.assertEqual(acc.category, 10) # Sensor
assert acc.char_co2.value == 0
assert acc.char_peak.value == 0
assert acc.char_detected.value == 0
self.assertEqual(acc.char_density.value, 0)
self.assertEqual(acc.char_quality.value, 0)
hass.states.async_set(entity_id, STATE_UNKNOWN)
await hass.async_block_till_done()
assert acc.char_co2.value == 0
assert acc.char_peak.value == 0
assert acc.char_detected.value == 0
self.hass.states.set(entity_id, STATE_UNKNOWN)
self.hass.block_till_done()
self.assertEqual(acc.char_density.value, 0)
self.assertEqual(acc.char_quality.value, 0)
hass.states.async_set(entity_id, '1100')
await hass.async_block_till_done()
assert acc.char_co2.value == 1100
assert acc.char_peak.value == 1100
assert acc.char_detected.value == 1
self.hass.states.set(entity_id, '34')
self.hass.block_till_done()
self.assertEqual(acc.char_density.value, 34)
self.assertEqual(acc.char_quality.value, 1)
hass.states.async_set(entity_id, '800')
await hass.async_block_till_done()
assert acc.char_co2.value == 800
assert acc.char_peak.value == 1100
assert acc.char_detected.value == 0
self.hass.states.set(entity_id, '200')
self.hass.block_till_done()
self.assertEqual(acc.char_density.value, 200)
self.assertEqual(acc.char_quality.value, 5)
def test_co2(self):
"""Test if accessory is updated after state change."""
entity_id = 'sensor.co2'
async def test_light(hass):
"""Test if accessory is updated after state change."""
entity_id = 'sensor.light'
acc = CarbonDioxideSensor(self.hass, 'CO2', entity_id, 2, config=None)
acc.run()
acc = LightSensor(hass, 'Light', entity_id, 2, config=None)
await hass.async_add_job(acc.run)
self.assertEqual(acc.aid, 2)
self.assertEqual(acc.category, 10) # Sensor
assert acc.aid == 2
assert acc.category == 10 # Sensor
self.assertEqual(acc.char_co2.value, 0)
self.assertEqual(acc.char_peak.value, 0)
self.assertEqual(acc.char_detected.value, 0)
assert acc.char_light.value == 0.0001
self.hass.states.set(entity_id, STATE_UNKNOWN)
self.hass.block_till_done()
self.assertEqual(acc.char_co2.value, 0)
self.assertEqual(acc.char_peak.value, 0)
self.assertEqual(acc.char_detected.value, 0)
hass.states.async_set(entity_id, STATE_UNKNOWN)
await hass.async_block_till_done()
assert acc.char_light.value == 0.0001
self.hass.states.set(entity_id, '1100')
self.hass.block_till_done()
self.assertEqual(acc.char_co2.value, 1100)
self.assertEqual(acc.char_peak.value, 1100)
self.assertEqual(acc.char_detected.value, 1)
hass.states.async_set(entity_id, '300')
await hass.async_block_till_done()
assert acc.char_light.value == 300
self.hass.states.set(entity_id, '800')
self.hass.block_till_done()
self.assertEqual(acc.char_co2.value, 800)
self.assertEqual(acc.char_peak.value, 1100)
self.assertEqual(acc.char_detected.value, 0)
def test_light(self):
"""Test if accessory is updated after state change."""
entity_id = 'sensor.light'
async def test_binary(hass):
"""Test if accessory is updated after state change."""
entity_id = 'binary_sensor.opening'
acc = LightSensor(self.hass, 'Light', entity_id, 2, config=None)
acc.run()
hass.states.async_set(entity_id, STATE_UNKNOWN,
{ATTR_DEVICE_CLASS: 'opening'})
await hass.async_block_till_done()
self.assertEqual(acc.aid, 2)
self.assertEqual(acc.category, 10) # Sensor
acc = BinarySensor(hass, 'Window Opening', entity_id, 2, config=None)
await hass.async_add_job(acc.run)
self.assertEqual(acc.char_light.value, 0.0001)
assert acc.aid == 2
assert acc.category == 10 # Sensor
self.hass.states.set(entity_id, STATE_UNKNOWN)
self.hass.block_till_done()
self.assertEqual(acc.char_light.value, 0.0001)
assert acc.char_detected.value == 0
self.hass.states.set(entity_id, '300')
self.hass.block_till_done()
self.assertEqual(acc.char_light.value, 300)
hass.states.async_set(entity_id, STATE_ON,
{ATTR_DEVICE_CLASS: 'opening'})
await hass.async_block_till_done()
assert acc.char_detected.value == 1
def test_binary(self):
"""Test if accessory is updated after state change."""
entity_id = 'binary_sensor.opening'
hass.states.async_set(entity_id, STATE_OFF,
{ATTR_DEVICE_CLASS: 'opening'})
await hass.async_block_till_done()
assert acc.char_detected.value == 0
self.hass.states.set(entity_id, STATE_UNKNOWN,
{ATTR_DEVICE_CLASS: "opening"})
self.hass.block_till_done()
hass.states.async_set(entity_id, STATE_HOME,
{ATTR_DEVICE_CLASS: 'opening'})
await hass.async_block_till_done()
assert acc.char_detected.value == 1
acc = BinarySensor(self.hass, 'Window Opening', entity_id,
2, config=None)
acc.run()
hass.states.async_set(entity_id, STATE_NOT_HOME,
{ATTR_DEVICE_CLASS: 'opening'})
await hass.async_block_till_done()
assert acc.char_detected.value == 0
self.assertEqual(acc.aid, 2)
self.assertEqual(acc.category, 10) # Sensor
hass.states.async_remove(entity_id)
await hass.async_block_till_done()
assert acc.char_detected.value == 0
self.assertEqual(acc.char_detected.value, 0)
self.hass.states.set(entity_id, STATE_ON,
{ATTR_DEVICE_CLASS: "opening"})
self.hass.block_till_done()
self.assertEqual(acc.char_detected.value, 1)
async def test_binary_device_classes(hass):
"""Test if services and characteristics are assigned correctly."""
entity_id = 'binary_sensor.demo'
self.hass.states.set(entity_id, STATE_OFF,
{ATTR_DEVICE_CLASS: "opening"})
self.hass.block_till_done()
self.assertEqual(acc.char_detected.value, 0)
for device_class, (service, char) in BINARY_SENSOR_SERVICE_MAP.items():
hass.states.async_set(entity_id, STATE_OFF,
{ATTR_DEVICE_CLASS: device_class})
await hass.async_block_till_done()
self.hass.states.set(entity_id, STATE_HOME,
{ATTR_DEVICE_CLASS: "opening"})
self.hass.block_till_done()
self.assertEqual(acc.char_detected.value, 1)
self.hass.states.set(entity_id, STATE_NOT_HOME,
{ATTR_DEVICE_CLASS: "opening"})
self.hass.block_till_done()
self.assertEqual(acc.char_detected.value, 0)
self.hass.states.remove(entity_id)
self.hass.block_till_done()
def test_binary_device_classes(self):
"""Test if services and characteristics are assigned correctly."""
entity_id = 'binary_sensor.demo'
for device_class, (service, char) in BINARY_SENSOR_SERVICE_MAP.items():
self.hass.states.set(entity_id, STATE_OFF,
{ATTR_DEVICE_CLASS: device_class})
self.hass.block_till_done()
acc = BinarySensor(self.hass, 'Binary Sensor', entity_id,
2, config=None)
self.assertEqual(acc.get_service(service).display_name, service)
self.assertEqual(acc.char_detected.display_name, char)
acc = BinarySensor(hass, 'Binary Sensor', entity_id, 2, config=None)
assert acc.get_service(service).display_name == service
assert acc.char_detected.display_name == char

121
tests/components/homekit/test_type_switches.py
View File

@ -1,104 +1,45 @@
"""Test different accessory types: Switches."""
import unittest
import pytest
from homeassistant.core import callback, split_entity_id
from homeassistant.core import split_entity_id
from homeassistant.components.homekit.type_switches import Switch
from homeassistant.const import (
ATTR_DOMAIN, ATTR_SERVICE, EVENT_CALL_SERVICE,
SERVICE_TURN_ON, SERVICE_TURN_OFF, STATE_ON, STATE_OFF)
from homeassistant.const import ATTR_ENTITY_ID, STATE_ON, STATE_OFF
from tests.common import get_test_home_assistant
from tests.common import async_mock_service
class TestHomekitSwitches(unittest.TestCase):
"""Test class for all accessory types regarding switches."""
@pytest.mark.parametrize('entity_id', [
'switch.test', 'remote.test', 'input_boolean.test'])
async def test_switch_set_state(hass, entity_id):
"""Test if accessory and HA are updated accordingly."""
domain = split_entity_id(entity_id)[0]
def setUp(self):
"""Setup things to be run when tests are started."""
self.hass = get_test_home_assistant()
self.events = []
acc = Switch(hass, 'Switch', entity_id, 2, config=None)
await hass.async_add_job(acc.run)
@callback
def record_event(event):
"""Track called event."""
self.events.append(event)
assert acc.aid == 2
assert acc.category == 8 # Switch
self.hass.bus.listen(EVENT_CALL_SERVICE, record_event)
assert acc.char_on.value is False
def tearDown(self):
"""Stop down everything that was started."""
self.hass.stop()
hass.states.async_set(entity_id, STATE_ON)
await hass.async_block_till_done()
assert acc.char_on.value is True
def test_switch_set_state(self):
"""Test if accessory and HA are updated accordingly."""
entity_id = 'switch.test'
domain = split_entity_id(entity_id)[0]
hass.states.async_set(entity_id, STATE_OFF)
await hass.async_block_till_done()
assert acc.char_on.value is False
acc = Switch(self.hass, 'Switch', entity_id, 2, config=None)
acc.run()
# Set from HomeKit
call_turn_on = async_mock_service(hass, domain, 'turn_on')
call_turn_off = async_mock_service(hass, domain, 'turn_off')
self.assertEqual(acc.aid, 2)
self.assertEqual(acc.category, 8) # Switch
await hass.async_add_job(acc.char_on.client_update_value, True)
await hass.async_block_till_done()
assert call_turn_on
assert call_turn_on[0].data[ATTR_ENTITY_ID] == entity_id
self.assertEqual(acc.char_on.value, False)
self.hass.states.set(entity_id, STATE_ON)
self.hass.block_till_done()
self.assertEqual(acc.char_on.value, True)
self.hass.states.set(entity_id, STATE_OFF)
self.hass.block_till_done()
self.assertEqual(acc.char_on.value, False)
# Set from HomeKit
acc.char_on.client_update_value(True)
self.hass.block_till_done()
self.assertEqual(
self.events[0].data[ATTR_DOMAIN], domain)
self.assertEqual(
self.events[0].data[ATTR_SERVICE], SERVICE_TURN_ON)
acc.char_on.client_update_value(False)
self.hass.block_till_done()
self.assertEqual(
self.events[1].data[ATTR_DOMAIN], domain)
self.assertEqual(
self.events[1].data[ATTR_SERVICE], SERVICE_TURN_OFF)
def test_remote_set_state(self):
"""Test service call for remote as domain."""
entity_id = 'remote.test'
domain = split_entity_id(entity_id)[0]
acc = Switch(self.hass, 'Switch', entity_id, 2, config=None)
acc.run()
self.assertEqual(acc.char_on.value, False)
# Set from HomeKit
acc.char_on.client_update_value(True)
self.hass.block_till_done()
self.assertEqual(
self.events[0].data[ATTR_DOMAIN], domain)
self.assertEqual(
self.events[0].data[ATTR_SERVICE], SERVICE_TURN_ON)
self.assertEqual(acc.char_on.value, True)
def test_input_boolean_set_state(self):
"""Test service call for remote as domain."""
entity_id = 'input_boolean.test'
domain = split_entity_id(entity_id)[0]
acc = Switch(self.hass, 'Switch', entity_id, 2, config=None)
acc.run()
self.assertEqual(acc.char_on.value, False)
# Set from HomeKit
acc.char_on.client_update_value(True)
self.hass.block_till_done()
self.assertEqual(
self.events[0].data[ATTR_DOMAIN], domain)
self.assertEqual(
self.events[0].data[ATTR_SERVICE], SERVICE_TURN_ON)
self.assertEqual(acc.char_on.value, True)
await hass.async_add_job(acc.char_on.client_update_value, False)
await hass.async_block_till_done()
assert call_turn_off
assert call_turn_off[0].data[ATTR_ENTITY_ID] == entity_id

607
tests/components/homekit/test_type_thermostats.py
View File

@ -1,364 +1,347 @@
"""Test different accessory types: Thermostats."""
import unittest
from collections import namedtuple
import pytest
from homeassistant.core import callback
from homeassistant.components.climate import (
ATTR_CURRENT_TEMPERATURE, ATTR_TEMPERATURE,
DOMAIN, ATTR_CURRENT_TEMPERATURE, ATTR_TEMPERATURE,
ATTR_TARGET_TEMP_LOW, ATTR_TARGET_TEMP_HIGH, ATTR_OPERATION_MODE,
ATTR_OPERATION_LIST, STATE_COOL, STATE_HEAT, STATE_AUTO)
from homeassistant.const import (
ATTR_ENTITY_ID, ATTR_SERVICE, ATTR_SERVICE_DATA, ATTR_SUPPORTED_FEATURES,
ATTR_UNIT_OF_MEASUREMENT, EVENT_CALL_SERVICE,
ATTR_ENTITY_ID, ATTR_SUPPORTED_FEATURES, ATTR_UNIT_OF_MEASUREMENT,
STATE_OFF, TEMP_CELSIUS, TEMP_FAHRENHEIT)
from tests.common import get_test_home_assistant
from tests.common import async_mock_service
from tests.components.homekit.test_accessories import patch_debounce
class TestHomekitThermostats(unittest.TestCase):
"""Test class for all accessory types regarding thermostats."""
@pytest.fixture(scope='module')
def cls(request):
"""Patch debounce decorator during import of type_thermostats."""
patcher = patch_debounce()
patcher.start()
_import = __import__('homeassistant.components.homekit.type_thermostats',
fromlist=['Thermostat'])
request.addfinalizer(patcher.stop)
patcher_tuple = namedtuple('Cls', ['thermostat'])
return patcher_tuple(thermostat=_import.Thermostat)
@classmethod
def setUpClass(cls):
"""Setup Thermostat class import and debounce patcher."""
cls.patcher = patch_debounce()
cls.patcher.start()
_import = __import__(
'homeassistant.components.homekit.type_thermostats',
fromlist=['Thermostat'])
cls.thermostat_cls = _import.Thermostat
@classmethod
def tearDownClass(cls):
"""Stop debounce patcher."""
cls.patcher.stop()
async def test_default_thermostat(hass, cls):
"""Test if accessory and HA are updated accordingly."""
entity_id = 'climate.test'
def setUp(self):
"""Setup things to be run when tests are started."""
self.hass = get_test_home_assistant()
self.events = []
hass.states.async_set(entity_id, STATE_OFF, {ATTR_SUPPORTED_FEATURES: 0})
await hass.async_block_till_done()
acc = cls.thermostat(hass, 'Climate', entity_id, 2, config=None)
await hass.async_add_job(acc.run)
@callback
def record_event(event):
"""Track called event."""
self.events.append(event)
assert acc.aid == 2
assert acc.category == 9 # Thermostat
self.hass.bus.listen(EVENT_CALL_SERVICE, record_event)
assert acc.char_current_heat_cool.value == 0
assert acc.char_target_heat_cool.value == 0
assert acc.char_current_temp.value == 21.0
assert acc.char_target_temp.value == 21.0
assert acc.char_display_units.value == 0
assert acc.char_cooling_thresh_temp is None
assert acc.char_heating_thresh_temp is None
def tearDown(self):
"""Stop down everything that was started."""
self.hass.stop()
hass.states.async_set(entity_id, STATE_HEAT,
{ATTR_OPERATION_MODE: STATE_HEAT,
ATTR_TEMPERATURE: 22.0,
ATTR_CURRENT_TEMPERATURE: 18.0,
ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS})
await hass.async_block_till_done()
assert acc.char_target_temp.value == 22.0
assert acc.char_current_heat_cool.value == 1
assert acc.char_target_heat_cool.value == 1
assert acc.char_current_temp.value == 18.0
assert acc.char_display_units.value == 0
def test_default_thermostat(self):
"""Test if accessory and HA are updated accordingly."""
climate = 'climate.test'
hass.states.async_set(entity_id, STATE_HEAT,
{ATTR_OPERATION_MODE: STATE_HEAT,
ATTR_TEMPERATURE: 22.0,
ATTR_CURRENT_TEMPERATURE: 23.0,
ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS})
await hass.async_block_till_done()
assert acc.char_target_temp.value == 22.0
assert acc.char_current_heat_cool.value == 0
assert acc.char_target_heat_cool.value == 1
assert acc.char_current_temp.value == 23.0
assert acc.char_display_units.value == 0
self.hass.states.set(climate, STATE_OFF, {ATTR_SUPPORTED_FEATURES: 0})
self.hass.block_till_done()
acc = self.thermostat_cls(self.hass, 'Climate', climate,
2, config=None)
acc.run()
hass.states.async_set(entity_id, STATE_COOL,
{ATTR_OPERATION_MODE: STATE_COOL,
ATTR_TEMPERATURE: 20.0,
ATTR_CURRENT_TEMPERATURE: 25.0,
ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS})
await hass.async_block_till_done()
assert acc.char_target_temp.value == 20.0
assert acc.char_current_heat_cool.value == 2
assert acc.char_target_heat_cool.value == 2
assert acc.char_current_temp.value == 25.0
assert acc.char_display_units.value == 0
self.assertEqual(acc.aid, 2)
self.assertEqual(acc.category, 9) # Thermostat
hass.states.async_set(entity_id, STATE_COOL,
{ATTR_OPERATION_MODE: STATE_COOL,
ATTR_TEMPERATURE: 20.0,
ATTR_CURRENT_TEMPERATURE: 19.0,
ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS})
await hass.async_block_till_done()
assert acc.char_target_temp.value == 20.0
assert acc.char_current_heat_cool.value == 0
assert acc.char_target_heat_cool.value == 2
assert acc.char_current_temp.value == 19.0
assert acc.char_display_units.value == 0
self.assertEqual(acc.char_current_heat_cool.value, 0)
self.assertEqual(acc.char_target_heat_cool.value, 0)
self.assertEqual(acc.char_current_temp.value, 21.0)
self.assertEqual(acc.char_target_temp.value, 21.0)
self.assertEqual(acc.char_display_units.value, 0)
self.assertEqual(acc.char_cooling_thresh_temp, None)
self.assertEqual(acc.char_heating_thresh_temp, None)
hass.states.async_set(entity_id, STATE_OFF,
{ATTR_OPERATION_MODE: STATE_OFF,
ATTR_TEMPERATURE: 22.0,
ATTR_CURRENT_TEMPERATURE: 18.0,
ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS})
await hass.async_block_till_done()
assert acc.char_target_temp.value == 22.0
assert acc.char_current_heat_cool.value == 0
assert acc.char_target_heat_cool.value == 0
assert acc.char_current_temp.value == 18.0
assert acc.char_display_units.value == 0
self.hass.states.set(climate, STATE_HEAT,
{ATTR_OPERATION_MODE: STATE_HEAT,
ATTR_TEMPERATURE: 22.0,
ATTR_CURRENT_TEMPERATURE: 18.0,
ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS})
self.hass.block_till_done()
self.assertEqual(acc.char_target_temp.value, 22.0)
self.assertEqual(acc.char_current_heat_cool.value, 1)
self.assertEqual(acc.char_target_heat_cool.value, 1)
self.assertEqual(acc.char_current_temp.value, 18.0)
self.assertEqual(acc.char_display_units.value, 0)
hass.states.async_set(entity_id, STATE_AUTO,
{ATTR_OPERATION_MODE: STATE_AUTO,
ATTR_OPERATION_LIST: [STATE_HEAT, STATE_COOL],
ATTR_TEMPERATURE: 22.0,
ATTR_CURRENT_TEMPERATURE: 18.0,
ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS})
await hass.async_block_till_done()
assert acc.char_target_temp.value == 22.0
assert acc.char_current_heat_cool.value == 1
assert acc.char_target_heat_cool.value == 3
assert acc.char_current_temp.value == 18.0
assert acc.char_display_units.value == 0
self.hass.states.set(climate, STATE_HEAT,
{ATTR_OPERATION_MODE: STATE_HEAT,
ATTR_TEMPERATURE: 22.0,
ATTR_CURRENT_TEMPERATURE: 23.0,
ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS})
self.hass.block_till_done()
self.assertEqual(acc.char_target_temp.value, 22.0)
self.assertEqual(acc.char_current_heat_cool.value, 0)
self.assertEqual(acc.char_target_heat_cool.value, 1)
self.assertEqual(acc.char_current_temp.value, 23.0)
self.assertEqual(acc.char_display_units.value, 0)
hass.states.async_set(entity_id, STATE_AUTO,
{ATTR_OPERATION_MODE: STATE_AUTO,
ATTR_OPERATION_LIST: [STATE_HEAT, STATE_COOL],
ATTR_TEMPERATURE: 22.0,
ATTR_CURRENT_TEMPERATURE: 25.0,
ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS})
await hass.async_block_till_done()
assert acc.char_target_temp.value == 22.0
assert acc.char_current_heat_cool.value == 2
assert acc.char_target_heat_cool.value == 3
assert acc.char_current_temp.value == 25.0
assert acc.char_display_units.value == 0
self.hass.states.set(climate, STATE_COOL,
{ATTR_OPERATION_MODE: STATE_COOL,
ATTR_TEMPERATURE: 20.0,
ATTR_CURRENT_TEMPERATURE: 25.0,
ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS})
self.hass.block_till_done()
self.assertEqual(acc.char_target_temp.value, 20.0)
self.assertEqual(acc.char_current_heat_cool.value, 2)
self.assertEqual(acc.char_target_heat_cool.value, 2)
self.assertEqual(acc.char_current_temp.value, 25.0)
self.assertEqual(acc.char_display_units.value, 0)
hass.states.async_set(entity_id, STATE_AUTO,
{ATTR_OPERATION_MODE: STATE_AUTO,
ATTR_OPERATION_LIST: [STATE_HEAT, STATE_COOL],
ATTR_TEMPERATURE: 22.0,
ATTR_CURRENT_TEMPERATURE: 22.0,
ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS})
await hass.async_block_till_done()
assert acc.char_target_temp.value == 22.0
assert acc.char_current_heat_cool.value == 0
assert acc.char_target_heat_cool.value == 3
assert acc.char_current_temp.value == 22.0
assert acc.char_display_units.value == 0
self.hass.states.set(climate, STATE_COOL,
{ATTR_OPERATION_MODE: STATE_COOL,
ATTR_TEMPERATURE: 20.0,
ATTR_CURRENT_TEMPERATURE: 19.0,
ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS})
self.hass.block_till_done()
self.assertEqual(acc.char_target_temp.value, 20.0)
self.assertEqual(acc.char_current_heat_cool.value, 0)
self.assertEqual(acc.char_target_heat_cool.value, 2)
self.assertEqual(acc.char_current_temp.value, 19.0)
self.assertEqual(acc.char_display_units.value, 0)
# Set from HomeKit
call_set_temperature = async_mock_service(hass, DOMAIN, 'set_temperature')
call_set_operation_mode = async_mock_service(hass, DOMAIN,
'set_operation_mode')
self.hass.states.set(climate, STATE_OFF,
{ATTR_OPERATION_MODE: STATE_OFF,
ATTR_TEMPERATURE: 22.0,
ATTR_CURRENT_TEMPERATURE: 18.0,
ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS})
self.hass.block_till_done()
self.assertEqual(acc.char_target_temp.value, 22.0)
self.assertEqual(acc.char_current_heat_cool.value, 0)
self.assertEqual(acc.char_target_heat_cool.value, 0)
self.assertEqual(acc.char_current_temp.value, 18.0)
self.assertEqual(acc.char_display_units.value, 0)
await hass.async_add_job(acc.char_target_temp.client_update_value, 19.0)
await hass.async_block_till_done()
assert call_set_temperature
assert call_set_temperature[0].data[ATTR_ENTITY_ID] == entity_id
assert call_set_temperature[0].data[ATTR_TEMPERATURE] == 19.0
assert acc.char_target_temp.value == 19.0
self.hass.states.set(climate, STATE_AUTO,
{ATTR_OPERATION_MODE: STATE_AUTO,
ATTR_OPERATION_LIST: [STATE_HEAT, STATE_COOL],
ATTR_TEMPERATURE: 22.0,
ATTR_CURRENT_TEMPERATURE: 18.0,
ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS})
self.hass.block_till_done()
self.assertEqual(acc.char_target_temp.value, 22.0)
self.assertEqual(acc.char_current_heat_cool.value, 1)
self.assertEqual(acc.char_target_heat_cool.value, 3)
self.assertEqual(acc.char_current_temp.value, 18.0)
self.assertEqual(acc.char_display_units.value, 0)
await hass.async_add_job(acc.char_target_heat_cool.client_update_value, 1)
await hass.async_block_till_done()
assert call_set_operation_mode
assert call_set_operation_mode[0].data[ATTR_ENTITY_ID] == entity_id
assert call_set_operation_mode[0].data[ATTR_OPERATION_MODE] == STATE_HEAT
assert acc.char_target_heat_cool.value == 1
self.hass.states.set(climate, STATE_AUTO,
{ATTR_OPERATION_MODE: STATE_AUTO,
ATTR_OPERATION_LIST: [STATE_HEAT, STATE_COOL],
ATTR_TEMPERATURE: 22.0,
ATTR_CURRENT_TEMPERATURE: 25.0,
ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS})
self.hass.block_till_done()
self.assertEqual(acc.char_target_temp.value, 22.0)
self.assertEqual(acc.char_current_heat_cool.value, 2)
self.assertEqual(acc.char_target_heat_cool.value, 3)
self.assertEqual(acc.char_current_temp.value, 25.0)
self.assertEqual(acc.char_display_units.value, 0)
self.hass.states.set(climate, STATE_AUTO,
{ATTR_OPERATION_MODE: STATE_AUTO,
ATTR_OPERATION_LIST: [STATE_HEAT, STATE_COOL],
ATTR_TEMPERATURE: 22.0,
ATTR_CURRENT_TEMPERATURE: 22.0,
ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS})
self.hass.block_till_done()
self.assertEqual(acc.char_target_temp.value, 22.0)
self.assertEqual(acc.char_current_heat_cool.value, 0)
self.assertEqual(acc.char_target_heat_cool.value, 3)
self.assertEqual(acc.char_current_temp.value, 22.0)
self.assertEqual(acc.char_display_units.value, 0)
async def test_auto_thermostat(hass, cls):
"""Test if accessory and HA are updated accordingly."""
entity_id = 'climate.test'
# Set from HomeKit
acc.char_target_temp.client_update_value(19.0)
self.hass.block_till_done()
self.assertEqual(
self.events[0].data[ATTR_SERVICE], 'set_temperature')
self.assertEqual(
self.events[0].data[ATTR_SERVICE_DATA][ATTR_TEMPERATURE], 19.0)
self.assertEqual(acc.char_target_temp.value, 19.0)
# support_auto = True
hass.states.async_set(entity_id, STATE_OFF, {ATTR_SUPPORTED_FEATURES: 6})
await hass.async_block_till_done()
acc = cls.thermostat(hass, 'Climate', entity_id, 2, config=None)
await hass.async_add_job(acc.run)
acc.char_target_heat_cool.client_update_value(1)
self.hass.block_till_done()
self.assertEqual(
self.events[1].data[ATTR_SERVICE], 'set_operation_mode')
self.assertEqual(
self.events[1].data[ATTR_SERVICE_DATA][ATTR_OPERATION_MODE],
STATE_HEAT)
self.assertEqual(acc.char_target_heat_cool.value, 1)
assert acc.char_cooling_thresh_temp.value == 23.0
assert acc.char_heating_thresh_temp.value == 19.0
def test_auto_thermostat(self):
"""Test if accessory and HA are updated accordingly."""
climate = 'climate.test'
hass.states.async_set(entity_id, STATE_AUTO,
{ATTR_OPERATION_MODE: STATE_AUTO,
ATTR_TARGET_TEMP_HIGH: 22.0,
ATTR_TARGET_TEMP_LOW: 20.0,
ATTR_CURRENT_TEMPERATURE: 18.0,
ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS})
await hass.async_block_till_done()
assert acc.char_heating_thresh_temp.value == 20.0
assert acc.char_cooling_thresh_temp.value == 22.0
assert acc.char_current_heat_cool.value == 1
assert acc.char_target_heat_cool.value == 3
assert acc.char_current_temp.value == 18.0
assert acc.char_display_units.value == 0
# support_auto = True
self.hass.states.set(climate, STATE_OFF, {ATTR_SUPPORTED_FEATURES: 6})
self.hass.block_till_done()
acc = self.thermostat_cls(self.hass, 'Climate', climate,
2, config=None)
acc.run()
hass.states.async_set(entity_id, STATE_AUTO,
{ATTR_OPERATION_MODE: STATE_AUTO,
ATTR_TARGET_TEMP_HIGH: 23.0,
ATTR_TARGET_TEMP_LOW: 19.0,
ATTR_CURRENT_TEMPERATURE: 24.0,
ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS})
await hass.async_block_till_done()
assert acc.char_heating_thresh_temp.value == 19.0
assert acc.char_cooling_thresh_temp.value == 23.0
assert acc.char_current_heat_cool.value == 2
assert acc.char_target_heat_cool.value == 3
assert acc.char_current_temp.value == 24.0
assert acc.char_display_units.value == 0
self.assertEqual(acc.char_cooling_thresh_temp.value, 23.0)
self.assertEqual(acc.char_heating_thresh_temp.value, 19.0)
hass.states.async_set(entity_id, STATE_AUTO,
{ATTR_OPERATION_MODE: STATE_AUTO,
ATTR_TARGET_TEMP_HIGH: 23.0,
ATTR_TARGET_TEMP_LOW: 19.0,
ATTR_CURRENT_TEMPERATURE: 21.0,
ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS})
await hass.async_block_till_done()
assert acc.char_heating_thresh_temp.value == 19.0
assert acc.char_cooling_thresh_temp.value == 23.0
assert acc.char_current_heat_cool.value == 0
assert acc.char_target_heat_cool.value == 3
assert acc.char_current_temp.value == 21.0
assert acc.char_display_units.value == 0
self.hass.states.set(climate, STATE_AUTO,
{ATTR_OPERATION_MODE: STATE_AUTO,
ATTR_TARGET_TEMP_HIGH: 22.0,
ATTR_TARGET_TEMP_LOW: 20.0,
ATTR_CURRENT_TEMPERATURE: 18.0,
ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS})
self.hass.block_till_done()
self.assertEqual(acc.char_heating_thresh_temp.value, 20.0)
self.assertEqual(acc.char_cooling_thresh_temp.value, 22.0)
self.assertEqual(acc.char_current_heat_cool.value, 1)
self.assertEqual(acc.char_target_heat_cool.value, 3)
self.assertEqual(acc.char_current_temp.value, 18.0)
self.assertEqual(acc.char_display_units.value, 0)
# Set from HomeKit
call_set_temperature = async_mock_service(hass, DOMAIN, 'set_temperature')
self.hass.states.set(climate, STATE_AUTO,
{ATTR_OPERATION_MODE: STATE_AUTO,
ATTR_TARGET_TEMP_HIGH: 23.0,
ATTR_TARGET_TEMP_LOW: 19.0,
ATTR_CURRENT_TEMPERATURE: 24.0,
ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS})
self.hass.block_till_done()
self.assertEqual(acc.char_heating_thresh_temp.value, 19.0)
self.assertEqual(acc.char_cooling_thresh_temp.value, 23.0)
self.assertEqual(acc.char_current_heat_cool.value, 2)
self.assertEqual(acc.char_target_heat_cool.value, 3)
self.assertEqual(acc.char_current_temp.value, 24.0)
self.assertEqual(acc.char_display_units.value, 0)
await hass.async_add_job(
acc.char_heating_thresh_temp.client_update_value, 20.0)
await hass.async_block_till_done()
assert call_set_temperature[0]
assert call_set_temperature[0].data[ATTR_ENTITY_ID] == entity_id
assert call_set_temperature[0].data[ATTR_TARGET_TEMP_LOW] == 20.0
assert acc.char_heating_thresh_temp.value == 20.0
self.hass.states.set(climate, STATE_AUTO,
{ATTR_OPERATION_MODE: STATE_AUTO,
ATTR_TARGET_TEMP_HIGH: 23.0,
ATTR_TARGET_TEMP_LOW: 19.0,
ATTR_CURRENT_TEMPERATURE: 21.0,
ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS})
self.hass.block_till_done()
self.assertEqual(acc.char_heating_thresh_temp.value, 19.0)
self.assertEqual(acc.char_cooling_thresh_temp.value, 23.0)
self.assertEqual(acc.char_current_heat_cool.value, 0)
self.assertEqual(acc.char_target_heat_cool.value, 3)
self.assertEqual(acc.char_current_temp.value, 21.0)
self.assertEqual(acc.char_display_units.value, 0)
await hass.async_add_job(
acc.char_cooling_thresh_temp.client_update_value, 25.0)
await hass.async_block_till_done()
assert call_set_temperature[1]
assert call_set_temperature[1].data[ATTR_ENTITY_ID] == entity_id
assert call_set_temperature[1].data[ATTR_TARGET_TEMP_HIGH] == 25.0
assert acc.char_cooling_thresh_temp.value == 25.0
# Set from HomeKit
acc.char_heating_thresh_temp.client_update_value(20.0)
self.hass.block_till_done()
self.assertEqual(
self.events[0].data[ATTR_SERVICE], 'set_temperature')
self.assertEqual(
self.events[0].data[ATTR_SERVICE_DATA][ATTR_TARGET_TEMP_LOW], 20.0)
self.assertEqual(acc.char_heating_thresh_temp.value, 20.0)
acc.char_cooling_thresh_temp.client_update_value(25.0)
self.hass.block_till_done()
self.assertEqual(
self.events[1].data[ATTR_SERVICE], 'set_temperature')
self.assertEqual(
self.events[1].data[ATTR_SERVICE_DATA][ATTR_TARGET_TEMP_HIGH],
25.0)
self.assertEqual(acc.char_cooling_thresh_temp.value, 25.0)
async def test_power_state(hass, cls):
"""Test if accessory and HA are updated accordingly."""
entity_id = 'climate.test'
def test_power_state(self):
"""Test if accessory and HA are updated accordingly."""
climate = 'climate.test'
# SUPPORT_ON_OFF = True
hass.states.async_set(entity_id, STATE_HEAT,
{ATTR_SUPPORTED_FEATURES: 4096,
ATTR_OPERATION_MODE: STATE_HEAT,
ATTR_TEMPERATURE: 23.0,
ATTR_CURRENT_TEMPERATURE: 18.0})
await hass.async_block_till_done()
acc = cls.thermostat(hass, 'Climate', entity_id, 2, config=None)
await hass.async_add_job(acc.run)
assert acc.support_power_state is True
# SUPPORT_ON_OFF = True
self.hass.states.set(climate, STATE_HEAT,
{ATTR_SUPPORTED_FEATURES: 4096,
ATTR_OPERATION_MODE: STATE_HEAT,
ATTR_TEMPERATURE: 23.0,
ATTR_CURRENT_TEMPERATURE: 18.0})
self.hass.block_till_done()
acc = self.thermostat_cls(self.hass, 'Climate', climate,
2, config=None)
acc.run()
self.assertTrue(acc.support_power_state)
assert acc.char_current_heat_cool.value == 1
assert acc.char_target_heat_cool.value == 1
self.assertEqual(acc.char_current_heat_cool.value, 1)
self.assertEqual(acc.char_target_heat_cool.value, 1)
hass.states.async_set(entity_id, STATE_OFF,
{ATTR_OPERATION_MODE: STATE_HEAT,
ATTR_TEMPERATURE: 23.0,
ATTR_CURRENT_TEMPERATURE: 18.0})
await hass.async_block_till_done()
assert acc.char_current_heat_cool.value == 0
assert acc.char_target_heat_cool.value == 0
self.hass.states.set(climate, STATE_OFF,
{ATTR_OPERATION_MODE: STATE_HEAT,
ATTR_TEMPERATURE: 23.0,
ATTR_CURRENT_TEMPERATURE: 18.0})
self.hass.block_till_done()
self.assertEqual(acc.char_current_heat_cool.value, 0)
self.assertEqual(acc.char_target_heat_cool.value, 0)
hass.states.async_set(entity_id, STATE_OFF,
{ATTR_OPERATION_MODE: STATE_OFF,
ATTR_TEMPERATURE: 23.0,
ATTR_CURRENT_TEMPERATURE: 18.0})
await hass.async_block_till_done()
assert acc.char_current_heat_cool.value == 0
assert acc.char_target_heat_cool.value == 0
self.hass.states.set(climate, STATE_OFF,
{ATTR_OPERATION_MODE: STATE_OFF,
ATTR_TEMPERATURE: 23.0,
ATTR_CURRENT_TEMPERATURE: 18.0})
self.hass.block_till_done()
self.assertEqual(acc.char_current_heat_cool.value, 0)
self.assertEqual(acc.char_target_heat_cool.value, 0)
# Set from HomeKit
call_turn_on = async_mock_service(hass, DOMAIN, 'turn_on')
call_turn_off = async_mock_service(hass, DOMAIN, 'turn_off')
call_set_operation_mode = async_mock_service(hass, DOMAIN,
'set_operation_mode')
# Set from HomeKit
acc.char_target_heat_cool.client_update_value(1)
self.hass.block_till_done()
self.assertEqual(
self.events[0].data[ATTR_SERVICE], 'turn_on')
self.assertEqual(
self.events[0].data[ATTR_SERVICE_DATA][ATTR_ENTITY_ID],
climate)
self.assertEqual(
self.events[1].data[ATTR_SERVICE], 'set_operation_mode')
self.assertEqual(
self.events[1].data[ATTR_SERVICE_DATA][ATTR_OPERATION_MODE],
STATE_HEAT)
self.assertEqual(acc.char_target_heat_cool.value, 1)
await hass.async_add_job(acc.char_target_heat_cool.client_update_value, 1)
await hass.async_block_till_done()
assert call_turn_on
assert call_turn_on[0].data[ATTR_ENTITY_ID] == entity_id
assert call_set_operation_mode
assert call_set_operation_mode[0].data[ATTR_ENTITY_ID] == entity_id
assert call_set_operation_mode[0].data[ATTR_OPERATION_MODE] == STATE_HEAT
assert acc.char_target_heat_cool.value == 1
acc.char_target_heat_cool.client_update_value(0)
self.hass.block_till_done()
self.assertEqual(
self.events[2].data[ATTR_SERVICE], 'turn_off')
self.assertEqual(
self.events[2].data[ATTR_SERVICE_DATA][ATTR_ENTITY_ID],
climate)
self.assertEqual(acc.char_target_heat_cool.value, 0)
await hass.async_add_job(acc.char_target_heat_cool.client_update_value, 0)
await hass.async_block_till_done()
assert call_turn_off
assert call_turn_off[0].data[ATTR_ENTITY_ID] == entity_id
assert acc.char_target_heat_cool.value == 0
def test_thermostat_fahrenheit(self):
"""Test if accessory and HA are updated accordingly."""
climate = 'climate.test'
# support_auto = True
self.hass.states.set(climate, STATE_OFF, {ATTR_SUPPORTED_FEATURES: 6})
self.hass.block_till_done()
acc = self.thermostat_cls(self.hass, 'Climate', climate,
2, config=None)
acc.run()
async def test_thermostat_fahrenheit(hass, cls):
"""Test if accessory and HA are updated accordingly."""
entity_id = 'climate.test'
self.hass.states.set(climate, STATE_AUTO,
{ATTR_OPERATION_MODE: STATE_AUTO,
ATTR_TARGET_TEMP_HIGH: 75.2,
ATTR_TARGET_TEMP_LOW: 68,
ATTR_TEMPERATURE: 71.6,
ATTR_CURRENT_TEMPERATURE: 73.4,
ATTR_UNIT_OF_MEASUREMENT: TEMP_FAHRENHEIT})
self.hass.block_till_done()
self.assertEqual(acc.char_heating_thresh_temp.value, 20.0)
self.assertEqual(acc.char_cooling_thresh_temp.value, 24.0)
self.assertEqual(acc.char_current_temp.value, 23.0)
self.assertEqual(acc.char_target_temp.value, 22.0)
self.assertEqual(acc.char_display_units.value, 1)
# support_auto = True
hass.states.async_set(entity_id, STATE_OFF, {ATTR_SUPPORTED_FEATURES: 6})
await hass.async_block_till_done()
acc = cls.thermostat(hass, 'Climate', entity_id, 2, config=None)
await hass.async_add_job(acc.run)
# Set from HomeKit
acc.char_cooling_thresh_temp.client_update_value(23)
self.hass.block_till_done()
service_data = self.events[-1].data[ATTR_SERVICE_DATA]
self.assertEqual(service_data[ATTR_TARGET_TEMP_HIGH], 73.4)
self.assertEqual(service_data[ATTR_TARGET_TEMP_LOW], 68)
hass.states.async_set(entity_id, STATE_AUTO,
{ATTR_OPERATION_MODE: STATE_AUTO,
ATTR_TARGET_TEMP_HIGH: 75.2,
ATTR_TARGET_TEMP_LOW: 68,
ATTR_TEMPERATURE: 71.6,
ATTR_CURRENT_TEMPERATURE: 73.4,
ATTR_UNIT_OF_MEASUREMENT: TEMP_FAHRENHEIT})
await hass.async_block_till_done()
assert acc.char_heating_thresh_temp.value == 20.0
assert acc.char_cooling_thresh_temp.value == 24.0
assert acc.char_current_temp.value == 23.0
assert acc.char_target_temp.value == 22.0
assert acc.char_display_units.value == 1
acc.char_heating_thresh_temp.client_update_value(22)
self.hass.block_till_done()
service_data = self.events[-1].data[ATTR_SERVICE_DATA]
self.assertEqual(service_data[ATTR_TARGET_TEMP_HIGH], 73.4)
self.assertEqual(service_data[ATTR_TARGET_TEMP_LOW], 71.6)
# Set from HomeKit
call_set_temperature = async_mock_service(hass, DOMAIN, 'set_temperature')
acc.char_target_temp.client_update_value(24.0)
self.hass.block_till_done()
service_data = self.events[-1].data[ATTR_SERVICE_DATA]
self.assertEqual(service_data[ATTR_TEMPERATURE], 75.2)
await hass.async_add_job(
acc.char_cooling_thresh_temp.client_update_value, 23)
await hass.async_block_till_done()
assert call_set_temperature[0]
assert call_set_temperature[0].data[ATTR_ENTITY_ID] == entity_id
assert call_set_temperature[0].data[ATTR_TARGET_TEMP_HIGH] == 73.4
assert call_set_temperature[0].data[ATTR_TARGET_TEMP_LOW] == 68
await hass.async_add_job(
acc.char_heating_thresh_temp.client_update_value, 22)
await hass.async_block_till_done()
assert call_set_temperature[1]
assert call_set_temperature[1].data[ATTR_ENTITY_ID] == entity_id
assert call_set_temperature[1].data[ATTR_TARGET_TEMP_HIGH] == 73.4
assert call_set_temperature[1].data[ATTR_TARGET_TEMP_LOW] == 71.6
await hass.async_add_job(acc.char_target_temp.client_update_value, 24.0)
await hass.async_block_till_done()
assert call_set_temperature[2]
assert call_set_temperature[2].data[ATTR_ENTITY_ID] == entity_id
assert call_set_temperature[2].data[ATTR_TEMPERATURE] == 75.2

73
tests/components/homekit/test_util.py
View File

@ -1,25 +1,20 @@
"""Test HomeKit util module."""
import unittest
import voluptuous as vol
import pytest
import voluptuous as vol
from homeassistant.core import callback
from homeassistant.components.homekit.accessories import HomeBridge
from homeassistant.components.homekit.const import HOMEKIT_NOTIFY_ID
from homeassistant.components.homekit.util import (
show_setup_message, dismiss_setup_message, convert_to_float,
temperature_to_homekit, temperature_to_states, ATTR_CODE,
density_to_air_quality)
temperature_to_homekit, temperature_to_states, density_to_air_quality)
from homeassistant.components.homekit.util import validate_entity_config \
as vec
from homeassistant.components.persistent_notification import (
SERVICE_CREATE, SERVICE_DISMISS, ATTR_NOTIFICATION_ID)
DOMAIN, ATTR_NOTIFICATION_ID)
from homeassistant.const import (
EVENT_CALL_SERVICE, ATTR_DOMAIN, ATTR_SERVICE, ATTR_SERVICE_DATA,
TEMP_CELSIUS, TEMP_FAHRENHEIT, STATE_UNKNOWN)
ATTR_CODE, STATE_UNKNOWN, TEMP_CELSIUS, TEMP_FAHRENHEIT)
from tests.common import get_test_home_assistant
from tests.common import async_mock_service
def test_validate_entity_config():
@ -68,51 +63,27 @@ def test_density_to_air_quality():
assert density_to_air_quality(300) == 5
class TestUtil(unittest.TestCase):
"""Test all HomeKit util methods."""
async def test_show_setup_msg(hass):
"""Test show setup message as persistence notification."""
bridge = HomeBridge(hass)
def setUp(self):
"""Setup things to be run when tests are started."""
self.hass = get_test_home_assistant()
self.events = []
call_create_notification = async_mock_service(hass, DOMAIN, 'create')
@callback
def record_event(event):
"""Track called event."""
self.events.append(event)
await hass.async_add_job(show_setup_message, hass, bridge)
await hass.async_block_till_done()
self.hass.bus.listen(EVENT_CALL_SERVICE, record_event)
assert call_create_notification
assert call_create_notification[0].data[ATTR_NOTIFICATION_ID] == \
HOMEKIT_NOTIFY_ID
def tearDown(self):
"""Stop down everything that was started."""
self.hass.stop()
def test_show_setup_msg(self):
"""Test show setup message as persistence notification."""
bridge = HomeBridge(self.hass)
async def test_dismiss_setup_msg(hass):
"""Test dismiss setup message."""
call_dismiss_notification = async_mock_service(hass, DOMAIN, 'dismiss')
show_setup_message(self.hass, bridge)
self.hass.block_till_done()
await hass.async_add_job(dismiss_setup_message, hass)
await hass.async_block_till_done()
data = self.events[0].data
self.assertEqual(
data.get(ATTR_DOMAIN, None), 'persistent_notification')
self.assertEqual(data.get(ATTR_SERVICE, None), SERVICE_CREATE)
self.assertNotEqual(data.get(ATTR_SERVICE_DATA, None), None)
self.assertEqual(
data[ATTR_SERVICE_DATA].get(ATTR_NOTIFICATION_ID, None),
HOMEKIT_NOTIFY_ID)
def test_dismiss_setup_msg(self):
"""Test dismiss setup message."""
dismiss_setup_message(self.hass)
self.hass.block_till_done()
data = self.events[0].data
self.assertEqual(
data.get(ATTR_DOMAIN, None), 'persistent_notification')
self.assertEqual(data.get(ATTR_SERVICE, None), SERVICE_DISMISS)
self.assertNotEqual(data.get(ATTR_SERVICE_DATA, None), None)
self.assertEqual(
data[ATTR_SERVICE_DATA].get(ATTR_NOTIFICATION_ID, None),
HOMEKIT_NOTIFY_ID)
assert call_dismiss_notification
assert call_dismiss_notification[0].data[ATTR_NOTIFICATION_ID] == \
HOMEKIT_NOTIFY_ID