fix the tests

This commit is contained in:
Bas Nijholt 2019-11-11 20:44:12 +01:00
parent 583c785611
commit b2973cda6c
6 changed files with 194 additions and 242 deletions

View File

@ -516,7 +516,7 @@ class SmartDevice:
if not await self.get_has_emeter(): if not await self.get_has_emeter():
raise SmartDeviceException("Device has no emeter") raise SmartDeviceException("Device has no emeter")
response = EmeterStatus(self.get_emeter_realtime()) response = EmeterStatus(await self.get_emeter_realtime())
return response["power"] return response["power"]
async def reboot(self, delay=1) -> None: async def reboot(self, delay=1) -> None:
@ -580,8 +580,7 @@ class SmartDevice:
def is_strip(self) -> bool: def is_strip(self) -> bool:
return self._device_type == DeviceType.Strip return self._device_type == DeviceType.Strip
@property async def is_dimmable(self):
def is_dimmable(self):
return False return False
@property @property

View File

@ -68,7 +68,7 @@ class SmartPlug(SmartDevice):
if not isinstance(value, int): if not isinstance(value, int):
raise ValueError("Brightness must be integer, " "not of %s.", type(value)) raise ValueError("Brightness must be integer, " "not of %s.", type(value))
elif 0 < value <= 100: elif 0 < value <= 100:
self.turn_on() await self.turn_on()
await self._query_helper( await self._query_helper(
"smartlife.iot.dimmer", "set_brightness", {"brightness": value} "smartlife.iot.dimmer", "set_brightness", {"brightness": value}
) )

View File

@ -56,7 +56,11 @@ class SmartStrip(SmartPlug):
self.num_children = len(children) self.num_children = len(children)
for plug in range(self.num_children): for plug in range(self.num_children):
self.plugs[plug] = SmartPlug( self.plugs[plug] = SmartPlug(
host, protocol, context=children[plug]["id"], cache_ttl=cache_ttl host,
protocol,
context=children[plug]["id"],
cache_ttl=cache_ttl,
ioloop=ioloop,
) )
def raise_for_index(self, index: int): def raise_for_index(self, index: int):
@ -165,12 +169,12 @@ class SmartStrip(SmartPlug):
self.raise_for_index(index) self.raise_for_index(index)
await self.plugs[index].turn_off() await self.plugs[index].turn_off()
async def get_on_since(self) -> datetime: async def get_max_on_since(self) -> datetime:
"""Returns the maximum on-time of all outlets.""" """Returns the maximum on-time of all outlets."""
on_since = await self._get_on_since() on_since = await self.get_on_since(index=-1)
return max(v for v in await on_since.values()) return max(v for v in on_since.values())
async def _get_on_since(self, *, index: int = -1) -> Any: async def get_on_since(self, *, index: Optional[int] = None) -> Any:
""" """
Returns pretty-printed on-time Returns pretty-printed on-time
@ -180,6 +184,9 @@ class SmartStrip(SmartPlug):
Dict[int, str] without index Dict[int, str] without index
:raises SmartStripException: index out of bounds :raises SmartStripException: index out of bounds
""" """
if index is None:
return await self.get_max_on_since()
if index < 0: if index < 0:
on_since = {} on_since = {}
sys_info = await self.get_sys_info() sys_info = await self.get_sys_info()
@ -204,7 +211,7 @@ class SmartStrip(SmartPlug):
""" """
state = {"LED state": await self.get_led()} # XXX: from where? state = {"LED state": await self.get_led()} # XXX: from where?
is_on = await self.get_is_on() is_on = await self.get_is_on()
on_since = await self._get_on_since() on_since = await self.get_on_since(index=-1)
for plug_index in range(self.num_children): for plug_index in range(self.num_children):
plug_number = plug_index + 1 plug_number = plug_index + 1
if is_on[plug_index]: if is_on[plug_index]:
@ -266,14 +273,13 @@ class SmartStrip(SmartPlug):
response = EmeterStatus(await self.get_emeter_realtime(index=index)) response = EmeterStatus(await self.get_emeter_realtime(index=index))
return response["power"] return response["power"]
@property async def get_icon(self):
def icon(self):
"""Override for base class icon property, SmartStrip and children do not """Override for base class icon property, SmartStrip and children do not
have icons so we return dummy strings. have icons so we return dummy strings.
""" """
return {"icon": "SMARTSTRIP-DUMMY", "hash": "SMARTSTRIP-DUMMY"} return {"icon": "SMARTSTRIP-DUMMY", "hash": "SMARTSTRIP-DUMMY"}
async def get_alias(self, *, index: int = -1) -> Union[str, Dict[int, str]]: async def get_alias(self, *, index: Optional[int] = None) -> Union[str, Dict[int, str]]:
"""Gets the alias for a plug. """Gets the alias for a plug.
:param index: plug index (-1 for all) :param index: plug index (-1 for all)
@ -283,6 +289,9 @@ class SmartStrip(SmartPlug):
Dict[int, str] if no index provided Dict[int, str] if no index provided
:raises SmartStripException: index out of bounds :raises SmartStripException: index out of bounds
""" """
if index is None:
return await super().get_alias()
sys_info = await self.get_sys_info() sys_info = await self.get_sys_info()
children = sys_info["children"] children = sys_info["children"]
@ -295,7 +304,7 @@ class SmartStrip(SmartPlug):
self.raise_for_index(index) self.raise_for_index(index)
return children[index]["alias"] return children[index]["alias"]
async def set_alias(self, alias: str, *, index: int = -1): async def set_alias(self, alias: str, *, index: Optional[int] = None):
"""Sets the alias for a plug """Sets the alias for a plug
:param index: plug index :param index: plug index
@ -304,8 +313,8 @@ class SmartStrip(SmartPlug):
:raises SmartStripException: index out of bounds :raises SmartStripException: index out of bounds
""" """
# Renaming the whole strip # Renaming the whole strip
if index < 0: if index is None:
return super().set_alias(alias) return await super().set_alias(alias)
self.raise_for_index(index) self.raise_for_index(index)
await self.plugs[index].set_alias(alias) await self.plugs[index].set_alias(alias)

View File

@ -36,6 +36,12 @@ def filter_model(filter):
return filtered return filtered
def get_ioloop():
ioloop = asyncio.new_event_loop()
asyncio.set_event_loop(ioloop)
return ioloop
has_emeter = pytest.mark.parametrize("dev", filter_model(EMETER), indirect=True) has_emeter = pytest.mark.parametrize("dev", filter_model(EMETER), indirect=True)
no_emeter = pytest.mark.parametrize( no_emeter = pytest.mark.parametrize(
"dev", filter_model(ALL_DEVICES - EMETER), indirect=True "dev", filter_model(ALL_DEVICES - EMETER), indirect=True
@ -69,18 +75,19 @@ turn_on = pytest.mark.parametrize("turn_on", [True, False])
def handle_turn_on(dev, turn_on): def handle_turn_on(dev, turn_on):
if turn_on: if turn_on:
dev.turn_on() dev.sync.turn_on()
else: else:
dev.turn_off() dev.sync.turn_off()
@pytest.fixture(params=SUPPORTED_DEVICES) @pytest.fixture(params=SUPPORTED_DEVICES)
def dev(request): def dev(request):
ioloop = get_ioloop()
file = request.param file = request.param
ip = request.config.getoption("--ip") ip = request.config.getoption("--ip")
if ip: if ip:
d = asyncio.run(Discover.discover_single(ip)) d = ioloop.run_until_complete(Discover.discover_single(ip))
print(d.model) print(d.model)
if d.model in file: if d.model in file:
return d return d
@ -95,11 +102,11 @@ def dev(request):
"cache_ttl": 0, "cache_ttl": 0,
} }
if "LB" in model or "KL" in model: if "LB" in model or "KL" in model:
p = SmartBulb(**params) p = SmartBulb(**params, ioloop=ioloop)
elif "HS300" in model: elif "HS300" in model:
p = SmartStrip(**params) p = SmartStrip(**params, ioloop=ioloop)
elif "HS" in model: elif "HS" in model:
p = SmartPlug(**params) p = SmartPlug(**params, ioloop=ioloop)
else: else:
raise Exception("No tests for %s" % model) raise Exception("No tests for %s" % model)
yield p yield p

View File

@ -290,14 +290,6 @@ class FakeTransportProtocol(TPLinkSmartHomeProtocol):
else: else:
self.proto["system"]["get_sysinfo"]["relay_state"] = x["state"] self.proto["system"]["get_sysinfo"]["relay_state"] = x["state"]
def set_alias_old(self, x):
_LOGGER.debug("Setting alias to %s", x["alias"])
self.proto["system"]["get_sysinfo"]["alias"] = x["alias"]
def set_relay_state_old(self, x):
_LOGGER.debug("Setting relay state to %s", x)
self.proto["system"]["get_sysinfo"]["relay_state"] = x["state"]
def set_led_off(self, x, *args): def set_led_off(self, x, *args):
_LOGGER.debug("Setting led off to %s", x) _LOGGER.debug("Setting led off to %s", x)
self.proto["system"]["get_sysinfo"]["led_off"] = x["off"] self.proto["system"]["get_sysinfo"]["led_off"] = x["off"]
@ -398,7 +390,7 @@ class FakeTransportProtocol(TPLinkSmartHomeProtocol):
"smartlife.iot.dimmer": {"set_brightness": set_hs220_brightness}, "smartlife.iot.dimmer": {"set_brightness": set_hs220_brightness},
} }
def query(self, host, request, port=9999): async def query(self, host, request, port=9999):
proto = self.proto proto = self.proto
# collect child ids from context # collect child ids from context

View File

@ -4,7 +4,7 @@ from unittest.mock import patch
import pytest import pytest
from pyHS100 import DeviceType, SmartDeviceException, SmartStripException from pyHS100 import DeviceType, SmartDeviceException, SmartStripException, SmartStrip
from .conftest import ( from .conftest import (
bulb, bulb,
@ -32,10 +32,10 @@ from .newfakes import (
@plug @plug
def test_plug_sysinfo(dev): def test_plug_sysinfo(dev):
assert dev.sys_info is not None assert dev.sync.get_sys_info() is not None
PLUG_SCHEMA(dev.sys_info) PLUG_SCHEMA(dev.sync.get_sys_info())
assert dev.model is not None assert dev.sync.get_model() is not None
assert dev.device_type == DeviceType.Plug or dev.device_type == DeviceType.Strip assert dev.device_type == DeviceType.Plug or dev.device_type == DeviceType.Strip
assert dev.is_plug or dev.is_strip assert dev.is_plug or dev.is_strip
@ -43,99 +43,63 @@ def test_plug_sysinfo(dev):
@bulb @bulb
def test_bulb_sysinfo(dev): def test_bulb_sysinfo(dev):
assert dev.sys_info is not None assert dev.sync.get_sys_info() is not None
BULB_SCHEMA(dev.sys_info) BULB_SCHEMA(dev.sync.get_sys_info())
assert dev.model is not None assert dev.sync.get_model() is not None
assert dev.device_type == DeviceType.Bulb assert dev.device_type == DeviceType.Bulb
assert dev.is_bulb assert dev.is_bulb
def test_state_info(dev): def test_state_info(dev):
assert isinstance(dev.state_information, dict) assert isinstance(dev.sync.get_state_information(), dict)
def test_invalid_connection(dev): def test_invalid_connection(dev):
with patch.object(FakeTransportProtocol, "query", side_effect=SmartDeviceException): with patch.object(FakeTransportProtocol, "query", side_effect=SmartDeviceException):
with pytest.raises(SmartDeviceException): with pytest.raises(SmartDeviceException):
dev.is_on dev.sync.is_on()
def test_query_helper(dev): def test_query_helper(dev):
with pytest.raises(SmartDeviceException): with pytest.raises(SmartDeviceException):
asyncio.run(dev._query_helper("test", "testcmd", {})) dev.sync._query_helper("test", "testcmd", {})
# TODO check for unwrapping? # TODO check for unwrapping?
def test_deprecated_state(dev):
with pytest.deprecated_call():
dev.state = "OFF"
assert dev.state == "OFF"
assert not dev.is_on
with pytest.deprecated_call():
dev.state = "ON"
assert dev.state == "ON"
assert dev.is_on
with pytest.deprecated_call():
with pytest.raises(ValueError):
dev.state = "foo"
with pytest.deprecated_call():
with pytest.raises(ValueError):
dev.state = 1234
def test_deprecated_alias(dev):
with pytest.deprecated_call():
dev.alias = "foo"
def test_deprecated_mac(dev):
with pytest.deprecated_call():
dev.mac = 123123123123
@plug
def test_deprecated_led(dev):
with pytest.deprecated_call():
dev.led = True
@turn_on @turn_on
def test_state(dev, turn_on): def test_state(dev, turn_on):
handle_turn_on(dev, turn_on) handle_turn_on(dev, turn_on)
orig_state = dev.is_on orig_state = dev.sync.is_on()
if orig_state: if orig_state:
dev.turn_off() dev.sync.turn_off()
assert not dev.is_on assert not dev.sync.is_on()
assert dev.is_off assert dev.sync.is_off()
dev.turn_on() dev.sync.turn_on()
assert dev.is_on assert dev.sync.is_on()
assert not dev.is_off assert not dev.sync.is_off()
else: else:
dev.turn_on() dev.sync.turn_on()
assert dev.is_on assert dev.sync.is_on()
assert not dev.is_off assert not dev.sync.is_off()
dev.turn_off() dev.sync.turn_off()
assert not dev.is_on assert not dev.sync.is_on()
assert dev.is_off assert dev.sync.is_off()
@no_emeter @no_emeter
def test_no_emeter(dev): def test_no_emeter(dev):
assert not dev.has_emeter assert not dev.sync.get_has_emeter()
with pytest.raises(SmartDeviceException): with pytest.raises(SmartDeviceException):
dev.get_emeter_realtime() dev.sync.get_emeter_realtime()
with pytest.raises(SmartDeviceException): with pytest.raises(SmartDeviceException):
dev.get_emeter_daily() dev.sync.get_emeter_daily()
with pytest.raises(SmartDeviceException): with pytest.raises(SmartDeviceException):
dev.get_emeter_monthly() dev.sync.get_emeter_monthly()
with pytest.raises(SmartDeviceException): with pytest.raises(SmartDeviceException):
dev.erase_emeter_stats() dev.sync.erase_emeter_stats()
@has_emeter @has_emeter
@ -143,9 +107,9 @@ def test_get_emeter_realtime(dev):
if dev.is_strip: if dev.is_strip:
pytest.skip("Disabled for HS300 temporarily") pytest.skip("Disabled for HS300 temporarily")
assert dev.has_emeter assert dev.sync.get_has_emeter()
current_emeter = dev.get_emeter_realtime() current_emeter = dev.sync.get_emeter_realtime()
CURRENT_CONSUMPTION_SCHEMA(current_emeter) CURRENT_CONSUMPTION_SCHEMA(current_emeter)
@ -154,11 +118,11 @@ def test_get_emeter_daily(dev):
if dev.is_strip: if dev.is_strip:
pytest.skip("Disabled for HS300 temporarily") pytest.skip("Disabled for HS300 temporarily")
assert dev.has_emeter assert dev.sync.get_has_emeter()
assert dev.get_emeter_daily(year=1900, month=1) == {} assert dev.sync.get_emeter_daily(year=1900, month=1) == {}
d = dev.get_emeter_daily() d = dev.sync.get_emeter_daily()
assert len(d) > 0 assert len(d) > 0
k, v = d.popitem() k, v = d.popitem()
@ -166,7 +130,7 @@ def test_get_emeter_daily(dev):
assert isinstance(v, float) assert isinstance(v, float)
# Test kwh (energy, energy_wh) # Test kwh (energy, energy_wh)
d = dev.get_emeter_daily(kwh=False) d = dev.sync.get_emeter_daily(kwh=False)
k2, v2 = d.popitem() k2, v2 = d.popitem()
assert v * 1000 == v2 assert v * 1000 == v2
@ -176,11 +140,11 @@ def test_get_emeter_monthly(dev):
if dev.is_strip: if dev.is_strip:
pytest.skip("Disabled for HS300 temporarily") pytest.skip("Disabled for HS300 temporarily")
assert dev.has_emeter assert dev.sync.get_has_emeter()
assert dev.get_emeter_monthly(year=1900) == {} assert dev.sync.get_emeter_monthly(year=1900) == {}
d = dev.get_emeter_monthly() d = dev.sync.get_emeter_monthly()
assert len(d) > 0 assert len(d) > 0
k, v = d.popitem() k, v = d.popitem()
@ -188,7 +152,7 @@ def test_get_emeter_monthly(dev):
assert isinstance(v, float) assert isinstance(v, float)
# Test kwh (energy, energy_wh) # Test kwh (energy, energy_wh)
d = dev.get_emeter_monthly(kwh=False) d = dev.sync.get_emeter_monthly(kwh=False)
k2, v2 = d.popitem() k2, v2 = d.popitem()
assert v * 1000 == v2 assert v * 1000 == v2
@ -198,9 +162,9 @@ def test_emeter_status(dev):
if dev.is_strip: if dev.is_strip:
pytest.skip("Disabled for HS300 temporarily") pytest.skip("Disabled for HS300 temporarily")
assert dev.has_emeter assert dev.sync.get_has_emeter()
d = dev.get_emeter_realtime() d = dev.sync.get_emeter_realtime()
with pytest.raises(KeyError): with pytest.raises(KeyError):
assert d["foo"] assert d["foo"]
@ -217,9 +181,9 @@ def test_emeter_status(dev):
@pytest.mark.skip("not clearing your stats..") @pytest.mark.skip("not clearing your stats..")
@has_emeter @has_emeter
def test_erase_emeter_stats(dev): def test_erase_emeter_stats(dev):
assert dev.has_emeter assert dev.sync.get_has_emeter()
dev.erase_emeter() dev.sync.erase_emeter()
@has_emeter @has_emeter
@ -227,144 +191,145 @@ def test_current_consumption(dev):
if dev.is_strip: if dev.is_strip:
pytest.skip("Disabled for HS300 temporarily") pytest.skip("Disabled for HS300 temporarily")
if dev.has_emeter: if dev.sync.get_has_emeter():
x = dev.current_consumption() x = dev.sync.current_consumption()
assert isinstance(x, float) assert isinstance(x, float)
assert x >= 0.0 assert x >= 0.0
else: else:
assert dev.current_consumption() is None assert dev.sync.current_consumption() is None
def test_alias(dev): def test_alias(dev):
test_alias = "TEST1234" test_alias = "TEST1234"
original = dev.alias original = dev.sync.get_alias()
assert isinstance(original, str) assert isinstance(original, str)
dev.set_alias(test_alias) dev.sync.set_alias(test_alias)
assert dev.alias == test_alias assert dev.sync.get_alias() == test_alias
dev.set_alias(original) dev.sync.set_alias(original)
assert dev.alias == original assert dev.sync.get_alias() == original
@plug @plug
def test_led(dev): def test_led(dev):
original = dev.led original = dev.sync.get_led()
dev.set_led(False) dev.sync.set_led(False)
assert not dev.led assert not dev.sync.get_led()
dev.set_led(True) dev.sync.set_led(True)
assert dev.led assert dev.sync.get_led()
dev.set_led(original) dev.sync.set_led(original)
@plug @plug
def test_on_since(dev): def test_on_since(dev):
assert isinstance(dev.on_since, datetime.datetime) assert isinstance(dev.sync.get_on_since(), datetime.datetime)
def test_icon(dev): def test_icon(dev):
assert set(dev.icon.keys()), {"icon", "hash"} assert set(dev.sync.get_icon().keys()), {"icon", "hash"}
def test_time(dev): def test_time(dev):
assert isinstance(dev.time, datetime.datetime) assert isinstance(dev.sync.get_time(), datetime.datetime)
# TODO check setting? # TODO check setting?
def test_timezone(dev): def test_timezone(dev):
TZ_SCHEMA(dev.timezone) TZ_SCHEMA(dev.sync.get_timezone())
def test_hw_info(dev): def test_hw_info(dev):
PLUG_SCHEMA(dev.hw_info) PLUG_SCHEMA(dev.sync.get_hw_info())
def test_location(dev): def test_location(dev):
PLUG_SCHEMA(dev.location) PLUG_SCHEMA(dev.sync.get_location())
def test_rssi(dev): def test_rssi(dev):
PLUG_SCHEMA({"rssi": dev.rssi}) # wrapping for vol PLUG_SCHEMA({"rssi": dev.sync.get_rssi()}) # wrapping for vol
def test_mac(dev): def test_mac(dev):
PLUG_SCHEMA({"mac": dev.mac}) # wrapping for val PLUG_SCHEMA({"mac": dev.sync.get_mac()}) # wrapping for val
# TODO check setting? # TODO check setting?
@non_variable_temp @non_variable_temp
def test_temperature_on_nonsupporting(dev): def test_temperature_on_nonsupporting(dev):
assert dev.valid_temperature_range == (0, 0) assert dev.sync.get_valid_temperature_range() == (0, 0)
# TODO test when device does not support temperature range # TODO test when device does not support temperature range
with pytest.raises(SmartDeviceException): with pytest.raises(SmartDeviceException):
dev.set_color_temp(2700) dev.sync.set_color_temp(2700)
with pytest.raises(SmartDeviceException): with pytest.raises(SmartDeviceException):
print(dev.color_temp) print(dev.sync.get_color_temp())
@variable_temp @variable_temp
def test_out_of_range_temperature(dev): def test_out_of_range_temperature(dev):
with pytest.raises(ValueError): with pytest.raises(ValueError):
dev.set_color_temp(1000) dev.sync.set_color_temp(1000)
with pytest.raises(ValueError): with pytest.raises(ValueError):
dev.set_color_temp(10000) dev.sync.set_color_temp(10000)
@non_dimmable @non_dimmable
def test_non_dimmable(dev): def test_non_dimmable(dev):
assert not dev.is_dimmable assert not dev.sync.is_dimmable()
with pytest.raises(SmartDeviceException): with pytest.raises(SmartDeviceException):
assert dev.brightness == 0 assert dev.sync.get_brightness() == 0
with pytest.raises(SmartDeviceException): with pytest.raises(SmartDeviceException):
dev.set_brightness(100) dev.sync.set_brightness(100)
@dimmable @dimmable
@turn_on @turn_on
def test_dimmable_brightness(dev, turn_on): def test_dimmable_brightness(dev, turn_on):
handle_turn_on(dev, turn_on) handle_turn_on(dev, turn_on)
assert dev.is_dimmable assert dev.sync.is_dimmable()
dev.set_brightness(50) dev.sync.set_brightness(50)
assert dev.brightness == 50 assert dev.sync.get_brightness() == 50
dev.set_brightness(10) dev.sync.set_brightness(10)
assert dev.brightness == 10 assert dev.sync.get_brightness() == 10
with pytest.raises(ValueError): with pytest.raises(ValueError):
dev.set_brightness("foo") dev.sync.set_brightness("foo")
@dimmable @dimmable
def test_invalid_brightness(dev): def test_invalid_brightness(dev):
assert dev.is_dimmable assert dev.sync.is_dimmable()
with pytest.raises(ValueError): with pytest.raises(ValueError):
dev.set_brightness(110) dev.sync.set_brightness(110)
with pytest.raises(ValueError): with pytest.raises(ValueError):
dev.set_brightness(-100) dev.sync.set_brightness(-100)
@color_bulb @color_bulb
@turn_on @turn_on
def test_hsv(dev, turn_on): def test_hsv(dev, turn_on):
handle_turn_on(dev, turn_on) handle_turn_on(dev, turn_on)
assert dev.is_color assert dev.sync.is_color()
hue, saturation, brightness = dev.hsv hue, saturation, brightness = dev.sync.get_hsv()
assert 0 <= hue <= 255 assert 0 <= hue <= 255
assert 0 <= saturation <= 100 assert 0 <= saturation <= 100
assert 0 <= brightness <= 100 assert 0 <= brightness <= 100
dev.set_hsv(hue=1, saturation=1, value=1) dev.sync.set_hsv(hue=1, saturation=1, value=1)
hue, saturation, brightness = dev.hsv hue, saturation, brightness = dev.sync.get_hsv()
assert hue == 1 assert hue == 1
assert saturation == 1 assert saturation == 1
assert brightness == 1 assert brightness == 1
@ -375,29 +340,29 @@ def test_hsv(dev, turn_on):
def test_invalid_hsv(dev, turn_on): def test_invalid_hsv(dev, turn_on):
handle_turn_on(dev, turn_on) handle_turn_on(dev, turn_on)
assert dev.is_color assert dev.sync.is_color()
for invalid_hue in [-1, 361, 0.5]: for invalid_hue in [-1, 361, 0.5]:
with pytest.raises(ValueError): with pytest.raises(ValueError):
dev.set_hsv(invalid_hue, 0, 0) dev.sync.set_hsv(invalid_hue, 0, 0)
for invalid_saturation in [-1, 101, 0.5]: for invalid_saturation in [-1, 101, 0.5]:
with pytest.raises(ValueError): with pytest.raises(ValueError):
dev.set_hsv(0, invalid_saturation, 0) dev.sync.set_hsv(0, invalid_saturation, 0)
for invalid_brightness in [-1, 101, 0.5]: for invalid_brightness in [-1, 101, 0.5]:
with pytest.raises(ValueError): with pytest.raises(ValueError):
dev.set_hsv(0, 0, invalid_brightness) dev.sync.set_hsv(0, 0, invalid_brightness)
@non_color_bulb @non_color_bulb
def test_hsv_on_non_color(dev): def test_hsv_on_non_color(dev):
assert not dev.is_color assert not dev.sync.is_color()
with pytest.raises(SmartDeviceException): with pytest.raises(SmartDeviceException):
dev.set_hsv(0, 0, 0) dev.sync.set_hsv(0, 0, 0)
with pytest.raises(SmartDeviceException): with pytest.raises(SmartDeviceException):
print(dev.hsv) print(dev.sync.get_hsv())
@variable_temp @variable_temp
@ -405,43 +370,21 @@ def test_hsv_on_non_color(dev):
def test_try_set_colortemp(dev, turn_on): def test_try_set_colortemp(dev, turn_on):
handle_turn_on(dev, turn_on) handle_turn_on(dev, turn_on)
dev.set_color_temp(2700) dev.sync.set_color_temp(2700)
assert dev.color_temp == 2700 assert dev.sync.get_color_temp() == 2700
@variable_temp
@turn_on
def test_deprecated_colortemp(dev, turn_on):
handle_turn_on(dev, turn_on)
with pytest.deprecated_call():
dev.color_temp = 2700
@dimmable
def test_deprecated_brightness(dev):
with pytest.deprecated_call():
dev.brightness = 10
@non_variable_temp @non_variable_temp
def test_non_variable_temp(dev): def test_non_variable_temp(dev):
with pytest.raises(SmartDeviceException): with pytest.raises(SmartDeviceException):
dev.set_color_temp(2700) dev.sync.set_color_temp(2700)
@color_bulb
@turn_on
def test_deprecated_hsv(dev, turn_on):
handle_turn_on(dev, turn_on)
with pytest.deprecated_call():
dev.hsv = (1, 1, 1)
@strip @strip
def test_children_is_on(dev): def test_children_is_on(dev):
is_on = dev.get_is_on() is_on = dev.sync.get_is_on()
for i in range(dev.num_children): for i in range(dev.num_children):
assert is_on[i] == dev.get_is_on(index=i) assert is_on[i] == dev.sync.get_is_on(index=i)
@strip @strip
@ -449,22 +392,22 @@ def test_children_is_on(dev):
def test_children_change_state(dev, turn_on): def test_children_change_state(dev, turn_on):
handle_turn_on(dev, turn_on) handle_turn_on(dev, turn_on)
for i in range(dev.num_children): for i in range(dev.num_children):
orig_state = dev.get_is_on(index=i) orig_state = dev.sync.get_is_on(index=i)
if orig_state: if orig_state:
dev.turn_off(index=i) dev.sync.turn_off(index=i)
assert not dev.get_is_on(index=i) assert not dev.sync.get_is_on(index=i)
assert dev.get_is_off(index=i) assert dev.sync.get_is_off(index=i)
dev.turn_on(index=i) dev.sync.turn_on(index=i)
assert dev.get_is_on(index=i) assert dev.sync.get_is_on(index=i)
assert not dev.get_is_off(index=i) assert not dev.sync.get_is_off(index=i)
else: else:
dev.turn_on(index=i) dev.sync.turn_on(index=i)
assert dev.get_is_on(index=i) assert dev.sync.get_is_on(index=i)
assert not dev.get_is_off(index=i) assert not dev.sync.get_is_off(index=i)
dev.turn_off(index=i) dev.sync.turn_off(index=i)
assert not dev.get_is_on(index=i) assert not dev.sync.get_is_on(index=i)
assert dev.get_is_off(index=i) assert dev.sync.get_is_off(index=i)
@strip @strip
@ -472,32 +415,32 @@ def test_children_bounds(dev):
out_of_bounds = dev.num_children + 100 out_of_bounds = dev.num_children + 100
with pytest.raises(SmartDeviceException): with pytest.raises(SmartDeviceException):
dev.turn_off(index=out_of_bounds) dev.sync.turn_off(index=out_of_bounds)
with pytest.raises(SmartDeviceException): with pytest.raises(SmartDeviceException):
dev.turn_on(index=out_of_bounds) dev.sync.turn_on(index=out_of_bounds)
with pytest.raises(SmartDeviceException): with pytest.raises(SmartDeviceException):
dev.get_is_on(index=out_of_bounds) dev.sync.get_is_on(index=out_of_bounds)
with pytest.raises(SmartDeviceException): with pytest.raises(SmartDeviceException):
dev.get_alias(index=out_of_bounds) dev.sync.get_alias(index=out_of_bounds)
with pytest.raises(SmartDeviceException): with pytest.raises(SmartDeviceException):
dev.get_on_since(index=out_of_bounds) dev.sync.get_on_since(index=out_of_bounds)
@strip @strip
def test_children_alias(dev): def test_children_alias(dev):
original = dev.get_alias() original = dev.sync.get_alias()
test_alias = "TEST1234" test_alias = "TEST1234"
for idx in range(dev.num_children): for idx in range(dev.num_children):
dev.set_alias(alias=test_alias, index=idx) dev.sync.set_alias(alias=test_alias, index=idx)
assert dev.get_alias(index=idx) == test_alias assert dev.sync.get_alias(index=idx) == test_alias
dev.set_alias(alias=original[idx], index=idx) dev.sync.set_alias(alias=original[idx], index=idx)
assert dev.get_alias(index=idx) == original[idx] assert dev.sync.get_alias(index=idx) == original[idx]
@strip @strip
def test_children_on_since(dev): def test_children_on_since(dev):
for idx in range(dev.num_children): for idx in range(dev.num_children):
assert dev.get_on_since(index=idx) assert dev.sync.get_on_since(index=idx)
@pytest.mark.skip("this test will wear out your relays") @pytest.mark.skip("this test will wear out your relays")
@ -510,12 +453,12 @@ def test_all_binary_states(dev):
state_map[plug_index] = bool((state >> plug_index) & 1) state_map[plug_index] = bool((state >> plug_index) & 1)
if state_map[plug_index]: if state_map[plug_index]:
dev.turn_on(index=plug_index) dev.sync.turn_on(index=plug_index)
else: else:
dev.turn_off(index=plug_index) dev.sync.turn_off(index=plug_index)
# check state map applied # check state map applied
for index, state in dev.get_is_on().items(): for index, state in dev.sync.get_is_on().items():
assert state_map[index] == state assert state_map[index] == state
# toggle each outlet with state map applied # toggle each outlet with state map applied
@ -523,12 +466,12 @@ def test_all_binary_states(dev):
# toggle state # toggle state
if state_map[plug_index]: if state_map[plug_index]:
dev.turn_off(index=plug_index) dev.sync.turn_off(index=plug_index)
else: else:
dev.turn_on(index=plug_index) dev.sync.turn_on(index=plug_index)
# only target outlet should have state changed # only target outlet should have state changed
for index, state in dev.get_is_on().items(): for index, state in dev.sync.get_is_on().items():
if index == plug_index: if index == plug_index:
assert state != state_map[index] assert state != state_map[index]
else: else:
@ -536,41 +479,41 @@ def test_all_binary_states(dev):
# reset state # reset state
if state_map[plug_index]: if state_map[plug_index]:
dev.turn_on(index=plug_index) dev.sync.turn_on(index=plug_index)
else: else:
dev.turn_off(index=plug_index) dev.sync.turn_off(index=plug_index)
# original state map should be restored # original state map should be restored
for index, state in dev.get_is_on().items(): for index, state in dev.sync.get_is_on().items():
assert state == state_map[index] assert state == state_map[index]
@strip @strip
def test_children_get_emeter_realtime(dev): def test_children_get_emeter_realtime(dev):
assert dev.has_emeter assert dev.sync.get_has_emeter()
# test with index # test with index
for plug_index in range(dev.num_children): for plug_index in range(dev.num_children):
emeter = dev.get_emeter_realtime(index=plug_index) emeter = dev.sync.get_emeter_realtime(index=plug_index)
CURRENT_CONSUMPTION_SCHEMA(emeter) CURRENT_CONSUMPTION_SCHEMA(emeter)
# test without index # test without index
for index, emeter in dev.get_emeter_realtime().items(): for index, emeter in dev.sync.get_emeter_realtime().items():
CURRENT_CONSUMPTION_SCHEMA(emeter) CURRENT_CONSUMPTION_SCHEMA(emeter)
# out of bounds # out of bounds
with pytest.raises(SmartStripException): with pytest.raises(SmartStripException):
dev.get_emeter_realtime(index=dev.num_children + 100) dev.sync.get_emeter_realtime(index=dev.num_children + 100)
@strip @strip
def test_children_get_emeter_daily(dev): def test_children_get_emeter_daily(dev):
assert dev.has_emeter assert dev.sync.get_has_emeter()
# test with index # test with index
for plug_index in range(dev.num_children): for plug_index in range(dev.num_children):
emeter = dev.get_emeter_daily(year=1900, month=1, index=plug_index) emeter = dev.sync.get_emeter_daily(year=1900, month=1, index=plug_index)
assert emeter == {} assert emeter == {}
emeter = dev.get_emeter_daily(index=plug_index) emeter = dev.sync.get_emeter_daily(index=plug_index)
assert len(emeter) > 0 assert len(emeter) > 0
k, v = emeter.popitem() k, v = emeter.popitem()
@ -578,11 +521,11 @@ def test_children_get_emeter_daily(dev):
assert isinstance(v, float) assert isinstance(v, float)
# test without index # test without index
all_emeter = dev.get_emeter_daily(year=1900, month=1) all_emeters = dev.sync.get_emeter_daily(year=1900, month=1)
for plug_index, emeter in all_emeter.items(): for plug_index, emeter in all_emeters.items():
assert emeter == {} assert emeter == {}
emeter = dev.get_emeter_daily(index=plug_index) emeter = dev.sync.get_emeter_daily()
k, v = emeter.popitem() k, v = emeter.popitem()
assert isinstance(k, int) assert isinstance(k, int)
@ -590,18 +533,18 @@ def test_children_get_emeter_daily(dev):
# out of bounds # out of bounds
with pytest.raises(SmartStripException): with pytest.raises(SmartStripException):
dev.get_emeter_daily(year=1900, month=1, index=dev.num_children + 100) dev.sync.get_emeter_daily(year=1900, month=1, index=dev.num_children + 100)
@strip @strip
def test_children_get_emeter_monthly(dev): def test_children_get_emeter_monthly(dev):
assert dev.has_emeter assert dev.sync.get_has_emeter()
# test with index # test with index
for plug_index in range(dev.num_children): for plug_index in range(dev.num_children):
emeter = dev.get_emeter_monthly(year=1900, index=plug_index) emeter = dev.sync.get_emeter_monthly(year=1900, index=plug_index)
assert emeter == {} assert emeter == {}
emeter = dev.get_emeter_monthly() emeter = dev.sync.get_emeter_monthly(index=plug_index)
assert len(emeter) > 0 assert len(emeter) > 0
k, v = emeter.popitem() k, v = emeter.popitem()
@ -609,9 +552,11 @@ def test_children_get_emeter_monthly(dev):
assert isinstance(v, float) assert isinstance(v, float)
# test without index # test without index
all_emeter = dev.get_emeter_monthly(year=1900) all_emeters = dev.sync.get_emeter_monthly(year=1900)
for index, emeter in all_emeter.items(): for index, emeter in all_emeters.items():
assert emeter == {} assert emeter == {}
emeter = dev.sync.get_emeter_daily()
assert len(emeter) > 0 assert len(emeter) > 0
k, v = emeter.popitem() k, v = emeter.popitem()
@ -620,13 +565,13 @@ def test_children_get_emeter_monthly(dev):
# out of bounds # out of bounds
with pytest.raises(SmartStripException): with pytest.raises(SmartStripException):
dev.get_emeter_monthly(year=1900, index=dev.num_children + 100) dev.sync.get_emeter_monthly(year=1900, index=dev.num_children + 100)
def test_cache(dev): def test_cache(dev):
from datetime import timedelta from datetime import timedelta
dev.cache_ttl = timedelta(seconds=3) dev.sync.cache_ttl = timedelta(seconds=3)
with patch.object( with patch.object(
FakeTransportProtocol, "query", wraps=dev.protocol.query FakeTransportProtocol, "query", wraps=dev.protocol.query
) as query_mock: ) as query_mock:
@ -636,23 +581,23 @@ def test_cache(dev):
if dev.is_strip: if dev.is_strip:
CHECK_COUNT = 0 CHECK_COUNT = 0
dev.get_sysinfo() dev.sync.get_sys_info()
assert query_mock.call_count == CHECK_COUNT assert query_mock.call_count == CHECK_COUNT
dev.get_sysinfo() dev.sync.get_sys_info()
assert query_mock.call_count == CHECK_COUNT assert query_mock.call_count == CHECK_COUNT
def test_cache_invalidates(dev): def test_cache_invalidates(dev):
from datetime import timedelta from datetime import timedelta
dev.cache_ttl = timedelta(seconds=0) dev.sync.cache_ttl = timedelta(seconds=0)
with patch.object( with patch.object(
FakeTransportProtocol, "query", wraps=dev.protocol.query FakeTransportProtocol, "query", wraps=dev.protocol.query
) as query_mock: ) as query_mock:
dev.get_sysinfo() dev.sync.get_sys_info()
assert query_mock.call_count == 1 assert query_mock.call_count == 1
dev.get_sysinfo() dev.sync.get_sys_info()
assert query_mock.call_count == 2 assert query_mock.call_count == 2
# assert query_mock.called_once() # assert query_mock.called_once()