Simplify API documentation by using doctests (#73)

* Add doctests to SmartBulb

* Add SmartDevice doctests, cleanup README.md

* add doctests for smartplug and smartstrip

* add discover doctests

* Fix bulb mock

* add smartdimmer doctests

* add sphinx-generated docs, cleanup readme a bit

* remove sphinx-click as it does not work with asyncclick

* in preparation for rtd hooking, move doc deps to be separate from dev deps

* pytestmark needs to be applied separately for each and every file, this fixes the tests

* use pathlib for resolving relative paths

* Skip discovery doctest on python3.7

The code is just fine, but some reason the mocking behaves differently between 3.7 and 3.8.
The latter seems to accept a discrete object for asyncio.run where the former expects a coroutine..
This commit is contained in:
Teemu R
2020-06-30 02:29:52 +02:00
committed by GitHub
parent 99e0c4a418
commit f9a987ca18
34 changed files with 748 additions and 303 deletions

View File

@@ -3,6 +3,7 @@ import glob
import json
import os
from os.path import basename
from pathlib import Path, PurePath
from unittest.mock import MagicMock
import pytest # type: ignore # see https://github.com/pytest-dev/pytest/issues/3342
@@ -100,6 +101,38 @@ async def handle_turn_on(dev, turn_on):
pytestmark = pytest.mark.asyncio
def device_for_file(model):
for d in STRIPS:
if d in model:
return SmartStrip
for d in PLUGS:
if d in model:
return SmartPlug
for d in BULBS:
if d in model:
return SmartBulb
for d in DIMMERS:
if d in model:
return SmartDimmer
raise Exception("Unable to find type for %s", model)
def get_device_for_file(file):
# if the wanted file is not an absolute path, prepend the fixtures directory
p = Path(file)
if not p.is_absolute():
p = Path(__file__).parent / "fixtures" / file
with open(p) as f:
sysinfo = json.load(f)
model = basename(file)
p = device_for_file(model)(host="123.123.123.123")
p.protocol = FakeTransportProtocol(sysinfo)
asyncio.run(p.update())
return p
@pytest.fixture(params=SUPPORTED_DEVICES)
def dev(request):
"""Device fixture.
@@ -117,29 +150,7 @@ def dev(request):
return d
raise Exception("Unable to find type for %s" % ip)
def device_for_file(model):
for d in STRIPS:
if d in model:
return SmartStrip
for d in PLUGS:
if d in model:
return SmartPlug
for d in BULBS:
if d in model:
return SmartBulb
for d in DIMMERS:
if d in model:
return SmartDimmer
raise Exception("Unable to find type for %s", model)
with open(file) as f:
sysinfo = json.load(f)
model = basename(file)
p = device_for_file(model)(host="123.123.123.123")
p.protocol = FakeTransportProtocol(sysinfo)
asyncio.run(p.update())
yield p
return get_device_for_file(file)
def pytest_addoption(parser):

View File

@@ -27,7 +27,7 @@
"system": {
"get_sysinfo": {
"active_mode": "none",
"alias": "Nick office tplink",
"alias": "KL130 office bulb",
"ctrl_protocols": {
"name": "Linkie",
"version": "1.0"
@@ -45,7 +45,7 @@
"is_factory": false,
"is_variable_color_temp": 1,
"light_state": {
"brightness": 0,
"brightness": 30,
"color_temp": 0,
"hue": 15,
"mode": "normal",

View File

@@ -240,42 +240,41 @@ emeter_commands = {
}
def error(target, cmd="no-command", msg="default msg"):
return {target: {cmd: {"err_code": -1323, "msg": msg}}}
def error(msg="default msg"):
return {"err_code": -1323, "msg": msg}
def success(target, cmd, res):
def success(res):
if res:
res.update({"err_code": 0})
else:
res = {"err_code": 0}
return {target: {cmd: res}}
return res
class FakeTransportProtocol(TPLinkSmartHomeProtocol):
def __init__(self, info):
self.discovery_data = info
proto = FakeTransportProtocol.baseproto
for target in info:
# print("target %s" % target)
for cmd in info[target]:
# print("initializing tgt %s cmd %s" % (target, cmd))
proto[target][cmd] = info[target][cmd]
# if we have emeter support, check for it
# if we have emeter support, we need to add the missing pieces
for module in ["emeter", "smartlife.iot.common.emeter"]:
if module not in info:
# TODO required for old tests
continue
if "get_realtime" in info[module]:
get_realtime_res = info[module]["get_realtime"]
# TODO remove when removing old tests
if callable(get_realtime_res):
get_realtime_res = get_realtime_res()
if (
"err_code" not in get_realtime_res
or not get_realtime_res["err_code"]
):
proto[module] = emeter_commands[module]
for etype in ["get_realtime", "get_daystat", "get_monthstat"]:
if etype in info[module]: # if the fixture has the data, use it
# print("got %s %s from fixture: %s" % (module, etype, info[module][etype]))
proto[module][etype] = info[module][etype]
else: # otherwise fall back to the static one
dummy_data = emeter_commands[module][etype]
# print("got %s %s from dummy: %s" % (module, etype, dummy_data))
proto[module][etype] = dummy_data
# print("initialized: %s" % proto[module])
self.proto = proto
def set_alias(self, x, child_ids=[]):
@@ -309,7 +308,7 @@ class FakeTransportProtocol(TPLinkSmartHomeProtocol):
def set_mac(self, x, *args):
_LOGGER.debug("Setting mac to %s", x)
self.proto["system"]["get_sysinfo"]["mac"] = x
self.proto["system"]["get_sysinfo"]["mac"] = x["mac"]
def set_hs220_brightness(self, x, *args):
_LOGGER.debug("Setting brightness to %s", x)
@@ -345,9 +344,10 @@ class FakeTransportProtocol(TPLinkSmartHomeProtocol):
if not light_state["on_off"] and "on_off" not in x:
light_state = light_state["dft_on_state"]
_LOGGER.debug("Current state: %s", light_state)
_LOGGER.debug("Old state: %s", light_state)
for key in x:
light_state[key] = x[key]
_LOGGER.debug("New state: %s", light_state)
def light_state(self, x, *args):
light_state = self.proto["system"]["get_sysinfo"]["light_state"]
@@ -417,26 +417,39 @@ class FakeTransportProtocol(TPLinkSmartHomeProtocol):
except KeyError:
child_ids = []
target = next(iter(request))
if target not in proto.keys():
return error(target, msg="target not found")
def get_response_for_module(target):
cmd = next(iter(request[target]))
if cmd not in proto[target].keys():
return error(target, cmd, msg="command not found")
if target not in proto.keys():
return error(msg="target not found")
params = request[target][cmd]
_LOGGER.debug(f"Going to execute {target}.{cmd} (params: {params}).. ")
def get_response_for_command(cmd):
if cmd not in proto[target].keys():
return error(msg=f"command {cmd} not found")
if callable(proto[target][cmd]):
res = proto[target][cmd](self, params, child_ids)
_LOGGER.debug("[callable] %s.%s: %s", target, cmd, res)
# verify that change didn't break schema, requires refactoring..
# TestSmartPlug.sysinfo_schema(self.proto["system"]["get_sysinfo"])
return success(target, cmd, res)
elif isinstance(proto[target][cmd], dict):
res = proto[target][cmd]
_LOGGER.debug("[static] %s.%s: %s", target, cmd, res)
return success(target, cmd, res)
else:
raise NotImplementedError(f"target {target} cmd {cmd}")
params = request[target][cmd]
_LOGGER.debug(f"Going to execute {target}.{cmd} (params: {params}).. ")
if callable(proto[target][cmd]):
res = proto[target][cmd](self, params, child_ids)
_LOGGER.debug("[callable] %s.%s: %s", target, cmd, res)
return success(res)
elif isinstance(proto[target][cmd], dict):
res = proto[target][cmd]
_LOGGER.debug("[static] %s.%s: %s", target, cmd, res)
return success(res)
else:
raise NotImplementedError(f"target {target} cmd {cmd}")
from collections import defaultdict
cmd_responses = defaultdict(dict)
for cmd in request[target]:
cmd_responses[target][cmd] = get_response_for_command(cmd)
return cmd_responses
response = {}
for target in request:
response.update(get_response_for_module(target))
return response

View File

@@ -10,6 +10,7 @@ from .conftest import (
non_color_bulb,
non_dimmable,
non_variable_temp,
pytestmark,
turn_on,
variable_temp,
)

View File

@@ -1,12 +1,9 @@
import pytest
from asyncclick.testing import CliRunner
from kasa import SmartDevice
from kasa.cli import alias, brightness, emeter, raw_command, state, sysinfo
from .conftest import handle_turn_on, turn_on
pytestmark = pytest.mark.asyncio
from .conftest import handle_turn_on, pytestmark, turn_on
async def test_sysinfo(dev):

View File

@@ -2,7 +2,7 @@ import pytest
from kasa import SmartDimmer
from .conftest import dimmer, handle_turn_on, turn_on
from .conftest import dimmer, handle_turn_on, pytestmark, turn_on
@dimmer

View File

@@ -3,10 +3,7 @@ import pytest # type: ignore # https://github.com/pytest-dev/pytest/issues/3342
from kasa import DeviceType, Discover, SmartDevice, SmartDeviceException
from .conftest import bulb, dimmer, plug, strip
# to avoid adding this for each async function separately
pytestmark = pytest.mark.asyncio
from .conftest import bulb, dimmer, plug, pytestmark, strip
@plug

View File

@@ -2,7 +2,7 @@ import pytest
from kasa import SmartDeviceException
from .conftest import has_emeter, no_emeter
from .conftest import has_emeter, no_emeter, pytestmark
from .newfakes import CURRENT_CONSUMPTION_SCHEMA

View File

@@ -1,6 +1,6 @@
from kasa import DeviceType
from .conftest import plug
from .conftest import plug, pytestmark
from .newfakes import PLUG_SCHEMA

View File

@@ -4,6 +4,7 @@ import pytest
from ..exceptions import SmartDeviceException
from ..protocol import TPLinkSmartHomeProtocol
from .conftest import pytestmark
@pytest.mark.parametrize("retry_count", [1, 3, 5])

View File

@@ -0,0 +1,65 @@
import sys
import pytest
import xdoctest
from kasa.tests.conftest import get_device_for_file
def test_bulb_examples(mocker):
"""Use KL130 (bulb with all features) to test the doctests."""
p = get_device_for_file("KL130(US)_1.0.json")
mocker.patch("kasa.smartbulb.SmartBulb", return_value=p)
mocker.patch("kasa.smartbulb.SmartBulb.update")
res = xdoctest.doctest_module("kasa.smartbulb", "all")
assert not res["failed"]
def test_smartdevice_examples(mocker):
"""Use HS110 for emeter examples."""
p = get_device_for_file("HS110(EU)_1.0_real.json")
mocker.patch("kasa.smartdevice.SmartDevice", return_value=p)
mocker.patch("kasa.smartdevice.SmartDevice.update")
res = xdoctest.doctest_module("kasa.smartdevice", "all")
assert not res["failed"]
def test_plug_examples(mocker):
"""Test plug examples."""
p = get_device_for_file("HS110(EU)_1.0_real.json")
mocker.patch("kasa.smartplug.SmartPlug", return_value=p)
mocker.patch("kasa.smartplug.SmartPlug.update")
res = xdoctest.doctest_module("kasa.smartplug", "all")
assert not res["failed"]
def test_strip_examples(mocker):
"""Test strip examples."""
p = get_device_for_file("KP303(UK)_1.0.json")
mocker.patch("kasa.smartstrip.SmartStrip", return_value=p)
mocker.patch("kasa.smartstrip.SmartStrip.update")
res = xdoctest.doctest_module("kasa.smartstrip", "all")
assert not res["failed"]
def test_dimmer_examples(mocker):
"""Test dimmer examples."""
p = get_device_for_file("HS220(US)_1.0_real.json")
mocker.patch("kasa.smartdimmer.SmartDimmer", return_value=p)
mocker.patch("kasa.smartdimmer.SmartDimmer.update")
res = xdoctest.doctest_module("kasa.smartdimmer", "all")
assert not res["failed"]
@pytest.mark.skipif(
sys.version_info < (3, 8), reason="3.7 handles asyncio.run differently"
)
def test_discovery_examples(mocker):
"""Test discovery examples."""
p = get_device_for_file("KP303(UK)_1.0.json")
# This succeeds on python 3.8 but fails on 3.7
# ValueError: a coroutine was expected, got [<DeviceType.Strip model KP303(UK) ...
mocker.patch("kasa.discover.Discover.discover", return_value=[p])
res = xdoctest.doctest_module("kasa.discover", "all")
assert not res["failed"]

View File

@@ -5,7 +5,7 @@ import pytest # type: ignore # https://github.com/pytest-dev/pytest/issues/3342
from kasa import SmartDeviceException
from .conftest import handle_turn_on, turn_on
from .conftest import handle_turn_on, pytestmark, turn_on
from .newfakes import PLUG_SCHEMA, TZ_SCHEMA, FakeTransportProtocol

View File

@@ -4,7 +4,7 @@ import pytest
from kasa import SmartDeviceException, SmartStrip
from .conftest import handle_turn_on, strip, turn_on
from .conftest import handle_turn_on, pytestmark, strip, turn_on
@strip