Rewrite dyson init test (#45409)

This commit is contained in:
Xiaonan Shen 2021-01-22 23:27:43 +08:00 committed by GitHub
parent 4aceb0dd27
commit 7e8d0a263c
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 163 additions and 258 deletions

View file

@ -4,16 +4,20 @@ from typing import Optional, Type
from unittest import mock
from unittest.mock import MagicMock
from libpurecool.const import SLEEP_TIMER_OFF, Dyson360EyeMode, FanMode, PowerMode
from libpurecool.dyson_360_eye import Dyson360Eye
from libpurecool.dyson_device import DysonDevice
from libpurecool.dyson_pure_cool import FanSpeed
from libpurecool.dyson_pure_cool import DysonPureCool, FanSpeed
from libpurecool.dyson_pure_cool_link import DysonPureCoolLink
from homeassistant.components.dyson import CONF_LANGUAGE, DOMAIN
from homeassistant.const import CONF_DEVICES, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.core import HomeAssistant, callback
SERIAL = "XX-XXXXX-XX"
NAME = "Temp Name"
ENTITY_NAME = "temp_name"
IP_ADDRESS = "0.0.0.0"
BASE_PATH = "homeassistant.components.dyson"
@ -25,7 +29,7 @@ CONFIG = {
CONF_DEVICES: [
{
"device_id": SERIAL,
"device_ip": "0.0.0.0",
"device_ip": IP_ADDRESS,
}
],
}
@ -61,6 +65,46 @@ def get_basic_device(spec: Type[DysonDevice]) -> DysonDevice:
return device
@callback
def async_get_360eye_device(state=Dyson360EyeMode.FULL_CLEAN_RUNNING) -> Dyson360Eye:
"""Return a Dyson 360 Eye device."""
device = get_basic_device(Dyson360Eye)
device.state.state = state
device.state.battery_level = 85
device.state.power_mode = PowerMode.QUIET
device.state.position = (0, 0)
return device
@callback
def async_get_purecoollink_device() -> DysonPureCoolLink:
"""Return a Dyson Pure Cool Link device."""
device = get_basic_device(DysonPureCoolLink)
device.state.fan_mode = FanMode.FAN.value
device.state.speed = FanSpeed.FAN_SPEED_1.value
device.state.night_mode = "ON"
device.state.oscillation = "ON"
return device
@callback
def async_get_purecool_device() -> DysonPureCool:
"""Return a Dyson Pure Cool device."""
device = get_basic_device(DysonPureCool)
device.state.fan_power = "ON"
device.state.speed = FanSpeed.FAN_SPEED_1.value
device.state.night_mode = "ON"
device.state.oscillation = "OION"
device.state.oscillation_angle_low = "0024"
device.state.oscillation_angle_high = "0254"
device.state.auto_mode = "OFF"
device.state.front_direction = "ON"
device.state.sleep_timer = SLEEP_TIMER_OFF
device.state.hepa_filter_state = "0100"
device.state.carbon_filter_state = "0100"
return device
async def async_update_device(
hass: HomeAssistant, device: DysonDevice, state_type: Optional[Type] = None
) -> None:
@ -70,8 +114,8 @@ async def async_update_device(
# Combining sync calls to avoid multiple executors
def _run_callbacks():
for callback in callbacks:
callback(message)
for callback_fn in callbacks:
callback_fn(message)
await hass.async_add_executor_job(_run_callbacks)
await hass.async_block_till_done()

View file

@ -16,7 +16,7 @@ from tests.common import async_setup_component
async def device(hass: HomeAssistant, request) -> DysonDevice:
"""Fixture to provide Dyson 360 Eye device."""
platform = request.module.PLATFORM_DOMAIN
get_device = request.module.get_device
get_device = request.module.async_get_device
if hasattr(request, "param"):
if isinstance(request.param, list):
device = get_device(*request.param)

View file

@ -1,7 +1,7 @@
"""Test the Dyson fan component."""
from typing import Type
from libpurecool.const import SLEEP_TIMER_OFF, FanMode, FanSpeed, NightMode, Oscillation
from libpurecool.const import FanMode, FanSpeed, NightMode, Oscillation
from libpurecool.dyson_pure_cool import DysonPureCool, DysonPureCoolLink
from libpurecool.dyson_pure_state import DysonPureCoolState
from libpurecool.dyson_pure_state_v2 import DysonPureCoolV2State
@ -50,33 +50,24 @@ from homeassistant.const import (
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import entity_registry
from .common import ENTITY_NAME, NAME, SERIAL, async_update_device, get_basic_device
from .common import (
ENTITY_NAME,
NAME,
SERIAL,
async_get_purecool_device,
async_get_purecoollink_device,
async_update_device,
)
ENTITY_ID = f"{PLATFORM_DOMAIN}.{ENTITY_NAME}"
@callback
def get_device(spec: Type[DysonPureCoolLink]) -> DysonPureCoolLink:
def async_get_device(spec: Type[DysonPureCoolLink]) -> DysonPureCoolLink:
"""Return a Dyson fan device."""
device = get_basic_device(spec)
if spec == DysonPureCoolLink:
device.state.fan_mode = FanMode.FAN.value
device.state.speed = FanSpeed.FAN_SPEED_1.value
device.state.night_mode = "ON"
device.state.oscillation = "ON"
else: # DysonPureCool
device.state.fan_power = "ON"
device.state.speed = FanSpeed.FAN_SPEED_1.value
device.state.night_mode = "ON"
device.state.oscillation = "OION"
device.state.oscillation_angle_low = "0024"
device.state.oscillation_angle_high = "0254"
device.state.auto_mode = "OFF"
device.state.front_direction = "ON"
device.state.sleep_timer = SLEEP_TIMER_OFF
device.state.hepa_filter_state = "0100"
device.state.carbon_filter_state = "0100"
return device
return async_get_purecoollink_device()
return async_get_purecool_device()
@pytest.mark.parametrize("device", [DysonPureCoolLink], indirect=True)

View file

@ -1,231 +1,100 @@
"""Test the parent Dyson component."""
import unittest
from unittest import mock
import copy
from unittest.mock import MagicMock, patch
from homeassistant.components import dyson
from homeassistant.components.dyson import DOMAIN
from homeassistant.const import CONF_DEVICES
from homeassistant.core import HomeAssistant
from .common import load_mock_device
from .common import (
BASE_PATH,
CONFIG,
ENTITY_NAME,
IP_ADDRESS,
async_get_360eye_device,
async_get_purecool_device,
async_get_purecoollink_device,
)
from tests.common import get_test_home_assistant
from tests.common import async_setup_component
def _get_dyson_account_device_available():
"""Return a valid device provide by Dyson web services."""
device = mock.Mock()
load_mock_device(device)
device.connect = mock.Mock(return_value=True)
device.auto_connect = mock.Mock(return_value=True)
return device
async def test_setup_manual(hass: HomeAssistant):
"""Test set up the component with manually configured device IPs."""
SERIAL_TEMPLATE = "XX-XXXXX-X{}"
# device1 works
device1 = async_get_purecoollink_device()
device1.serial = SERIAL_TEMPLATE.format(1)
# device2 failed to connect
device2 = async_get_purecool_device()
device2.serial = SERIAL_TEMPLATE.format(2)
device2.connect = MagicMock(return_value=False)
# device3 throws exception during connection
device3 = async_get_360eye_device()
device3.serial = SERIAL_TEMPLATE.format(3)
device3.connect = MagicMock(side_effect=OSError)
# device4 not configured in configuration
device4 = async_get_360eye_device()
device4.serial = SERIAL_TEMPLATE.format(4)
devices = [device1, device2, device3, device4]
config = copy.deepcopy(CONFIG)
config[DOMAIN][CONF_DEVICES] = [
{
"device_id": SERIAL_TEMPLATE.format(i),
"device_ip": IP_ADDRESS,
}
for i in [1, 2, 3, 5] # 1 device missing and 1 device not existed
]
with patch(f"{BASE_PATH}.DysonAccount.login", return_value=True) as login, patch(
f"{BASE_PATH}.DysonAccount.devices", return_value=devices
) as devices_method, patch(
f"{BASE_PATH}.DYSON_PLATFORMS", ["fan", "vacuum"]
): # Patch platforms to get rid of sensors
assert await async_setup_component(hass, DOMAIN, config)
await hass.async_block_till_done()
login.assert_called_once_with()
devices_method.assert_called_once_with()
# Only one fan and zero vacuum is set up successfully
assert hass.states.async_entity_ids() == [f"fan.{ENTITY_NAME}"]
device1.connect.assert_called_once_with(IP_ADDRESS)
device2.connect.assert_called_once_with(IP_ADDRESS)
device3.connect.assert_called_once_with(IP_ADDRESS)
device4.connect.assert_not_called()
def _get_dyson_account_device_not_available():
"""Return an invalid device provide by Dyson web services."""
device = mock.Mock()
load_mock_device(device)
device.connect = mock.Mock(return_value=False)
device.auto_connect = mock.Mock(return_value=False)
return device
async def test_setup_autoconnect(hass: HomeAssistant):
"""Test set up the component with auto connect."""
# device1 works
device1 = async_get_purecoollink_device()
# device2 failed to auto connect
device2 = async_get_purecool_device()
device2.auto_connect = MagicMock(return_value=False)
devices = [device1, device2]
config = copy.deepcopy(CONFIG)
config[DOMAIN].pop(CONF_DEVICES)
with patch(f"{BASE_PATH}.DysonAccount.login", return_value=True), patch(
f"{BASE_PATH}.DysonAccount.devices", return_value=devices
), patch(
f"{BASE_PATH}.DYSON_PLATFORMS", ["fan"]
): # Patch platforms to get rid of sensors
assert await async_setup_component(hass, DOMAIN, config)
await hass.async_block_till_done()
assert hass.states.async_entity_ids_count() == 1
def _get_dyson_account_device_error():
"""Return an invalid device raising OSError while connecting."""
device = mock.Mock()
load_mock_device(device)
device.connect = mock.Mock(side_effect=OSError("Network error"))
return device
class DysonTest(unittest.TestCase):
"""Dyson parent component test class."""
def setUp(self): # pylint: disable=invalid-name
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
self.addCleanup(self.tear_down_cleanup)
def tear_down_cleanup(self):
"""Stop everything that was started."""
self.hass.stop()
@mock.patch("libpurecool.dyson.DysonAccount.login", return_value=False)
def test_dyson_login_failed(self, mocked_login):
"""Test if Dyson connection failed."""
dyson.setup(
self.hass,
{
dyson.DOMAIN: {
dyson.CONF_USERNAME: "email",
dyson.CONF_PASSWORD: "password",
dyson.CONF_LANGUAGE: "FR",
}
},
)
assert mocked_login.call_count == 1
@mock.patch("libpurecool.dyson.DysonAccount.devices", return_value=[])
@mock.patch("libpurecool.dyson.DysonAccount.login", return_value=True)
def test_dyson_login(self, mocked_login, mocked_devices):
"""Test valid connection to dyson web service."""
dyson.setup(
self.hass,
{
dyson.DOMAIN: {
dyson.CONF_USERNAME: "email",
dyson.CONF_PASSWORD: "password",
dyson.CONF_LANGUAGE: "FR",
}
},
)
assert mocked_login.call_count == 1
assert mocked_devices.call_count == 1
assert len(self.hass.data[dyson.DYSON_DEVICES]) == 0
@mock.patch("homeassistant.helpers.discovery.load_platform")
@mock.patch(
"libpurecool.dyson.DysonAccount.devices",
return_value=[_get_dyson_account_device_available()],
)
@mock.patch("libpurecool.dyson.DysonAccount.login", return_value=True)
def test_dyson_custom_conf(self, mocked_login, mocked_devices, mocked_discovery):
"""Test device connection using custom configuration."""
dyson.setup(
self.hass,
{
dyson.DOMAIN: {
dyson.CONF_USERNAME: "email",
dyson.CONF_PASSWORD: "password",
dyson.CONF_LANGUAGE: "FR",
dyson.CONF_DEVICES: [
{"device_id": "XX-XXXXX-XX", "device_ip": "192.168.0.1"}
],
}
},
)
assert mocked_login.call_count == 1
assert mocked_devices.call_count == 1
assert len(self.hass.data[dyson.DYSON_DEVICES]) == 1
assert mocked_discovery.call_count == 5
@mock.patch(
"libpurecool.dyson.DysonAccount.devices",
return_value=[_get_dyson_account_device_not_available()],
)
@mock.patch("libpurecool.dyson.DysonAccount.login", return_value=True)
def test_dyson_custom_conf_device_not_available(self, mocked_login, mocked_devices):
"""Test device connection with an invalid device."""
dyson.setup(
self.hass,
{
dyson.DOMAIN: {
dyson.CONF_USERNAME: "email",
dyson.CONF_PASSWORD: "password",
dyson.CONF_LANGUAGE: "FR",
dyson.CONF_DEVICES: [
{"device_id": "XX-XXXXX-XX", "device_ip": "192.168.0.1"}
],
}
},
)
assert mocked_login.call_count == 1
assert mocked_devices.call_count == 1
assert len(self.hass.data[dyson.DYSON_DEVICES]) == 0
@mock.patch(
"libpurecool.dyson.DysonAccount.devices",
return_value=[_get_dyson_account_device_error()],
)
@mock.patch("libpurecool.dyson.DysonAccount.login", return_value=True)
def test_dyson_custom_conf_device_error(self, mocked_login, mocked_devices):
"""Test device connection with device raising an exception."""
dyson.setup(
self.hass,
{
dyson.DOMAIN: {
dyson.CONF_USERNAME: "email",
dyson.CONF_PASSWORD: "password",
dyson.CONF_LANGUAGE: "FR",
dyson.CONF_DEVICES: [
{"device_id": "XX-XXXXX-XX", "device_ip": "192.168.0.1"}
],
}
},
)
assert mocked_login.call_count == 1
assert mocked_devices.call_count == 1
assert len(self.hass.data[dyson.DYSON_DEVICES]) == 0
@mock.patch("homeassistant.helpers.discovery.load_platform")
@mock.patch(
"libpurecool.dyson.DysonAccount.devices",
return_value=[_get_dyson_account_device_available()],
)
@mock.patch("libpurecool.dyson.DysonAccount.login", return_value=True)
def test_dyson_custom_conf_with_unknown_device(
self, mocked_login, mocked_devices, mocked_discovery
):
"""Test device connection with custom conf and unknown device."""
dyson.setup(
self.hass,
{
dyson.DOMAIN: {
dyson.CONF_USERNAME: "email",
dyson.CONF_PASSWORD: "password",
dyson.CONF_LANGUAGE: "FR",
dyson.CONF_DEVICES: [
{"device_id": "XX-XXXXX-XY", "device_ip": "192.168.0.1"}
],
}
},
)
assert mocked_login.call_count == 1
assert mocked_devices.call_count == 1
assert len(self.hass.data[dyson.DYSON_DEVICES]) == 0
assert mocked_discovery.call_count == 0
@mock.patch("homeassistant.helpers.discovery.load_platform")
@mock.patch(
"libpurecool.dyson.DysonAccount.devices",
return_value=[_get_dyson_account_device_available()],
)
@mock.patch("libpurecool.dyson.DysonAccount.login", return_value=True)
def test_dyson_discovery(self, mocked_login, mocked_devices, mocked_discovery):
"""Test device connection using discovery."""
dyson.setup(
self.hass,
{
dyson.DOMAIN: {
dyson.CONF_USERNAME: "email",
dyson.CONF_PASSWORD: "password",
dyson.CONF_LANGUAGE: "FR",
dyson.CONF_TIMEOUT: 5,
dyson.CONF_RETRY: 2,
}
},
)
assert mocked_login.call_count == 1
assert mocked_devices.call_count == 1
assert len(self.hass.data[dyson.DYSON_DEVICES]) == 1
assert mocked_discovery.call_count == 5
@mock.patch(
"libpurecool.dyson.DysonAccount.devices",
return_value=[_get_dyson_account_device_not_available()],
)
@mock.patch("libpurecool.dyson.DysonAccount.login", return_value=True)
def test_dyson_discovery_device_not_available(self, mocked_login, mocked_devices):
"""Test device connection with discovery and invalid device."""
dyson.setup(
self.hass,
{
dyson.DOMAIN: {
dyson.CONF_USERNAME: "email",
dyson.CONF_PASSWORD: "password",
dyson.CONF_LANGUAGE: "FR",
dyson.CONF_TIMEOUT: 5,
dyson.CONF_RETRY: 2,
}
},
)
assert mocked_login.call_count == 1
assert mocked_devices.call_count == 1
assert len(self.hass.data[dyson.DYSON_DEVICES]) == 0
async def test_login_failed(hass: HomeAssistant):
"""Test login failure during setup."""
with patch(f"{BASE_PATH}.DysonAccount.login", return_value=False):
assert not await async_setup_component(hass, DOMAIN, CONFIG)
await hass.async_block_till_done()

View file

@ -79,7 +79,7 @@ def _async_assign_values(
@callback
def get_device(spec: Type[DysonPureCoolLink], combi=False) -> DysonPureCoolLink:
def async_get_device(spec: Type[DysonPureCoolLink], combi=False) -> DysonPureCoolLink:
"""Return a device of the given type."""
device = get_basic_device(spec)
_async_assign_values(device, combi=combi)
@ -165,7 +165,7 @@ async def test_temperature(
"""Test the temperature sensor in different units."""
hass.config.units = unit_system
device = get_device(DysonPureCoolLink)
device = async_get_device(DysonPureCoolLink)
with patch(f"{BASE_PATH}.DysonAccount.login", return_value=True), patch(
f"{BASE_PATH}.DysonAccount.devices", return_value=[device]
), patch(f"{BASE_PATH}.DYSON_PLATFORMS", [PLATFORM_DOMAIN]):

View file

@ -26,20 +26,21 @@ from homeassistant.const import (
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import entity_registry
from .common import ENTITY_NAME, NAME, SERIAL, async_update_device, get_basic_device
from .common import (
ENTITY_NAME,
NAME,
SERIAL,
async_get_360eye_device,
async_update_device,
)
ENTITY_ID = f"{PLATFORM_DOMAIN}.{ENTITY_NAME}"
@callback
def get_device(state=Dyson360EyeMode.FULL_CLEAN_RUNNING) -> Dyson360Eye:
def async_get_device(state=Dyson360EyeMode.FULL_CLEAN_RUNNING) -> Dyson360Eye:
"""Return a Dyson 360 Eye device."""
device = get_basic_device(Dyson360Eye)
device.state.state = state
device.state.battery_level = 85
device.state.power_mode = PowerMode.QUIET
device.state.position = (0, 0)
return device
return async_get_360eye_device(state)
async def test_state(hass: HomeAssistant, device: Dyson360Eye) -> None: