jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-24 09:17:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Refactor homekit_controller to be fully asynchronous (#32111)

Browse Source 
* Port homekit_controller to aiohomekit

* Remove succeed() test helper

* Remove fail() test helper
This commit is contained in:
Jc2k 2020-02-24 09:55:33 +00:00 committed by GitHub
parent a1a835cf54
commit df9363610c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
31 changed files with 560 additions and 583 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
homeassistant/components/homekit_controller/__init__.py
View File

@ -1,8 +1,8 @@
"""Support for Homekit device discovery."""
import logging
import homekit
from homekit.model.characteristics import CharacteristicsTypes
import aiohomekit
from aiohomekit.model.characteristics import CharacteristicsTypes
from homeassistant.core import callback
from homeassistant.exceptions import ConfigEntryNotReady
@ -94,7 +94,8 @@ class HomeKitEntity(Entity):
def _setup_characteristic(self, char):
"""Configure an entity based on a HomeKit characteristics metadata."""
# Build up a list of (aid, iid) tuples to poll on update()
self.pollable_characteristics.append((self._aid, char["iid"]))
if "pr" in char["perms"]:
self.pollable_characteristics.append((self._aid, char["iid"]))
# Build a map of ctype -> iid
short_name = CharacteristicsTypes.get_short(char["type"])
@ -223,7 +224,7 @@ async def async_setup(hass, config):
map_storage = hass.data[ENTITY_MAP] = EntityMapStorage(hass)
await map_storage.async_initialize()
hass.data[CONTROLLER] = homekit.Controller()
hass.data[CONTROLLER] = aiohomekit.Controller()
hass.data[KNOWN_DEVICES] = {}
return True

2
homeassistant/components/homekit_controller/air_quality.py
View File

@ -1,5 +1,5 @@
"""Support for HomeKit Controller air quality sensors."""
from homekit.model.characteristics import CharacteristicsTypes
from aiohomekit.model.characteristics import CharacteristicsTypes
from homeassistant.components.air_quality import AirQualityEntity
from homeassistant.core import callback

2
homeassistant/components/homekit_controller/alarm_control_panel.py
View File

@ -1,7 +1,7 @@
"""Support for Homekit Alarm Control Panel."""
import logging
from homekit.model.characteristics import CharacteristicsTypes
from aiohomekit.model.characteristics import CharacteristicsTypes
from homeassistant.components.alarm_control_panel import AlarmControlPanel
from homeassistant.components.alarm_control_panel.const import (

2
homeassistant/components/homekit_controller/binary_sensor.py
View File

@ -1,7 +1,7 @@
"""Support for Homekit motion sensors."""
import logging
from homekit.model.characteristics import CharacteristicsTypes
from aiohomekit.model.characteristics import CharacteristicsTypes
from homeassistant.components.binary_sensor import (
DEVICE_CLASS_SMOKE,

2
homeassistant/components/homekit_controller/climate.py
View File

@ -1,7 +1,7 @@
"""Support for Homekit climate devices."""
import logging
from homekit.model.characteristics import CharacteristicsTypes
from aiohomekit.model.characteristics import CharacteristicsTypes
from homeassistant.components.climate import (
DEFAULT_MAX_HUMIDITY,

66
homeassistant/components/homekit_controller/config_flow.py
View File

@ -4,8 +4,9 @@ import logging
import os
import re
import homekit
from homekit.controller.ip_implementation import IpPairing
import aiohomekit
from aiohomekit import Controller
from aiohomekit.controller.ip import IpPairing
import voluptuous as vol
from homeassistant import config_entries
@ -72,7 +73,7 @@ def ensure_pin_format(pin):
"""
match = PIN_FORMAT.search(pin)
if not match:
raise homekit.exceptions.MalformedPinError(f"Invalid PIN code f{pin}")
raise aiohomekit.exceptions.MalformedPinError(f"Invalid PIN code f{pin}")
return "{}-{}-{}".format(*match.groups())
@ -88,7 +89,7 @@ class HomekitControllerFlowHandler(config_entries.ConfigFlow):
self.model = None
self.hkid = None
self.devices = {}
self.controller = homekit.Controller()
self.controller = Controller()
self.finish_pairing = None
async def async_step_user(self, user_input=None):
@ -97,22 +98,22 @@ class HomekitControllerFlowHandler(config_entries.ConfigFlow):
if user_input is not None:
key = user_input["device"]
self.hkid = self.devices[key]["id"]
self.model = self.devices[key]["md"]
self.hkid = self.devices[key].device_id
self.model = self.devices[key].info["md"]
await self.async_set_unique_id(
normalize_hkid(self.hkid), raise_on_progress=False
)
return await self.async_step_pair()
all_hosts = await self.hass.async_add_executor_job(self.controller.discover, 5)
all_hosts = await self.controller.discover_ip()
self.devices = {}
for host in all_hosts:
status_flags = int(host["sf"])
status_flags = int(host.info["sf"])
paired = not status_flags & 0x01
if paired:
continue
self.devices[host["name"]] = host
self.devices[host.info["name"]] = host
if not self.devices:
return self.async_abort(reason="no_devices")
@ -130,10 +131,11 @@ class HomekitControllerFlowHandler(config_entries.ConfigFlow):
unique_id = user_input["unique_id"]
await self.async_set_unique_id(unique_id)
records = await self.hass.async_add_executor_job(self.controller.discover, 5)
for record in records:
if normalize_hkid(record["id"]) != unique_id:
devices = await self.controller.discover_ip(5)
for device in devices:
if normalize_hkid(device.device_id) != unique_id:
continue
record = device.info
return await self.async_step_zeroconf(
{
"host": record["address"],
@ -295,55 +297,49 @@ class HomekitControllerFlowHandler(config_entries.ConfigFlow):
code = pair_info["pairing_code"]
try:
code = ensure_pin_format(code)
await self.hass.async_add_executor_job(self.finish_pairing, code)
pairing = self.controller.pairings.get(self.hkid)
if pairing:
return await self._entry_from_accessory(pairing)
errors["pairing_code"] = "unable_to_pair"
except homekit.exceptions.MalformedPinError:
pairing = await self.finish_pairing(code)
return await self._entry_from_accessory(pairing)
except aiohomekit.exceptions.MalformedPinError:
# Library claimed pin was invalid before even making an API call
errors["pairing_code"] = "authentication_error"
except homekit.AuthenticationError:
except aiohomekit.AuthenticationError:
# PairSetup M4 - SRP proof failed
# PairSetup M6 - Ed25519 signature verification failed
# PairVerify M4 - Decryption failed
# PairVerify M4 - Device not recognised
# PairVerify M4 - Ed25519 signature verification failed
errors["pairing_code"] = "authentication_error"
except homekit.UnknownError:
except aiohomekit.UnknownError:
# An error occurred on the device whilst performing this
# operation.
errors["pairing_code"] = "unknown_error"
except homekit.MaxPeersError:
except aiohomekit.MaxPeersError:
# The device can't pair with any more accessories.
errors["pairing_code"] = "max_peers_error"
except homekit.AccessoryNotFoundError:
except aiohomekit.AccessoryNotFoundError:
# Can no longer find the device on the network
return self.async_abort(reason="accessory_not_found_error")
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Pairing attempt failed with an unhandled exception")
errors["pairing_code"] = "pairing_failed"
start_pairing = self.controller.start_pairing
discovery = await self.controller.find_ip_by_device_id(self.hkid)
try:
self.finish_pairing = await self.hass.async_add_executor_job(
start_pairing, self.hkid, self.hkid
)
except homekit.BusyError:
self.finish_pairing = await discovery.start_pairing(self.hkid)
except aiohomekit.BusyError:
# Already performing a pair setup operation with a different
# controller
errors["pairing_code"] = "busy_error"
except homekit.MaxTriesError:
except aiohomekit.MaxTriesError:
# The accessory has received more than 100 unsuccessful auth
# attempts.
errors["pairing_code"] = "max_tries_error"
except homekit.UnavailableError:
except aiohomekit.UnavailableError:
# The accessory is already paired - cannot try to pair again.
return self.async_abort(reason="already_paired")
except homekit.AccessoryNotFoundError:
except aiohomekit.AccessoryNotFoundError:
# Can no longer find the device on the network
return self.async_abort(reason="accessory_not_found_error")
except Exception: # pylint: disable=broad-except
@ -376,9 +372,7 @@ class HomekitControllerFlowHandler(config_entries.ConfigFlow):
# the same time.
accessories = pairing_data.pop("accessories", None)
if not accessories:
accessories = await self.hass.async_add_executor_job(
pairing.list_accessories_and_characteristics
)
accessories = await pairing.list_accessories_and_characteristics()
bridge_info = get_bridge_information(accessories)
name = get_accessory_name(bridge_info)

22
homeassistant/components/homekit_controller/connection.py
View File

@ -3,14 +3,14 @@ import asyncio
import datetime
import logging
from homekit.controller.ip_implementation import IpPairing
from homekit.exceptions import (
from aiohomekit.controller.ip import IpPairing
from aiohomekit.exceptions import (
AccessoryDisconnectedError,
AccessoryNotFoundError,
EncryptionError,
)
from homekit.model.characteristics import CharacteristicsTypes
from homekit.model.services import ServicesTypes
from aiohomekit.model.characteristics import CharacteristicsTypes
from aiohomekit.model.services import ServicesTypes
from homeassistant.core import callback
from homeassistant.helpers.event import async_track_time_interval
@ -186,10 +186,7 @@ class HKDevice:
async def async_refresh_entity_map(self, config_num):
"""Handle setup of a HomeKit accessory."""
try:
async with self.pairing_lock:
self.accessories = await self.hass.async_add_executor_job(
self.pairing.list_accessories_and_characteristics
)
self.accessories = await self.pairing.list_accessories_and_characteristics()
except AccessoryDisconnectedError:
# If we fail to refresh this data then we will naturally retry
# later when Bonjour spots c# is still not up to date.
@ -305,10 +302,7 @@ class HKDevice:
async def get_characteristics(self, *args, **kwargs):
"""Read latest state from homekit accessory."""
async with self.pairing_lock:
chars = await self.hass.async_add_executor_job(
self.pairing.get_characteristics, *args, **kwargs
)
return chars
return await self.pairing.get_characteristics(*args, **kwargs)
async def put_characteristics(self, characteristics):
"""Control a HomeKit device state from Home Assistant."""
@ -317,9 +311,7 @@ class HKDevice:
chars.append((row["aid"], row["iid"], row["value"]))
async with self.pairing_lock:
results = await self.hass.async_add_executor_job(
self.pairing.put_characteristics, chars
)
results = await self.pairing.put_characteristics(chars)
# Feed characteristics back into HA and update the current state
# results will only contain failures, so anythin in characteristics

2
homeassistant/components/homekit_controller/cover.py
View File

@ -1,7 +1,7 @@
"""Support for Homekit covers."""
import logging
from homekit.model.characteristics import CharacteristicsTypes
from aiohomekit.model.characteristics import CharacteristicsTypes
from homeassistant.components.cover import (
ATTR_POSITION,

2
homeassistant/components/homekit_controller/fan.py
View File

@ -1,7 +1,7 @@
"""Support for Homekit fans."""
import logging
from homekit.model.characteristics import CharacteristicsTypes
from aiohomekit.model.characteristics import CharacteristicsTypes
from homeassistant.components.fan import (
DIRECTION_FORWARD,

2
homeassistant/components/homekit_controller/light.py
View File

@ -1,7 +1,7 @@
"""Support for Homekit lights."""
import logging
from homekit.model.characteristics import CharacteristicsTypes
from aiohomekit.model.characteristics import CharacteristicsTypes
from homeassistant.components.light import (
ATTR_BRIGHTNESS,

2
homeassistant/components/homekit_controller/lock.py
View File

@ -1,7 +1,7 @@
"""Support for HomeKit Controller locks."""
import logging
from homekit.model.characteristics import CharacteristicsTypes
from aiohomekit.model.characteristics import CharacteristicsTypes
from homeassistant.components.lock import LockDevice
from homeassistant.const import ATTR_BATTERY_LEVEL, STATE_LOCKED, STATE_UNLOCKED

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

@ -3,7 +3,7 @@
"name": "HomeKit Controller",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/homekit_controller",
"requirements": ["homekit[IP]==0.15.0"],
"requirements": ["aiohomekit[IP]==0.2.10"],
"dependencies": [],
"zeroconf": ["_hap._tcp.local."],
"codeowners": ["@Jc2k"]

2
homeassistant/components/homekit_controller/sensor.py
View File

@ -1,5 +1,5 @@
"""Support for Homekit sensors."""
from homekit.model.characteristics import CharacteristicsTypes
from aiohomekit.model.characteristics import CharacteristicsTypes
from homeassistant.const import DEVICE_CLASS_BATTERY, TEMP_CELSIUS
from homeassistant.core import callback

2
homeassistant/components/homekit_controller/switch.py
View File

@ -1,7 +1,7 @@
"""Support for Homekit switches."""
import logging
from homekit.model.characteristics import CharacteristicsTypes
from aiohomekit.model.characteristics import CharacteristicsTypes
from homeassistant.components.switch import SwitchDevice
from homeassistant.core import callback

6
requirements_all.txt
View File

@ -161,6 +161,9 @@ aioftp==0.12.0
# homeassistant.components.harmony
aioharmony==0.1.13
# homeassistant.components.homekit_controller
aiohomekit[IP]==0.2.10
# homeassistant.components.emulated_hue
# homeassistant.components.http
aiohttp_cors==0.7.0
@ -688,9 +691,6 @@ home-assistant-frontend==20200220.1
# homeassistant.components.zwave
homeassistant-pyozw==0.1.8
# homeassistant.components.homekit_controller
homekit[IP]==0.15.0
# homeassistant.components.homematicip_cloud
homematicip==0.10.17

6
requirements_test_all.txt
View File

@ -61,6 +61,9 @@ aiobotocore==0.11.1
# homeassistant.components.esphome
aioesphomeapi==2.6.1
# homeassistant.components.homekit_controller
aiohomekit[IP]==0.2.10
# homeassistant.components.emulated_hue
# homeassistant.components.http
aiohttp_cors==0.7.0
@ -259,9 +262,6 @@ home-assistant-frontend==20200220.1
# homeassistant.components.zwave
homeassistant-pyozw==0.1.8
# homeassistant.components.homekit_controller
homekit[IP]==0.15.0
# homeassistant.components.homematicip_cloud
homematicip==0.10.17

104
tests/components/homekit_controller/common.py
View File

@ -4,14 +4,10 @@ import json
import os
from unittest import mock
from homekit.exceptions import AccessoryNotFoundError
from homekit.model import Accessory, get_id
from homekit.model.characteristics import (
AbstractCharacteristic,
CharacteristicPermissions,
CharacteristicsTypes,
)
from homekit.model.services import AbstractService, ServicesTypes
from aiohomekit.exceptions import AccessoryNotFoundError
from aiohomekit.model import Accessory
from aiohomekit.model.characteristics import CharacteristicsTypes
from aiohomekit.model.services import ServicesTypes
from homeassistant import config_entries
from homeassistant.components.homekit_controller import config_flow
@ -40,14 +36,14 @@ class FakePairing:
self.pairing_data = {}
self.available = True
def list_accessories_and_characteristics(self):
async def list_accessories_and_characteristics(self):
"""Fake implementation of list_accessories_and_characteristics."""
accessories = [a.to_accessory_and_service_list() for a in self.accessories]
# replicate what happens upstream right now
self.pairing_data["accessories"] = accessories
return accessories
def get_characteristics(self, characteristics):
async def get_characteristics(self, characteristics):
"""Fake implementation of get_characteristics."""
if not self.available:
raise AccessoryNotFoundError("Accessory not found")
@ -64,7 +60,7 @@ class FakePairing:
results[(aid, cid)] = {"value": char.get_value()}
return results
def put_characteristics(self, characteristics):
async def put_characteristics(self, characteristics):
"""Fake implementation of put_characteristics."""
for aid, cid, new_val in characteristics:
for accessory in self.accessories:
@ -124,45 +120,6 @@ class Helper:
return state
class FakeCharacteristic(AbstractCharacteristic):
"""
A model of a generic HomeKit characteristic.
Base is abstract and can't be instanced directly so this subclass is
needed even though it doesn't add any methods.
"""
def to_accessory_and_service_list(self):
"""Serialize the characteristic."""
# Upstream doesn't correctly serialize valid_values
# This fix will be upstreamed and this function removed when it
# is fixed.
record = super().to_accessory_and_service_list()
if self.valid_values:
record["valid-values"] = self.valid_values
return record
class FakeService(AbstractService):
"""A model of a generic HomeKit service."""
def __init__(self, service_name):
"""Create a fake service by its short form HAP spec name."""
char_type = ServicesTypes.get_uuid(service_name)
super().__init__(char_type, get_id())
def add_characteristic(self, name):
"""Add a characteristic to this service by name."""
full_name = "public.hap.characteristic." + name
char = FakeCharacteristic(get_id(), full_name, None)
char.perms = [
CharacteristicPermissions.paired_read,
CharacteristicPermissions.paired_write,
]
self.characteristics.append(char)
return char
async def time_changed(hass, seconds):
"""Trigger time changed."""
next_update = dt_util.utcnow() + timedelta(seconds)
@ -176,40 +133,7 @@ async def setup_accessories_from_file(hass, path):
load_fixture, os.path.join("homekit_controller", path)
)
accessories_json = json.loads(accessories_fixture)
accessories = []
for accessory_data in accessories_json:
accessory = Accessory("Name", "Mfr", "Model", "0001", "0.1")
accessory.services = []
accessory.aid = accessory_data["aid"]
for service_data in accessory_data["services"]:
service = FakeService("public.hap.service.accessory-information")
service.type = service_data["type"]
service.iid = service_data["iid"]
for char_data in service_data["characteristics"]:
char = FakeCharacteristic(1, "23", None)
char.type = char_data["type"]
char.iid = char_data["iid"]
char.perms = char_data["perms"]
char.format = char_data["format"]
if "description" in char_data:
char.description = char_data["description"]
if "value" in char_data:
char.value = char_data["value"]
if "minValue" in char_data:
char.minValue = char_data["minValue"]
if "maxValue" in char_data:
char.maxValue = char_data["maxValue"]
if "valid-values" in char_data:
char.valid_values = char_data["valid-values"]
service.characteristics.append(char)
accessory.services.append(service)
accessories.append(accessory)
accessories = Accessory.setup_accessories_from_list(accessories_json)
return accessories
@ -217,7 +141,7 @@ async def setup_platform(hass):
"""Load the platform but with a fake Controller API."""
config = {"discovery": {}}
with mock.patch("homekit.Controller") as controller:
with mock.patch("aiohomekit.Controller") as controller:
fake_controller = controller.return_value = FakeController()
await async_setup_component(hass, DOMAIN, config)
@ -293,15 +217,18 @@ async def device_config_changed(hass, accessories):
await hass.async_block_till_done()
async def setup_test_component(hass, services, capitalize=False, suffix=None):
async def setup_test_component(hass, setup_accessory, capitalize=False, suffix=None):
"""Load a fake homekit accessory based on a homekit accessory model.
If capitalize is True, property names will be in upper case.
If suffix is set, entityId will include the suffix
"""
accessory = Accessory("TestDevice", "example.com", "Test", "0001", "0.1")
setup_accessory(accessory)
domain = None
for service in services:
for service in accessory.services:
service_name = ServicesTypes.get_short(service.type)
if service_name in HOMEKIT_ACCESSORY_DISPATCH:
domain = HOMEKIT_ACCESSORY_DISPATCH[service_name]
@ -309,9 +236,6 @@ async def setup_test_component(hass, services, capitalize=False, suffix=None):
assert domain, "Cannot map test homekit services to Home Assistant domain"
accessory = Accessory("TestDevice", "example.com", "Test", "0001", "0.1")
accessory.services.extend(services)
config_entry, pairing = await setup_test_accessories(hass, [accessory])
entity = "testdevice" if suffix is None else "testdevice_{}".format(suffix)
return Helper(hass, ".".join((domain, entity)), pairing, accessory, config_entry)

2
tests/components/homekit_controller/specific_devices/test_ecobee3.py
View File

@ -6,7 +6,7 @@ https://github.com/home-assistant/home-assistant/issues/15336
from unittest import mock
from homekit import AccessoryDisconnectedError
from aiohomekit import AccessoryDisconnectedError
from homeassistant.components.climate.const import (
SUPPORT_TARGET_HUMIDITY,

2
tests/components/homekit_controller/specific_devices/test_koogeek_ls1.py
View File

@ -3,7 +3,7 @@
from datetime import timedelta
from unittest import mock
from homekit.exceptions import AccessoryDisconnectedError, EncryptionError
from aiohomekit.exceptions import AccessoryDisconnectedError, EncryptionError
import pytest
from homeassistant.components.light import SUPPORT_BRIGHTNESS, SUPPORT_COLOR

28
tests/components/homekit_controller/test_air_quality.py
View File

@ -1,39 +1,39 @@
"""Basic checks for HomeKit air quality sensor."""
from tests.components.homekit_controller.common import FakeService, setup_test_component
from aiohomekit.model.characteristics import CharacteristicsTypes
from aiohomekit.model.services import ServicesTypes
from tests.components.homekit_controller.common import setup_test_component
def create_air_quality_sensor_service():
def create_air_quality_sensor_service(accessory):
"""Define temperature characteristics."""
service = FakeService("public.hap.service.sensor.air-quality")
service = accessory.add_service(ServicesTypes.AIR_QUALITY_SENSOR)
cur_state = service.add_characteristic("air-quality")
cur_state = service.add_char(CharacteristicsTypes.AIR_QUALITY)
cur_state.value = 5
cur_state = service.add_characteristic("density.ozone")
cur_state = service.add_char(CharacteristicsTypes.DENSITY_OZONE)
cur_state.value = 1111
cur_state = service.add_characteristic("density.no2")
cur_state = service.add_char(CharacteristicsTypes.DENSITY_NO2)
cur_state.value = 2222
cur_state = service.add_characteristic("density.so2")
cur_state = service.add_char(CharacteristicsTypes.DENSITY_SO2)
cur_state.value = 3333
cur_state = service.add_characteristic("density.pm25")
cur_state = service.add_char(CharacteristicsTypes.DENSITY_PM25)
cur_state.value = 4444
cur_state = service.add_characteristic("density.pm10")
cur_state = service.add_char(CharacteristicsTypes.DENSITY_PM10)
cur_state.value = 5555
cur_state = service.add_characteristic("density.voc")
cur_state = service.add_char(CharacteristicsTypes.DENSITY_VOC)
cur_state.value = 6666
return service
async def test_air_quality_sensor_read_state(hass, utcnow):
"""Test reading the state of a HomeKit temperature sensor accessory."""
sensor = create_air_quality_sensor_service()
helper = await setup_test_component(hass, [sensor])
helper = await setup_test_component(hass, create_air_quality_sensor_service)
state = await helper.poll_and_get_state()
assert state.state == "4444"

23
tests/components/homekit_controller/test_alarm_control_panel.py
View File

@ -1,34 +1,34 @@
"""Basic checks for HomeKitalarm_control_panel."""
from tests.components.homekit_controller.common import FakeService, setup_test_component
from aiohomekit.model.characteristics import CharacteristicsTypes
from aiohomekit.model.services import ServicesTypes
from tests.components.homekit_controller.common import setup_test_component
CURRENT_STATE = ("security-system", "security-system-state.current")
TARGET_STATE = ("security-system", "security-system-state.target")
def create_security_system_service():
def create_security_system_service(accessory):
"""Define a security-system characteristics as per page 219 of HAP spec."""
service = FakeService("public.hap.service.security-system")
service = accessory.add_service(ServicesTypes.SECURITY_SYSTEM)
cur_state = service.add_characteristic("security-system-state.current")
cur_state = service.add_char(CharacteristicsTypes.SECURITY_SYSTEM_STATE_CURRENT)
cur_state.value = 0
targ_state = service.add_characteristic("security-system-state.target")
targ_state = service.add_char(CharacteristicsTypes.SECURITY_SYSTEM_STATE_TARGET)
targ_state.value = 0
# According to the spec, a battery-level characteristic is normally
# part of a separate service. However as the code was written (which
# predates this test) the battery level would have to be part of the lock
# service as it is here.
targ_state = service.add_characteristic("battery-level")
targ_state = service.add_char(CharacteristicsTypes.BATTERY_LEVEL)
targ_state.value = 50
return service
async def test_switch_change_alarm_state(hass, utcnow):
"""Test that we can turn a HomeKit alarm on and off again."""
alarm_control_panel = create_security_system_service()
helper = await setup_test_component(hass, [alarm_control_panel])
helper = await setup_test_component(hass, create_security_system_service)
await hass.services.async_call(
"alarm_control_panel",
@ -65,8 +65,7 @@ async def test_switch_change_alarm_state(hass, utcnow):
async def test_switch_read_alarm_state(hass, utcnow):
"""Test that we can read the state of a HomeKit alarm accessory."""
alarm_control_panel = create_security_system_service()
helper = await setup_test_component(hass, [alarm_control_panel])
helper = await setup_test_component(hass, create_security_system_service)
helper.characteristics[CURRENT_STATE].value = 0
state = await helper.poll_and_get_state()

38
tests/components/homekit_controller/test_binary_sensor.py
View File

@ -1,25 +1,25 @@
"""Basic checks for HomeKit motion sensors and contact sensors."""
from tests.components.homekit_controller.common import FakeService, setup_test_component
from aiohomekit.model.characteristics import CharacteristicsTypes
from aiohomekit.model.services import ServicesTypes
from tests.components.homekit_controller.common import setup_test_component
MOTION_DETECTED = ("motion", "motion-detected")
CONTACT_STATE = ("contact", "contact-state")
SMOKE_DETECTED = ("smoke", "smoke-detected")
def create_motion_sensor_service():
def create_motion_sensor_service(accessory):
"""Define motion characteristics as per page 225 of HAP spec."""
service = FakeService("public.hap.service.sensor.motion")
service = accessory.add_service(ServicesTypes.MOTION_SENSOR)
cur_state = service.add_characteristic("motion-detected")
cur_state = service.add_char(CharacteristicsTypes.MOTION_DETECTED)
cur_state.value = 0
return service
async def test_motion_sensor_read_state(hass, utcnow):
"""Test that we can read the state of a HomeKit motion sensor accessory."""
sensor = create_motion_sensor_service()
helper = await setup_test_component(hass, [sensor])
helper = await setup_test_component(hass, create_motion_sensor_service)
helper.characteristics[MOTION_DETECTED].value = False
state = await helper.poll_and_get_state()
@ -30,20 +30,17 @@ async def test_motion_sensor_read_state(hass, utcnow):
assert state.state == "on"
def create_contact_sensor_service():
def create_contact_sensor_service(accessory):
"""Define contact characteristics."""
service = FakeService("public.hap.service.sensor.contact")
service = accessory.add_service(ServicesTypes.CONTACT_SENSOR)
cur_state = service.add_characteristic("contact-state")
cur_state = service.add_char(CharacteristicsTypes.CONTACT_STATE)
cur_state.value = 0
return service
async def test_contact_sensor_read_state(hass, utcnow):
"""Test that we can read the state of a HomeKit contact accessory."""
sensor = create_contact_sensor_service()
helper = await setup_test_component(hass, [sensor])
helper = await setup_test_component(hass, create_contact_sensor_service)
helper.characteristics[CONTACT_STATE].value = 0
state = await helper.poll_and_get_state()
@ -54,20 +51,17 @@ async def test_contact_sensor_read_state(hass, utcnow):
assert state.state == "on"
def create_smoke_sensor_service():
def create_smoke_sensor_service(accessory):
"""Define smoke sensor characteristics."""
service = FakeService("public.hap.service.sensor.smoke")
service = accessory.add_service(ServicesTypes.SMOKE_SENSOR)
cur_state = service.add_characteristic("smoke-detected")
cur_state = service.add_char(CharacteristicsTypes.SMOKE_DETECTED)
cur_state.value = 0
return service
async def test_smoke_sensor_read_state(hass, utcnow):
"""Test that we can read the state of a HomeKit contact accessory."""
sensor = create_smoke_sensor_service()
helper = await setup_test_component(hass, [sensor])
helper = await setup_test_component(hass, create_smoke_sensor_service)
helper.characteristics[SMOKE_DETECTED].value = 0
state = await helper.poll_and_get_state()

63
tests/components/homekit_controller/test_climate.py
View File

@ -1,4 +1,7 @@
"""Basic checks for HomeKitclimate."""
from aiohomekit.model.characteristics import CharacteristicsTypes
from aiohomekit.model.services import ServicesTypes
from homeassistant.components.climate.const import (
DOMAIN,
HVAC_MODE_COOL,
@ -10,7 +13,7 @@ from homeassistant.components.climate.const import (
SERVICE_SET_TEMPERATURE,
)
from tests.components.homekit_controller.common import FakeService, setup_test_component
from tests.components.homekit_controller.common import setup_test_component
HEATING_COOLING_TARGET = ("thermostat", "heating-cooling.target")
HEATING_COOLING_CURRENT = ("thermostat", "heating-cooling.current")
@ -20,63 +23,65 @@ HUMIDITY_TARGET = ("thermostat", "relative-humidity.target")
HUMIDITY_CURRENT = ("thermostat", "relative-humidity.current")
def create_thermostat_service():
def create_thermostat_service(accessory):
"""Define thermostat characteristics."""
service = FakeService("public.hap.service.thermostat")
service = accessory.add_service(ServicesTypes.THERMOSTAT)
char = service.add_characteristic("heating-cooling.target")
char = service.add_char(CharacteristicsTypes.HEATING_COOLING_TARGET)
char.value = 0
char = service.add_characteristic("heating-cooling.current")
char = service.add_char(CharacteristicsTypes.HEATING_COOLING_CURRENT)
char.value = 0
char = service.add_characteristic("temperature.target")
char = service.add_char(CharacteristicsTypes.TEMPERATURE_TARGET)
char.minValue = 7
char.maxValue = 35
char.value = 0
char = service.add_characteristic("temperature.current")
char = service.add_char(CharacteristicsTypes.TEMPERATURE_CURRENT)
char.value = 0
char = service.add_characteristic("relative-humidity.target")
char = service.add_char(CharacteristicsTypes.RELATIVE_HUMIDITY_TARGET)
char.value = 0
char = service.add_characteristic("relative-humidity.current")
char = service.add_char(CharacteristicsTypes.RELATIVE_HUMIDITY_CURRENT)
char.value = 0
return service
async def test_climate_respect_supported_op_modes_1(hass, utcnow):
"""Test that climate respects minValue/maxValue hints."""
service = FakeService("public.hap.service.thermostat")
char = service.add_characteristic("heating-cooling.target")
def create_thermostat_service_min_max(accessory):
"""Define thermostat characteristics."""
service = accessory.add_service(ServicesTypes.THERMOSTAT)
char = service.add_char(CharacteristicsTypes.HEATING_COOLING_TARGET)
char.value = 0
char.minValue = 0
char.maxValue = 1
helper = await setup_test_component(hass, [service])
async def test_climate_respect_supported_op_modes_1(hass, utcnow):
"""Test that climate respects minValue/maxValue hints."""
helper = await setup_test_component(hass, create_thermostat_service_min_max)
state = await helper.poll_and_get_state()
assert state.attributes["hvac_modes"] == ["off", "heat"]
async def test_climate_respect_supported_op_modes_2(hass, utcnow):
"""Test that climate respects validValue hints."""
service = FakeService("public.hap.service.thermostat")
char = service.add_characteristic("heating-cooling.target")
def create_thermostat_service_valid_vals(accessory):
"""Define thermostat characteristics."""
service = accessory.add_service(ServicesTypes.THERMOSTAT)
char = service.add_char(CharacteristicsTypes.HEATING_COOLING_TARGET)
char.value = 0
char.valid_values = [0, 1, 2]
helper = await setup_test_component(hass, [service])
async def test_climate_respect_supported_op_modes_2(hass, utcnow):
"""Test that climate respects validValue hints."""
helper = await setup_test_component(hass, create_thermostat_service_valid_vals)
state = await helper.poll_and_get_state()
assert state.attributes["hvac_modes"] == ["off", "heat", "cool"]
async def test_climate_change_thermostat_state(hass, utcnow):
"""Test that we can turn a HomeKit thermostat on and off again."""
from homekit.model.services import ThermostatService
helper = await setup_test_component(hass, [ThermostatService()])
helper = await setup_test_component(hass, create_thermostat_service)
await hass.services.async_call(
DOMAIN,
@ -114,9 +119,7 @@ async def test_climate_change_thermostat_state(hass, utcnow):
async def test_climate_change_thermostat_temperature(hass, utcnow):
"""Test that we can turn a HomeKit thermostat on and off again."""
from homekit.model.services import ThermostatService
helper = await setup_test_component(hass, [ThermostatService()])
helper = await setup_test_component(hass, create_thermostat_service)
await hass.services.async_call(
DOMAIN,
@ -137,7 +140,7 @@ async def test_climate_change_thermostat_temperature(hass, utcnow):
async def test_climate_change_thermostat_humidity(hass, utcnow):
"""Test that we can turn a HomeKit thermostat on and off again."""
helper = await setup_test_component(hass, [create_thermostat_service()])
helper = await setup_test_component(hass, create_thermostat_service)
await hass.services.async_call(
DOMAIN,
@ -158,7 +161,7 @@ async def test_climate_change_thermostat_humidity(hass, utcnow):
async def test_climate_read_thermostat_state(hass, utcnow):
"""Test that we can read the state of a HomeKit thermostat accessory."""
helper = await setup_test_component(hass, [create_thermostat_service()])
helper = await setup_test_component(hass, create_thermostat_service)
# Simulate that heating is on
helper.characteristics[TEMPERATURE_CURRENT].value = 19
@ -200,7 +203,7 @@ async def test_climate_read_thermostat_state(hass, utcnow):
async def test_hvac_mode_vs_hvac_action(hass, utcnow):
"""Check that we haven't conflated hvac_mode and hvac_action."""
helper = await setup_test_component(hass, [create_thermostat_service()])
helper = await setup_test_component(hass, create_thermostat_service)
# Simulate that current temperature is above target temp
# Heating might be on, but hvac_action currently 'off'

383
tests/components/homekit_controller/test_config_flow.py
View File

@ -2,40 +2,39 @@
import json
from unittest import mock
import homekit
import aiohomekit
from aiohomekit.model.characteristics import CharacteristicsTypes
from aiohomekit.model.services import ServicesTypes
import asynctest
import pytest
from homeassistant.components.homekit_controller import config_flow
from homeassistant.components.homekit_controller.const import KNOWN_DEVICES
from tests.common import MockConfigEntry
from tests.components.homekit_controller.common import (
Accessory,
FakeService,
setup_platform,
)
from tests.components.homekit_controller.common import Accessory, setup_platform
PAIRING_START_FORM_ERRORS = [
(homekit.BusyError, "busy_error"),
(homekit.MaxTriesError, "max_tries_error"),
(aiohomekit.BusyError, "busy_error"),
(aiohomekit.MaxTriesError, "max_tries_error"),
(KeyError, "pairing_failed"),
]
PAIRING_START_ABORT_ERRORS = [
(homekit.AccessoryNotFoundError, "accessory_not_found_error"),
(homekit.UnavailableError, "already_paired"),
(aiohomekit.AccessoryNotFoundError, "accessory_not_found_error"),
(aiohomekit.UnavailableError, "already_paired"),
]
PAIRING_FINISH_FORM_ERRORS = [
(homekit.exceptions.MalformedPinError, "authentication_error"),
(homekit.MaxPeersError, "max_peers_error"),
(homekit.AuthenticationError, "authentication_error"),
(homekit.UnknownError, "unknown_error"),
(aiohomekit.exceptions.MalformedPinError, "authentication_error"),
(aiohomekit.MaxPeersError, "max_peers_error"),
(aiohomekit.AuthenticationError, "authentication_error"),
(aiohomekit.UnknownError, "unknown_error"),
(KeyError, "pairing_failed"),
]
PAIRING_FINISH_ABORT_ERRORS = [
(homekit.AccessoryNotFoundError, "accessory_not_found_error")
(aiohomekit.AccessoryNotFoundError, "accessory_not_found_error")
]
INVALID_PAIRING_CODES = [
@ -60,13 +59,22 @@ VALID_PAIRING_CODES = [
]
def _setup_flow_handler(hass):
def _setup_flow_handler(hass, pairing=None):
flow = config_flow.HomekitControllerFlowHandler()
flow.hass = hass
flow.context = {}
finish_pairing = asynctest.CoroutineMock(return_value=pairing)
discovery = mock.Mock()
discovery.device_id = "00:00:00:00:00:00"
discovery.start_pairing = asynctest.CoroutineMock(return_value=finish_pairing)
flow.controller = mock.Mock()
flow.controller.pairings = {}
flow.controller.find_ip_by_device_id = asynctest.CoroutineMock(
return_value=discovery
)
return flow
@ -81,7 +89,7 @@ async def _setup_flow_zeroconf(hass, discovery_info):
@pytest.mark.parametrize("pairing_code", INVALID_PAIRING_CODES)
def test_invalid_pairing_codes(pairing_code):
"""Test ensure_pin_format raises for an invalid pin code."""
with pytest.raises(homekit.exceptions.MalformedPinError):
with pytest.raises(aiohomekit.exceptions.MalformedPinError):
config_flow.ensure_pin_format(pairing_code)
@ -106,6 +114,15 @@ async def test_discovery_works(hass):
flow = _setup_flow_handler(hass)
finish_pairing = asynctest.CoroutineMock()
discovery = mock.Mock()
discovery.start_pairing = asynctest.CoroutineMock(return_value=finish_pairing)
flow.controller.find_ip_by_device_id = asynctest.CoroutineMock(
return_value=discovery
)
# Device is discovered
result = await flow.async_step_zeroconf(discovery_info)
assert result["type"] == "form"
@ -120,21 +137,27 @@ async def test_discovery_works(hass):
result = await flow.async_step_pair({})
assert result["type"] == "form"
assert result["step_id"] == "pair"
assert flow.controller.start_pairing.call_count == 1
assert discovery.start_pairing.call_count == 1
pairing = mock.Mock(pairing_data={"AccessoryPairingID": "00:00:00:00:00:00"})
pairing.list_accessories_and_characteristics.return_value = [
{
"aid": 1,
"services": [
{
"characteristics": [{"type": "23", "value": "Koogeek-LS1-20833F"}],
"type": "3e",
}
],
}
]
pairing.list_accessories_and_characteristics = asynctest.CoroutineMock(
return_value=[
{
"aid": 1,
"services": [
{
"characteristics": [
{"type": "23", "value": "Koogeek-LS1-20833F"}
],
"type": "3e",
}
],
}
]
)
finish_pairing.return_value = pairing
# Pairing doesn't error error and pairing results
flow.controller.pairings = {"00:00:00:00:00:00": pairing}
@ -155,6 +178,15 @@ async def test_discovery_works_upper_case(hass):
flow = _setup_flow_handler(hass)
finish_pairing = asynctest.CoroutineMock()
discovery = mock.Mock()
discovery.start_pairing = asynctest.CoroutineMock(return_value=finish_pairing)
flow.controller.find_ip_by_device_id = asynctest.CoroutineMock(
return_value=discovery
)
# Device is discovered
result = await flow.async_step_zeroconf(discovery_info)
assert result["type"] == "form"
@ -169,21 +201,27 @@ async def test_discovery_works_upper_case(hass):
result = await flow.async_step_pair({})
assert result["type"] == "form"
assert result["step_id"] == "pair"
assert flow.controller.start_pairing.call_count == 1
assert discovery.start_pairing.call_count == 1
pairing = mock.Mock(pairing_data={"AccessoryPairingID": "00:00:00:00:00:00"})
pairing.list_accessories_and_characteristics.return_value = [
{
"aid": 1,
"services": [
{
"characteristics": [{"type": "23", "value": "Koogeek-LS1-20833F"}],
"type": "3e",
}
],
}
]
pairing.list_accessories_and_characteristics = asynctest.CoroutineMock(
return_value=[
{
"aid": 1,
"services": [
{
"characteristics": [
{"type": "23", "value": "Koogeek-LS1-20833F"}
],
"type": "3e",
}
],
}
]
)
finish_pairing.return_value = pairing
flow.controller.pairings = {"00:00:00:00:00:00": pairing}
result = await flow.async_step_pair({"pairing_code": "111-22-333"})
@ -203,6 +241,15 @@ async def test_discovery_works_missing_csharp(hass):
flow = _setup_flow_handler(hass)
finish_pairing = asynctest.CoroutineMock()
discovery = mock.Mock()
discovery.start_pairing = asynctest.CoroutineMock(return_value=finish_pairing)
flow.controller.find_ip_by_device_id = asynctest.CoroutineMock(
return_value=discovery
)
# Device is discovered
result = await flow.async_step_zeroconf(discovery_info)
assert result["type"] == "form"
@ -217,21 +264,27 @@ async def test_discovery_works_missing_csharp(hass):
result = await flow.async_step_pair({})
assert result["type"] == "form"
assert result["step_id"] == "pair"
assert flow.controller.start_pairing.call_count == 1
assert discovery.start_pairing.call_count == 1
pairing = mock.Mock(pairing_data={"AccessoryPairingID": "00:00:00:00:00:00"})
pairing.list_accessories_and_characteristics.return_value = [
{
"aid": 1,
"services": [
{
"characteristics": [{"type": "23", "value": "Koogeek-LS1-20833F"}],
"type": "3e",
}
],
}
]
pairing.list_accessories_and_characteristics = asynctest.CoroutineMock(
return_value=[
{
"aid": 1,
"services": [
{
"characteristics": [
{"type": "23", "value": "Koogeek-LS1-20833F"}
],
"type": "3e",
}
],
}
]
)
finish_pairing.return_value = pairing
flow.controller.pairings = {"00:00:00:00:00:00": pairing}
@ -390,39 +443,6 @@ async def test_discovery_already_configured_config_change(hass):
assert conn.async_refresh_entity_map.call_args == mock.call(2)
async def test_pair_unable_to_pair(hass):
"""Pairing completed without exception, but didn't create a pairing."""
discovery_info = {
"name": "TestDevice",
"host": "127.0.0.1",
"port": 8080,
"properties": {"md": "TestDevice", "id": "00:00:00:00:00:00", "c#": 1, "sf": 1},
}
flow = _setup_flow_handler(hass)
# Device is discovered
result = await flow.async_step_zeroconf(discovery_info)
assert result["type"] == "form"
assert result["step_id"] == "pair"
assert flow.context == {
"hkid": "00:00:00:00:00:00",
"title_placeholders": {"name": "TestDevice"},
"unique_id": "00:00:00:00:00:00",
}
# User initiates pairing - device enters pairing mode and displays code
result = await flow.async_step_pair({})
assert result["type"] == "form"
assert result["step_id"] == "pair"
assert flow.controller.start_pairing.call_count == 1
# Pairing doesn't error but no pairing object is generated
result = await flow.async_step_pair({"pairing_code": "111-22-333"})
assert result["type"] == "form"
assert result["errors"]["pairing_code"] == "unable_to_pair"
@pytest.mark.parametrize("exception,expected", PAIRING_START_ABORT_ERRORS)
async def test_pair_abort_errors_on_start(hass, exception, expected):
"""Test various pairing errors."""
@ -435,6 +455,13 @@ async def test_pair_abort_errors_on_start(hass, exception, expected):
flow = _setup_flow_handler(hass)
discovery = mock.Mock()
discovery.start_pairing = asynctest.CoroutineMock(side_effect=exception("error"))
flow.controller.find_ip_by_device_id = asynctest.CoroutineMock(
return_value=discovery
)
# Device is discovered
result = await flow.async_step_zeroconf(discovery_info)
assert result["type"] == "form"
@ -446,10 +473,7 @@ async def test_pair_abort_errors_on_start(hass, exception, expected):
}
# User initiates pairing - device refuses to enter pairing mode
with mock.patch.object(flow.controller, "start_pairing") as start_pairing:
start_pairing.side_effect = exception("error")
result = await flow.async_step_pair({})
result = await flow.async_step_pair({})
assert result["type"] == "abort"
assert result["reason"] == expected
assert flow.context == {
@ -471,6 +495,13 @@ async def test_pair_form_errors_on_start(hass, exception, expected):
flow = _setup_flow_handler(hass)
discovery = mock.Mock()
discovery.start_pairing = asynctest.CoroutineMock(side_effect=exception("error"))
flow.controller.find_ip_by_device_id = asynctest.CoroutineMock(
return_value=discovery
)
# Device is discovered
result = await flow.async_step_zeroconf(discovery_info)
assert result["type"] == "form"
@ -482,10 +513,7 @@ async def test_pair_form_errors_on_start(hass, exception, expected):
}
# User initiates pairing - device refuses to enter pairing mode
with mock.patch.object(flow.controller, "start_pairing") as start_pairing:
start_pairing.side_effect = exception("error")
result = await flow.async_step_pair({})
result = await flow.async_step_pair({})
assert result["type"] == "form"
assert result["errors"]["pairing_code"] == expected
assert flow.context == {
@ -507,6 +535,15 @@ async def test_pair_abort_errors_on_finish(hass, exception, expected):
flow = _setup_flow_handler(hass)
finish_pairing = asynctest.CoroutineMock(side_effect=exception("error"))
discovery = mock.Mock()
discovery.start_pairing = asynctest.CoroutineMock(return_value=finish_pairing)
flow.controller.find_ip_by_device_id = asynctest.CoroutineMock(
return_value=discovery
)
# Device is discovered
result = await flow.async_step_zeroconf(discovery_info)
assert result["type"] == "form"
@ -521,10 +558,9 @@ async def test_pair_abort_errors_on_finish(hass, exception, expected):
result = await flow.async_step_pair({})
assert result["type"] == "form"
assert result["step_id"] == "pair"
assert flow.controller.start_pairing.call_count == 1
assert discovery.start_pairing.call_count == 1
# User submits code - pairing fails but can be retried
flow.finish_pairing.side_effect = exception("error")
result = await flow.async_step_pair({"pairing_code": "111-22-333"})
assert result["type"] == "abort"
assert result["reason"] == expected
@ -547,6 +583,15 @@ async def test_pair_form_errors_on_finish(hass, exception, expected):
flow = _setup_flow_handler(hass)
finish_pairing = asynctest.CoroutineMock(side_effect=exception("error"))
discovery = mock.Mock()
discovery.start_pairing = asynctest.CoroutineMock(return_value=finish_pairing)
flow.controller.find_ip_by_device_id = asynctest.CoroutineMock(
return_value=discovery
)
# Device is discovered
result = await flow.async_step_zeroconf(discovery_info)
assert result["type"] == "form"
@ -561,10 +606,9 @@ async def test_pair_form_errors_on_finish(hass, exception, expected):
result = await flow.async_step_pair({})
assert result["type"] == "form"
assert result["step_id"] == "pair"
assert flow.controller.start_pairing.call_count == 1
assert discovery.start_pairing.call_count == 1
# User submits code - pairing fails but can be retried
flow.finish_pairing.side_effect = exception("error")
result = await flow.async_step_pair({"pairing_code": "111-22-333"})
assert result["type"] == "form"
assert result["errors"]["pairing_code"] == expected
@ -588,17 +632,21 @@ async def test_import_works(hass):
pairing = mock.Mock(pairing_data={"AccessoryPairingID": "00:00:00:00:00:00"})
pairing.list_accessories_and_characteristics.return_value = [
{
"aid": 1,
"services": [
{
"characteristics": [{"type": "23", "value": "Koogeek-LS1-20833F"}],
"type": "3e",
}
],
}
]
pairing.list_accessories_and_characteristics = asynctest.CoroutineMock(
return_value=[
{
"aid": 1,
"services": [
{
"characteristics": [
{"type": "23", "value": "Koogeek-LS1-20833F"}
],
"type": "3e",
}
],
}
]
)
flow = _setup_flow_handler(hass)
@ -653,22 +701,35 @@ async def test_user_works(hass):
}
pairing = mock.Mock(pairing_data={"AccessoryPairingID": "00:00:00:00:00:00"})
pairing.list_accessories_and_characteristics.return_value = [
{
"aid": 1,
"services": [
{
"characteristics": [{"type": "23", "value": "Koogeek-LS1-20833F"}],
"type": "3e",
}
],
}
]
pairing.list_accessories_and_characteristics = asynctest.CoroutineMock(
return_value=[
{
"aid": 1,
"services": [
{
"characteristics": [
{"type": "23", "value": "Koogeek-LS1-20833F"}
],
"type": "3e",
}
],
}
]
)
flow = _setup_flow_handler(hass)
finish_pairing = asynctest.CoroutineMock(return_value=pairing)
discovery = mock.Mock()
discovery.info = discovery_info
discovery.start_pairing = asynctest.CoroutineMock(return_value=finish_pairing)
flow.controller.pairings = {"00:00:00:00:00:00": pairing}
flow.controller.discover.return_value = [discovery_info]
flow.controller.discover_ip = asynctest.CoroutineMock(return_value=[discovery])
flow.controller.find_ip_by_device_id = asynctest.CoroutineMock(
return_value=discovery
)
result = await flow.async_step_user()
assert result["type"] == "form"
@ -688,7 +749,7 @@ async def test_user_no_devices(hass):
"""Test user initiated pairing where no devices discovered."""
flow = _setup_flow_handler(hass)
flow.controller.discover.return_value = []
flow.controller.discover_ip = asynctest.CoroutineMock(return_value=[])
result = await flow.async_step_user()
assert result["type"] == "abort"
@ -709,7 +770,10 @@ async def test_user_no_unpaired_devices(hass):
"sf": 0,
}
flow.controller.discover.return_value = [discovery_info]
discovery = mock.Mock()
discovery.info = discovery_info
flow.controller.discover_ip = asynctest.CoroutineMock(return_value=[discovery])
result = await flow.async_step_user()
assert result["type"] == "abort"
@ -718,12 +782,10 @@ async def test_user_no_unpaired_devices(hass):
async def test_parse_new_homekit_json(hass):
"""Test migrating recent .homekit/pairings.json files."""
service = FakeService("public.hap.service.lightbulb")
on_char = service.add_characteristic("on")
on_char.value = 1
accessory = Accessory("TestDevice", "example.com", "Test", "0001", "0.1")
accessory.services.append(service)
service = accessory.add_service(ServicesTypes.LIGHTBULB)
on_char = service.add_char(CharacteristicsTypes.ON)
on_char.value = 0
fake_controller = await setup_platform(hass)
pairing = fake_controller.add([accessory])
@ -766,12 +828,10 @@ async def test_parse_new_homekit_json(hass):
async def test_parse_old_homekit_json(hass):
"""Test migrating original .homekit/hk-00:00:00:00:00:00 files."""
service = FakeService("public.hap.service.lightbulb")
on_char = service.add_characteristic("on")
on_char.value = 1
accessory = Accessory("TestDevice", "example.com", "Test", "0001", "0.1")
accessory.services.append(service)
service = accessory.add_service(ServicesTypes.LIGHTBULB)
on_char = service.add_char(CharacteristicsTypes.ON)
on_char.value = 0
fake_controller = await setup_platform(hass)
pairing = fake_controller.add([accessory])
@ -818,12 +878,10 @@ async def test_parse_old_homekit_json(hass):
async def test_parse_overlapping_homekit_json(hass):
"""Test migrating .homekit/pairings.json files when hk- exists too."""
service = FakeService("public.hap.service.lightbulb")
on_char = service.add_characteristic("on")
on_char.value = 1
accessory = Accessory("TestDevice", "example.com", "Test", "0001", "0.1")
accessory.services.append(service)
service = accessory.add_service(ServicesTypes.LIGHTBULB)
on_char = service.add_char(CharacteristicsTypes.ON)
on_char.value = 0
fake_controller = await setup_platform(hass)
pairing = fake_controller.add([accessory])
@ -857,7 +915,6 @@ async def test_parse_overlapping_homekit_json(hass):
pairing_cls_imp = (
"homeassistant.components.homekit_controller.config_flow.IpPairing"
)
with mock.patch(pairing_cls_imp) as pairing_cls:
pairing_cls.return_value = pairing
with mock.patch("builtins.open", side_effect=side_effects):
@ -894,22 +951,39 @@ async def test_unignore_works(hass):
}
pairing = mock.Mock(pairing_data={"AccessoryPairingID": "00:00:00:00:00:00"})
pairing.list_accessories_and_characteristics.return_value = [
{
"aid": 1,
"services": [
{
"characteristics": [{"type": "23", "value": "Koogeek-LS1-20833F"}],
"type": "3e",
}
],
}
]
pairing.list_accessories_and_characteristics = asynctest.CoroutineMock(
return_value=[
{
"aid": 1,
"services": [
{
"characteristics": [
{"type": "23", "value": "Koogeek-LS1-20833F"}
],
"type": "3e",
}
],
}
]
)
finish_pairing = asynctest.CoroutineMock()
discovery = mock.Mock()
discovery.device_id = "00:00:00:00:00:00"
discovery.info = discovery_info
discovery.start_pairing = asynctest.CoroutineMock(return_value=finish_pairing)
finish_pairing.return_value = pairing
flow = _setup_flow_handler(hass)
flow.controller.pairings = {"00:00:00:00:00:00": pairing}
flow.controller.discover.return_value = [discovery_info]
flow.controller.discover_ip = asynctest.CoroutineMock(return_value=[discovery])
flow.controller.find_ip_by_device_id = asynctest.CoroutineMock(
return_value=discovery
)
result = await flow.async_step_unignore({"unique_id": "00:00:00:00:00:00"})
assert result["type"] == "form"
@ -924,7 +998,6 @@ async def test_unignore_works(hass):
result = await flow.async_step_pair({})
assert result["type"] == "form"
assert result["step_id"] == "pair"
assert flow.controller.start_pairing.call_count == 1
# Pairing finalized
result = await flow.async_step_pair({"pairing_code": "111-22-333"})
@ -949,8 +1022,12 @@ async def test_unignore_ignores_missing_devices(hass):
"sf": 1,
}
discovery = mock.Mock()
discovery.device_id = "00:00:00:00:00:00"
discovery.info = discovery_info
flow = _setup_flow_handler(hass)
flow.controller.discover.return_value = [discovery_info]
flow.controller.discover_ip = asynctest.CoroutineMock(return_value=[discovery])
result = await flow.async_step_unignore({"unique_id": "00:00:00:00:00:01"})
assert result["type"] == "abort"

92
tests/components/homekit_controller/test_cover.py
View File

@ -1,5 +1,8 @@
"""Basic checks for HomeKitalarm_control_panel."""
from tests.components.homekit_controller.common import FakeService, setup_test_component
from aiohomekit.model.characteristics import CharacteristicsTypes
from aiohomekit.model.services import ServicesTypes
from tests.components.homekit_controller.common import setup_test_component
POSITION_STATE = ("window-covering", "position.state")
POSITION_CURRENT = ("window-covering", "position.current")
@ -19,61 +22,56 @@ DOOR_TARGET = ("garage-door-opener", "door-state.target")
DOOR_OBSTRUCTION = ("garage-door-opener", "obstruction-detected")
def create_window_covering_service():
def create_window_covering_service(accessory):
"""Define a window-covering characteristics as per page 219 of HAP spec."""
service = FakeService("public.hap.service.window-covering")
service = accessory.add_service(ServicesTypes.WINDOW_COVERING)
cur_state = service.add_characteristic("position.current")
cur_state = service.add_char(CharacteristicsTypes.POSITION_CURRENT)
cur_state.value = 0
targ_state = service.add_characteristic("position.target")
targ_state = service.add_char(CharacteristicsTypes.POSITION_TARGET)
targ_state.value = 0
position_state = service.add_characteristic("position.state")
position_state = service.add_char(CharacteristicsTypes.POSITION_STATE)
position_state.value = 0
position_hold = service.add_characteristic("position.hold")
position_hold = service.add_char(CharacteristicsTypes.POSITION_HOLD)
position_hold.value = 0
obstruction = service.add_characteristic("obstruction-detected")
obstruction = service.add_char(CharacteristicsTypes.OBSTRUCTION_DETECTED)
obstruction.value = False
name = service.add_characteristic("name")
name = service.add_char(CharacteristicsTypes.NAME)
name.value = "testdevice"
return service
def create_window_covering_service_with_h_tilt():
def create_window_covering_service_with_h_tilt(accessory):
"""Define a window-covering characteristics as per page 219 of HAP spec."""
service = create_window_covering_service()
service = create_window_covering_service(accessory)
tilt_current = service.add_characteristic("horizontal-tilt.current")
tilt_current = service.add_char(CharacteristicsTypes.HORIZONTAL_TILT_CURRENT)
tilt_current.value = 0
tilt_target = service.add_characteristic("horizontal-tilt.target")
tilt_target = service.add_char(CharacteristicsTypes.HORIZONTAL_TILT_TARGET)
tilt_target.value = 0
return service
def create_window_covering_service_with_v_tilt():
def create_window_covering_service_with_v_tilt(accessory):
"""Define a window-covering characteristics as per page 219 of HAP spec."""
service = create_window_covering_service()
service = create_window_covering_service(accessory)
tilt_current = service.add_characteristic("vertical-tilt.current")
tilt_current = service.add_char(CharacteristicsTypes.VERTICAL_TILT_CURRENT)
tilt_current.value = 0
tilt_target = service.add_characteristic("vertical-tilt.target")
tilt_target = service.add_char(CharacteristicsTypes.VERTICAL_TILT_TARGET)
tilt_target.value = 0
return service
async def test_change_window_cover_state(hass, utcnow):
"""Test that we can turn a HomeKit alarm on and off again."""
window_cover = create_window_covering_service()
helper = await setup_test_component(hass, [window_cover])
helper = await setup_test_component(hass, create_window_covering_service)
await hass.services.async_call(
"cover", "open_cover", {"entity_id": helper.entity_id}, blocking=True
@ -88,8 +86,7 @@ async def test_change_window_cover_state(hass, utcnow):
async def test_read_window_cover_state(hass, utcnow):
"""Test that we can read the state of a HomeKit alarm accessory."""
window_cover = create_window_covering_service()
helper = await setup_test_component(hass, [window_cover])
helper = await setup_test_component(hass, create_window_covering_service)
helper.characteristics[POSITION_STATE].value = 0
state = await helper.poll_and_get_state()
@ -110,8 +107,9 @@ async def test_read_window_cover_state(hass, utcnow):
async def test_read_window_cover_tilt_horizontal(hass, utcnow):
"""Test that horizontal tilt is handled correctly."""
window_cover = create_window_covering_service_with_h_tilt()
helper = await setup_test_component(hass, [window_cover])
helper = await setup_test_component(
hass, create_window_covering_service_with_h_tilt
)
helper.characteristics[H_TILT_CURRENT].value = 75
state = await helper.poll_and_get_state()
@ -120,8 +118,9 @@ async def test_read_window_cover_tilt_horizontal(hass, utcnow):
async def test_read_window_cover_tilt_vertical(hass, utcnow):
"""Test that vertical tilt is handled correctly."""
window_cover = create_window_covering_service_with_v_tilt()
helper = await setup_test_component(hass, [window_cover])
helper = await setup_test_component(
hass, create_window_covering_service_with_v_tilt
)
helper.characteristics[V_TILT_CURRENT].value = 75
state = await helper.poll_and_get_state()
@ -130,8 +129,9 @@ async def test_read_window_cover_tilt_vertical(hass, utcnow):
async def test_write_window_cover_tilt_horizontal(hass, utcnow):
"""Test that horizontal tilt is written correctly."""
window_cover = create_window_covering_service_with_h_tilt()
helper = await setup_test_component(hass, [window_cover])
helper = await setup_test_component(
hass, create_window_covering_service_with_h_tilt
)
await hass.services.async_call(
"cover",
@ -144,8 +144,9 @@ async def test_write_window_cover_tilt_horizontal(hass, utcnow):
async def test_write_window_cover_tilt_vertical(hass, utcnow):
"""Test that vertical tilt is written correctly."""
window_cover = create_window_covering_service_with_v_tilt()
helper = await setup_test_component(hass, [window_cover])
helper = await setup_test_component(
hass, create_window_covering_service_with_v_tilt
)
await hass.services.async_call(
"cover",
@ -158,8 +159,9 @@ async def test_write_window_cover_tilt_vertical(hass, utcnow):
async def test_window_cover_stop(hass, utcnow):
"""Test that vertical tilt is written correctly."""
window_cover = create_window_covering_service_with_v_tilt()
helper = await setup_test_component(hass, [window_cover])
helper = await setup_test_component(
hass, create_window_covering_service_with_v_tilt
)
await hass.services.async_call(
"cover", "stop_cover", {"entity_id": helper.entity_id}, blocking=True
@ -167,20 +169,20 @@ async def test_window_cover_stop(hass, utcnow):
assert helper.characteristics[POSITION_HOLD].value == 1
def create_garage_door_opener_service():
def create_garage_door_opener_service(accessory):
"""Define a garage-door-opener chars as per page 217 of HAP spec."""
service = FakeService("public.hap.service.garage-door-opener")
service = accessory.add_service(ServicesTypes.GARAGE_DOOR_OPENER)
cur_state = service.add_characteristic("door-state.current")
cur_state = service.add_char(CharacteristicsTypes.DOOR_STATE_CURRENT)
cur_state.value = 0
targ_state = service.add_characteristic("door-state.target")
targ_state.value = 0
cur_state = service.add_char(CharacteristicsTypes.DOOR_STATE_TARGET)
cur_state.value = 0
obstruction = service.add_characteristic("obstruction-detected")
obstruction = service.add_char(CharacteristicsTypes.OBSTRUCTION_DETECTED)
obstruction.value = False
name = service.add_characteristic("name")
name = service.add_char(CharacteristicsTypes.NAME)
name.value = "testdevice"
return service
@ -188,8 +190,7 @@ def create_garage_door_opener_service():
async def test_change_door_state(hass, utcnow):
"""Test that we can turn open and close a HomeKit garage door."""
door = create_garage_door_opener_service()
helper = await setup_test_component(hass, [door])
helper = await setup_test_component(hass, create_garage_door_opener_service)
await hass.services.async_call(
"cover", "open_cover", {"entity_id": helper.entity_id}, blocking=True
@ -204,8 +205,7 @@ async def test_change_door_state(hass, utcnow):
async def test_read_door_state(hass, utcnow):
"""Test that we can read the state of a HomeKit garage door."""
door = create_garage_door_opener_service()
helper = await setup_test_component(hass, [door])
helper = await setup_test_component(hass, create_garage_door_opener_service)
helper.characteristics[DOOR_CURRENT].value = 0
state = await helper.poll_and_get_state()

89
tests/components/homekit_controller/test_fan.py
View File

@ -1,5 +1,8 @@
"""Basic checks for HomeKit motion sensors and contact sensors."""
from tests.components.homekit_controller.common import FakeService, setup_test_component
from aiohomekit.model.characteristics import CharacteristicsTypes
from aiohomekit.model.services import ServicesTypes
from tests.components.homekit_controller.common import setup_test_component
V1_ON = ("fan", "on")
V1_ROTATION_DIRECTION = ("fan", "rotation.direction")
@ -11,50 +14,45 @@ V2_ROTATION_SPEED = ("fanv2", "rotation.speed")
V2_SWING_MODE = ("fanv2", "swing-mode")
def create_fan_service():
def create_fan_service(accessory):
"""
Define fan v1 characteristics as per HAP spec.
This service is no longer documented in R2 of the public HAP spec but existing
devices out there use it (like the SIMPLEconnect fan)
"""
service = FakeService("public.hap.service.fan")
service = accessory.add_service(ServicesTypes.FAN)
cur_state = service.add_characteristic("on")
cur_state = service.add_char(CharacteristicsTypes.ON)
cur_state.value = 0
cur_state = service.add_characteristic("rotation.direction")
cur_state.value = 0
direction = service.add_char(CharacteristicsTypes.ROTATION_DIRECTION)
direction.value = 0
cur_state = service.add_characteristic("rotation.speed")
cur_state.value = 0
return service
speed = service.add_char(CharacteristicsTypes.ROTATION_SPEED)
speed.value = 0
def create_fanv2_service():
def create_fanv2_service(accessory):
"""Define fan v2 characteristics as per HAP spec."""
service = FakeService("public.hap.service.fanv2")
service = accessory.add_service(ServicesTypes.FAN_V2)
cur_state = service.add_characteristic("active")
cur_state = service.add_char(CharacteristicsTypes.ACTIVE)
cur_state.value = 0
cur_state = service.add_characteristic("rotation.direction")
cur_state.value = 0
direction = service.add_char(CharacteristicsTypes.ROTATION_DIRECTION)
direction.value = 0
cur_state = service.add_characteristic("rotation.speed")
cur_state.value = 0
speed = service.add_char(CharacteristicsTypes.ROTATION_SPEED)
speed.value = 0
cur_state = service.add_characteristic("swing-mode")
cur_state.value = 0
return service
swing_mode = service.add_char(CharacteristicsTypes.SWING_MODE)
swing_mode.value = 0
async def test_fan_read_state(hass, utcnow):
"""Test that we can read the state of a HomeKit fan accessory."""
sensor = create_fan_service()
helper = await setup_test_component(hass, [sensor])
helper = await setup_test_component(hass, create_fan_service)
helper.characteristics[V1_ON].value = False
state = await helper.poll_and_get_state()
@ -67,8 +65,7 @@ async def test_fan_read_state(hass, utcnow):
async def test_turn_on(hass, utcnow):
"""Test that we can turn a fan on."""
fan = create_fan_service()
helper = await setup_test_component(hass, [fan])
helper = await setup_test_component(hass, create_fan_service)
await hass.services.async_call(
"fan",
@ -100,8 +97,7 @@ async def test_turn_on(hass, utcnow):
async def test_turn_off(hass, utcnow):
"""Test that we can turn a fan off."""
fan = create_fan_service()
helper = await setup_test_component(hass, [fan])
helper = await setup_test_component(hass, create_fan_service)
helper.characteristics[V1_ON].value = 1
@ -113,8 +109,7 @@ async def test_turn_off(hass, utcnow):
async def test_set_speed(hass, utcnow):
"""Test that we set fan speed."""
fan = create_fan_service()
helper = await setup_test_component(hass, [fan])
helper = await setup_test_component(hass, create_fan_service)
helper.characteristics[V1_ON].value = 1
@ -153,8 +148,7 @@ async def test_set_speed(hass, utcnow):
async def test_speed_read(hass, utcnow):
"""Test that we can read a fans oscillation."""
fan = create_fan_service()
helper = await setup_test_component(hass, [fan])
helper = await setup_test_component(hass, create_fan_service)
helper.characteristics[V1_ON].value = 1
helper.characteristics[V1_ROTATION_SPEED].value = 100
@ -177,8 +171,7 @@ async def test_speed_read(hass, utcnow):
async def test_set_direction(hass, utcnow):
"""Test that we can set fan spin direction."""
fan = create_fan_service()
helper = await setup_test_component(hass, [fan])
helper = await setup_test_component(hass, create_fan_service)
await hass.services.async_call(
"fan",
@ -199,8 +192,7 @@ async def test_set_direction(hass, utcnow):
async def test_direction_read(hass, utcnow):
"""Test that we can read a fans oscillation."""
fan = create_fan_service()
helper = await setup_test_component(hass, [fan])
helper = await setup_test_component(hass, create_fan_service)
helper.characteristics[V1_ROTATION_DIRECTION].value = 0
state = await helper.poll_and_get_state()
@ -213,8 +205,7 @@ async def test_direction_read(hass, utcnow):
async def test_fanv2_read_state(hass, utcnow):
"""Test that we can read the state of a HomeKit fan accessory."""
sensor = create_fanv2_service()
helper = await setup_test_component(hass, [sensor])
helper = await setup_test_component(hass, create_fanv2_service)
helper.characteristics[V2_ACTIVE].value = False
state = await helper.poll_and_get_state()
@ -227,8 +218,7 @@ async def test_fanv2_read_state(hass, utcnow):
async def test_v2_turn_on(hass, utcnow):
"""Test that we can turn a fan on."""
fan = create_fanv2_service()
helper = await setup_test_component(hass, [fan])
helper = await setup_test_component(hass, create_fanv2_service)
await hass.services.async_call(
"fan",
@ -260,8 +250,7 @@ async def test_v2_turn_on(hass, utcnow):
async def test_v2_turn_off(hass, utcnow):
"""Test that we can turn a fan off."""
fan = create_fanv2_service()
helper = await setup_test_component(hass, [fan])
helper = await setup_test_component(hass, create_fanv2_service)
helper.characteristics[V2_ACTIVE].value = 1
@ -273,8 +262,7 @@ async def test_v2_turn_off(hass, utcnow):
async def test_v2_set_speed(hass, utcnow):
"""Test that we set fan speed."""
fan = create_fanv2_service()
helper = await setup_test_component(hass, [fan])
helper = await setup_test_component(hass, create_fanv2_service)
helper.characteristics[V2_ACTIVE].value = 1
@ -313,8 +301,7 @@ async def test_v2_set_speed(hass, utcnow):
async def test_v2_speed_read(hass, utcnow):
"""Test that we can read a fans oscillation."""
fan = create_fanv2_service()
helper = await setup_test_component(hass, [fan])
helper = await setup_test_component(hass, create_fanv2_service)
helper.characteristics[V2_ACTIVE].value = 1
helper.characteristics[V2_ROTATION_SPEED].value = 100
@ -337,8 +324,7 @@ async def test_v2_speed_read(hass, utcnow):
async def test_v2_set_direction(hass, utcnow):
"""Test that we can set fan spin direction."""
fan = create_fanv2_service()
helper = await setup_test_component(hass, [fan])
helper = await setup_test_component(hass, create_fanv2_service)
await hass.services.async_call(
"fan",
@ -359,8 +345,7 @@ async def test_v2_set_direction(hass, utcnow):
async def test_v2_direction_read(hass, utcnow):
"""Test that we can read a fans oscillation."""
fan = create_fanv2_service()
helper = await setup_test_component(hass, [fan])
helper = await setup_test_component(hass, create_fanv2_service)
helper.characteristics[V2_ROTATION_DIRECTION].value = 0
state = await helper.poll_and_get_state()
@ -373,8 +358,7 @@ async def test_v2_direction_read(hass, utcnow):
async def test_v2_oscillate(hass, utcnow):
"""Test that we can control a fans oscillation."""
fan = create_fanv2_service()
helper = await setup_test_component(hass, [fan])
helper = await setup_test_component(hass, create_fanv2_service)
await hass.services.async_call(
"fan",
@ -395,8 +379,7 @@ async def test_v2_oscillate(hass, utcnow):
async def test_v2_oscillate_read(hass, utcnow):
"""Test that we can read a fans oscillation."""
fan = create_fanv2_service()
helper = await setup_test_component(hass, [fan])
helper = await setup_test_component(hass, create_fanv2_service)
helper.characteristics[V2_SWING_MODE].value = 0
state = await helper.poll_and_get_state()

45
tests/components/homekit_controller/test_light.py
View File

@ -1,7 +1,10 @@
"""Basic checks for HomeKitSwitch."""
from aiohomekit.model.characteristics import CharacteristicsTypes
from aiohomekit.model.services import ServicesTypes
from homeassistant.components.homekit_controller.const import KNOWN_DEVICES
from tests.components.homekit_controller.common import FakeService, setup_test_component
from tests.components.homekit_controller.common import setup_test_component
LIGHT_ON = ("lightbulb", "on")
LIGHT_BRIGHTNESS = ("lightbulb", "brightness")
@ -10,37 +13,37 @@ LIGHT_SATURATION = ("lightbulb", "saturation")
LIGHT_COLOR_TEMP = ("lightbulb", "color-temperature")
def create_lightbulb_service():
def create_lightbulb_service(accessory):
"""Define lightbulb characteristics."""
service = FakeService("public.hap.service.lightbulb")
service = accessory.add_service(ServicesTypes.LIGHTBULB)
on_char = service.add_characteristic("on")
on_char = service.add_char(CharacteristicsTypes.ON)
on_char.value = 0
brightness = service.add_characteristic("brightness")
brightness = service.add_char(CharacteristicsTypes.BRIGHTNESS)
brightness.value = 0
return service
def create_lightbulb_service_with_hs():
def create_lightbulb_service_with_hs(accessory):
"""Define a lightbulb service with hue + saturation."""
service = create_lightbulb_service()
service = create_lightbulb_service(accessory)
hue = service.add_characteristic("hue")
hue = service.add_char(CharacteristicsTypes.HUE)
hue.value = 0
saturation = service.add_characteristic("saturation")
saturation = service.add_char(CharacteristicsTypes.SATURATION)
saturation.value = 0
return service
def create_lightbulb_service_with_color_temp():
def create_lightbulb_service_with_color_temp(accessory):
"""Define a lightbulb service with color temp."""
service = create_lightbulb_service()
service = create_lightbulb_service(accessory)
color_temp = service.add_characteristic("color-temperature")
color_temp = service.add_char(CharacteristicsTypes.COLOR_TEMPERATURE)
color_temp.value = 0
return service
@ -48,8 +51,7 @@ def create_lightbulb_service_with_color_temp():
async def test_switch_change_light_state(hass, utcnow):
"""Test that we can turn a HomeKit light on and off again."""
bulb = create_lightbulb_service_with_hs()
helper = await setup_test_component(hass, [bulb])
helper = await setup_test_component(hass, create_lightbulb_service_with_hs)
await hass.services.async_call(
"light",
@ -71,8 +73,7 @@ async def test_switch_change_light_state(hass, utcnow):
async def test_switch_change_light_state_color_temp(hass, utcnow):
"""Test that we can turn change color_temp."""
bulb = create_lightbulb_service_with_color_temp()
helper = await setup_test_component(hass, [bulb])
helper = await setup_test_component(hass, create_lightbulb_service_with_color_temp)
await hass.services.async_call(
"light",
@ -87,8 +88,7 @@ async def test_switch_change_light_state_color_temp(hass, utcnow):
async def test_switch_read_light_state(hass, utcnow):
"""Test that we can read the state of a HomeKit light accessory."""
bulb = create_lightbulb_service_with_hs()
helper = await setup_test_component(hass, [bulb])
helper = await setup_test_component(hass, create_lightbulb_service_with_hs)
# Initial state is that the light is off
state = await helper.poll_and_get_state()
@ -112,8 +112,7 @@ async def test_switch_read_light_state(hass, utcnow):
async def test_switch_read_light_state_color_temp(hass, utcnow):
"""Test that we can read the color_temp of a light accessory."""
bulb = create_lightbulb_service_with_color_temp()
helper = await setup_test_component(hass, [bulb])
helper = await setup_test_component(hass, create_lightbulb_service_with_color_temp)
# Initial state is that the light is off
state = await helper.poll_and_get_state()
@ -132,8 +131,7 @@ async def test_switch_read_light_state_color_temp(hass, utcnow):
async def test_light_becomes_unavailable_but_recovers(hass, utcnow):
"""Test transition to and from unavailable state."""
bulb = create_lightbulb_service_with_color_temp()
helper = await setup_test_component(hass, [bulb])
helper = await setup_test_component(hass, create_lightbulb_service_with_color_temp)
# Initial state is that the light is off
state = await helper.poll_and_get_state()
@ -158,8 +156,7 @@ async def test_light_becomes_unavailable_but_recovers(hass, utcnow):
async def test_light_unloaded(hass, utcnow):
"""Test entity and HKDevice are correctly unloaded."""
bulb = create_lightbulb_service_with_color_temp()
helper = await setup_test_component(hass, [bulb])
helper = await setup_test_component(hass, create_lightbulb_service_with_color_temp)
# Initial state is that the light is off
state = await helper.poll_and_get_state()

21
tests/components/homekit_controller/test_lock.py
View File

@ -1,25 +1,28 @@
"""Basic checks for HomeKitLock."""
from tests.components.homekit_controller.common import FakeService, setup_test_component
from aiohomekit.model.characteristics import CharacteristicsTypes
from aiohomekit.model.services import ServicesTypes
from tests.components.homekit_controller.common import setup_test_component
LOCK_CURRENT_STATE = ("lock-mechanism", "lock-mechanism.current-state")
LOCK_TARGET_STATE = ("lock-mechanism", "lock-mechanism.target-state")
def create_lock_service():
def create_lock_service(accessory):
"""Define a lock characteristics as per page 219 of HAP spec."""
service = FakeService("public.hap.service.lock-mechanism")
service = accessory.add_service(ServicesTypes.LOCK_MECHANISM)
cur_state = service.add_characteristic("lock-mechanism.current-state")
cur_state = service.add_char(CharacteristicsTypes.LOCK_MECHANISM_CURRENT_STATE)
cur_state.value = 0
targ_state = service.add_characteristic("lock-mechanism.target-state")
targ_state = service.add_char(CharacteristicsTypes.LOCK_MECHANISM_TARGET_STATE)
targ_state.value = 0
# According to the spec, a battery-level characteristic is normally
# part of a separate service. However as the code was written (which
# predates this test) the battery level would have to be part of the lock
# service as it is here.
targ_state = service.add_characteristic("battery-level")
targ_state = service.add_char(CharacteristicsTypes.BATTERY_LEVEL)
targ_state.value = 50
return service
@ -27,8 +30,7 @@ def create_lock_service():
async def test_switch_change_lock_state(hass, utcnow):
"""Test that we can turn a HomeKit lock on and off again."""
lock = create_lock_service()
helper = await setup_test_component(hass, [lock])
helper = await setup_test_component(hass, create_lock_service)
await hass.services.async_call(
"lock", "lock", {"entity_id": "lock.testdevice"}, blocking=True
@ -43,8 +45,7 @@ async def test_switch_change_lock_state(hass, utcnow):
async def test_switch_read_lock_state(hass, utcnow):
"""Test that we can read the state of a HomeKit lock accessory."""
lock = create_lock_service()
helper = await setup_test_component(hass, [lock])
helper = await setup_test_component(hass, create_lock_service)
helper.characteristics[LOCK_CURRENT_STATE].value = 0
helper.characteristics[LOCK_TARGET_STATE].value = 0

82
tests/components/homekit_controller/test_sensor.py
View File

@ -1,5 +1,8 @@
"""Basic checks for HomeKit sensor."""
from tests.components.homekit_controller.common import FakeService, setup_test_component
from aiohomekit.model.characteristics import CharacteristicsTypes
from aiohomekit.model.services import ServicesTypes
from tests.components.homekit_controller.common import setup_test_component
TEMPERATURE = ("temperature", "temperature.current")
HUMIDITY = ("humidity", "relative-humidity.current")
@ -10,57 +13,49 @@ CHARGING_STATE = ("battery", "charging-state")
LO_BATT = ("battery", "status-lo-batt")
def create_temperature_sensor_service():
def create_temperature_sensor_service(accessory):
"""Define temperature characteristics."""
service = FakeService("public.hap.service.sensor.temperature")
service = accessory.add_service(ServicesTypes.TEMPERATURE_SENSOR)
cur_state = service.add_characteristic("temperature.current")
cur_state = service.add_char(CharacteristicsTypes.TEMPERATURE_CURRENT)
cur_state.value = 0
return service
def create_humidity_sensor_service():
def create_humidity_sensor_service(accessory):
"""Define humidity characteristics."""
service = FakeService("public.hap.service.sensor.humidity")
service = accessory.add_service(ServicesTypes.HUMIDITY_SENSOR)
cur_state = service.add_characteristic("relative-humidity.current")
cur_state = service.add_char(CharacteristicsTypes.RELATIVE_HUMIDITY_CURRENT)
cur_state.value = 0
return service
def create_light_level_sensor_service():
def create_light_level_sensor_service(accessory):
"""Define light level characteristics."""
service = FakeService("public.hap.service.sensor.light")
service = accessory.add_service(ServicesTypes.LIGHT_SENSOR)
cur_state = service.add_characteristic("light-level.current")
cur_state = service.add_char(CharacteristicsTypes.LIGHT_LEVEL_CURRENT)
cur_state.value = 0
return service
def create_carbon_dioxide_level_sensor_service():
def create_carbon_dioxide_level_sensor_service(accessory):
"""Define carbon dioxide level characteristics."""
service = FakeService("public.hap.service.sensor.carbon-dioxide")
service = accessory.add_service(ServicesTypes.CARBON_DIOXIDE_SENSOR)
cur_state = service.add_characteristic("carbon-dioxide.level")
cur_state = service.add_char(CharacteristicsTypes.CARBON_DIOXIDE_LEVEL)
cur_state.value = 0
return service
def create_battery_level_sensor():
def create_battery_level_sensor(accessory):
"""Define battery level characteristics."""
service = FakeService("public.hap.service.battery")
service = accessory.add_service(ServicesTypes.BATTERY_SERVICE)
cur_state = service.add_characteristic("battery-level")
cur_state = service.add_char(CharacteristicsTypes.BATTERY_LEVEL)
cur_state.value = 100
low_battery = service.add_characteristic("status-lo-batt")
low_battery = service.add_char(CharacteristicsTypes.STATUS_LO_BATT)
low_battery.value = 0
charging_state = service.add_characteristic("charging-state")
charging_state = service.add_char(CharacteristicsTypes.CHARGING_STATE)
charging_state.value = 0
return service
@ -68,8 +63,9 @@ def create_battery_level_sensor():
async def test_temperature_sensor_read_state(hass, utcnow):
"""Test reading the state of a HomeKit temperature sensor accessory."""
sensor = create_temperature_sensor_service()
helper = await setup_test_component(hass, [sensor], suffix="temperature")
helper = await setup_test_component(
hass, create_temperature_sensor_service, suffix="temperature"
)
helper.characteristics[TEMPERATURE].value = 10
state = await helper.poll_and_get_state()
@ -82,8 +78,9 @@ async def test_temperature_sensor_read_state(hass, utcnow):
async def test_humidity_sensor_read_state(hass, utcnow):
"""Test reading the state of a HomeKit humidity sensor accessory."""
sensor = create_humidity_sensor_service()
helper = await setup_test_component(hass, [sensor], suffix="humidity")
helper = await setup_test_component(
hass, create_humidity_sensor_service, suffix="humidity"
)
helper.characteristics[HUMIDITY].value = 10
state = await helper.poll_and_get_state()
@ -96,8 +93,9 @@ async def test_humidity_sensor_read_state(hass, utcnow):
async def test_light_level_sensor_read_state(hass, utcnow):
"""Test reading the state of a HomeKit temperature sensor accessory."""
sensor = create_light_level_sensor_service()
helper = await setup_test_component(hass, [sensor], suffix="light_level")
helper = await setup_test_component(
hass, create_light_level_sensor_service, suffix="light_level"
)
helper.characteristics[LIGHT_LEVEL].value = 10
state = await helper.poll_and_get_state()
@ -110,8 +108,9 @@ async def test_light_level_sensor_read_state(hass, utcnow):
async def test_carbon_dioxide_level_sensor_read_state(hass, utcnow):
"""Test reading the state of a HomeKit carbon dioxide sensor accessory."""
sensor = create_carbon_dioxide_level_sensor_service()
helper = await setup_test_component(hass, [sensor], suffix="co2")
helper = await setup_test_component(
hass, create_carbon_dioxide_level_sensor_service, suffix="co2"
)
helper.characteristics[CARBON_DIOXIDE_LEVEL].value = 10
state = await helper.poll_and_get_state()
@ -124,8 +123,9 @@ async def test_carbon_dioxide_level_sensor_read_state(hass, utcnow):
async def test_battery_level_sensor(hass, utcnow):
"""Test reading the state of a HomeKit battery level sensor."""
sensor = create_battery_level_sensor()
helper = await setup_test_component(hass, [sensor], suffix="battery")
helper = await setup_test_component(
hass, create_battery_level_sensor, suffix="battery"
)
helper.characteristics[BATTERY_LEVEL].value = 100
state = await helper.poll_and_get_state()
@ -140,8 +140,9 @@ async def test_battery_level_sensor(hass, utcnow):
async def test_battery_charging(hass, utcnow):
"""Test reading the state of a HomeKit battery's charging state."""
sensor = create_battery_level_sensor()
helper = await setup_test_component(hass, [sensor], suffix="battery")
helper = await setup_test_component(
hass, create_battery_level_sensor, suffix="battery"
)
helper.characteristics[BATTERY_LEVEL].value = 0
helper.characteristics[CHARGING_STATE].value = 1
@ -155,8 +156,9 @@ async def test_battery_charging(hass, utcnow):
async def test_battery_low(hass, utcnow):
"""Test reading the state of a HomeKit battery's low state."""
sensor = create_battery_level_sensor()
helper = await setup_test_component(hass, [sensor], suffix="battery")
helper = await setup_test_component(
hass, create_battery_level_sensor, suffix="battery"
)
helper.characteristics[LO_BATT].value = 0
helper.characteristics[BATTERY_LEVEL].value = 1

17
tests/components/homekit_controller/test_storage.py
View File

@ -1,11 +1,13 @@
"""Basic checks for entity map storage."""
from aiohomekit.model.characteristics import CharacteristicsTypes
from aiohomekit.model.services import ServicesTypes
from homeassistant import config_entries
from homeassistant.components.homekit_controller import async_remove_entry
from homeassistant.components.homekit_controller.const import ENTITY_MAP
from tests.common import flush_store
from tests.components.homekit_controller.common import (
FakeService,
setup_platform,
setup_test_component,
)
@ -57,18 +59,16 @@ async def test_storage_is_removed_idempotent(hass):
assert hkid not in entity_map.storage_data
def create_lightbulb_service():
def create_lightbulb_service(accessory):
"""Define lightbulb characteristics."""
service = FakeService("public.hap.service.lightbulb")
on_char = service.add_characteristic("on")
service = accessory.add_service(ServicesTypes.LIGHTBULB)
on_char = service.add_char(CharacteristicsTypes.ON)
on_char.value = 0
return service
async def test_storage_is_updated_on_add(hass, hass_storage, utcnow):
"""Test entity map storage is cleaned up on adding an accessory."""
bulb = create_lightbulb_service()
await setup_test_component(hass, [bulb])
await setup_test_component(hass, create_lightbulb_service)
entity_map = hass.data[ENTITY_MAP]
hkid = "00:00:00:00:00:00"
@ -83,8 +83,7 @@ async def test_storage_is_updated_on_add(hass, hass_storage, utcnow):
async def test_storage_is_removed_on_config_entry_removal(hass, utcnow):
"""Test entity map storage is cleaned up on config entry removal."""
bulb = create_lightbulb_service()
await setup_test_component(hass, [bulb])
await setup_test_component(hass, create_lightbulb_service)
hkid = "00:00:00:00:00:00"

23
tests/components/homekit_controller/test_switch.py
View File

@ -1,12 +1,25 @@
"""Basic checks for HomeKitSwitch."""
from aiohomekit.model.characteristics import CharacteristicsTypes
from aiohomekit.model.services import ServicesTypes
from tests.components.homekit_controller.common import setup_test_component
def create_switch_service(accessory):
"""Define outlet characteristics."""
service = accessory.add_service(ServicesTypes.OUTLET)
on_char = service.add_char(CharacteristicsTypes.ON)
on_char.value = False
outlet_in_use = service.add_char(CharacteristicsTypes.OUTLET_IN_USE)
outlet_in_use.value = False
async def test_switch_change_outlet_state(hass, utcnow):
"""Test that we can turn a HomeKit outlet on and off again."""
from homekit.model.services import OutletService
helper = await setup_test_component(hass, [OutletService()])
helper = await setup_test_component(hass, create_switch_service)
await hass.services.async_call(
"switch", "turn_on", {"entity_id": "switch.testdevice"}, blocking=True
@ -21,9 +34,7 @@ async def test_switch_change_outlet_state(hass, utcnow):
async def test_switch_read_outlet_state(hass, utcnow):
"""Test that we can read the state of a HomeKit outlet accessory."""
from homekit.model.services import OutletService
helper = await setup_test_component(hass, [OutletService()])
helper = await setup_test_component(hass, create_switch_service)
# Initial state is that the switch is off and the outlet isn't in use
switch_1 = await helper.poll_and_get_state()