mirror of
https://github.com/python-kasa/python-kasa.git
synced 2025-01-22 20:57:07 +00:00
Add tests for dump_devinfo parent/child smartcam fixture generation (#1428)
Some checks failed
CI / Perform linting checks (3.13) (push) Has been cancelled
CodeQL checks / Analyze (python) (push) Has been cancelled
CI / Python ${{ matrix.python-version}} on ${{ matrix.os }}${{ fromJSON('[" (extras)", ""]')[matrix.extras == ''] }} (false, macos-latest, 3.11) (push) Has been cancelled
CI / Python ${{ matrix.python-version}} on ${{ matrix.os }}${{ fromJSON('[" (extras)", ""]')[matrix.extras == ''] }} (false, macos-latest, 3.12) (push) Has been cancelled
CI / Python ${{ matrix.python-version}} on ${{ matrix.os }}${{ fromJSON('[" (extras)", ""]')[matrix.extras == ''] }} (false, macos-latest, 3.13) (push) Has been cancelled
CI / Python ${{ matrix.python-version}} on ${{ matrix.os }}${{ fromJSON('[" (extras)", ""]')[matrix.extras == ''] }} (false, ubuntu-latest, 3.11) (push) Has been cancelled
CI / Python ${{ matrix.python-version}} on ${{ matrix.os }}${{ fromJSON('[" (extras)", ""]')[matrix.extras == ''] }} (false, ubuntu-latest, 3.12) (push) Has been cancelled
CI / Python ${{ matrix.python-version}} on ${{ matrix.os }}${{ fromJSON('[" (extras)", ""]')[matrix.extras == ''] }} (false, ubuntu-latest, 3.13) (push) Has been cancelled
CI / Python ${{ matrix.python-version}} on ${{ matrix.os }}${{ fromJSON('[" (extras)", ""]')[matrix.extras == ''] }} (false, windows-latest, 3.11) (push) Has been cancelled
CI / Python ${{ matrix.python-version}} on ${{ matrix.os }}${{ fromJSON('[" (extras)", ""]')[matrix.extras == ''] }} (false, windows-latest, 3.12) (push) Has been cancelled
CI / Python ${{ matrix.python-version}} on ${{ matrix.os }}${{ fromJSON('[" (extras)", ""]')[matrix.extras == ''] }} (false, windows-latest, 3.13) (push) Has been cancelled
CI / Python ${{ matrix.python-version}} on ${{ matrix.os }}${{ fromJSON('[" (extras)", ""]')[matrix.extras == ''] }} (true, ubuntu-latest, 3.11) (push) Has been cancelled
CI / Python ${{ matrix.python-version}} on ${{ matrix.os }}${{ fromJSON('[" (extras)", ""]')[matrix.extras == ''] }} (true, ubuntu-latest, 3.12) (push) Has been cancelled
CI / Python ${{ matrix.python-version}} on ${{ matrix.os }}${{ fromJSON('[" (extras)", ""]')[matrix.extras == ''] }} (true, ubuntu-latest, 3.13) (push) Has been cancelled
Some checks failed
CI / Perform linting checks (3.13) (push) Has been cancelled
CodeQL checks / Analyze (python) (push) Has been cancelled
CI / Python ${{ matrix.python-version}} on ${{ matrix.os }}${{ fromJSON('[" (extras)", ""]')[matrix.extras == ''] }} (false, macos-latest, 3.11) (push) Has been cancelled
CI / Python ${{ matrix.python-version}} on ${{ matrix.os }}${{ fromJSON('[" (extras)", ""]')[matrix.extras == ''] }} (false, macos-latest, 3.12) (push) Has been cancelled
CI / Python ${{ matrix.python-version}} on ${{ matrix.os }}${{ fromJSON('[" (extras)", ""]')[matrix.extras == ''] }} (false, macos-latest, 3.13) (push) Has been cancelled
CI / Python ${{ matrix.python-version}} on ${{ matrix.os }}${{ fromJSON('[" (extras)", ""]')[matrix.extras == ''] }} (false, ubuntu-latest, 3.11) (push) Has been cancelled
CI / Python ${{ matrix.python-version}} on ${{ matrix.os }}${{ fromJSON('[" (extras)", ""]')[matrix.extras == ''] }} (false, ubuntu-latest, 3.12) (push) Has been cancelled
CI / Python ${{ matrix.python-version}} on ${{ matrix.os }}${{ fromJSON('[" (extras)", ""]')[matrix.extras == ''] }} (false, ubuntu-latest, 3.13) (push) Has been cancelled
CI / Python ${{ matrix.python-version}} on ${{ matrix.os }}${{ fromJSON('[" (extras)", ""]')[matrix.extras == ''] }} (false, windows-latest, 3.11) (push) Has been cancelled
CI / Python ${{ matrix.python-version}} on ${{ matrix.os }}${{ fromJSON('[" (extras)", ""]')[matrix.extras == ''] }} (false, windows-latest, 3.12) (push) Has been cancelled
CI / Python ${{ matrix.python-version}} on ${{ matrix.os }}${{ fromJSON('[" (extras)", ""]')[matrix.extras == ''] }} (false, windows-latest, 3.13) (push) Has been cancelled
CI / Python ${{ matrix.python-version}} on ${{ matrix.os }}${{ fromJSON('[" (extras)", ""]')[matrix.extras == ''] }} (true, ubuntu-latest, 3.11) (push) Has been cancelled
CI / Python ${{ matrix.python-version}} on ${{ matrix.os }}${{ fromJSON('[" (extras)", ""]')[matrix.extras == ''] }} (true, ubuntu-latest, 3.12) (push) Has been cancelled
CI / Python ${{ matrix.python-version}} on ${{ matrix.os }}${{ fromJSON('[" (extras)", ""]')[matrix.extras == ''] }} (true, ubuntu-latest, 3.13) (push) Has been cancelled
Currently the dump_devinfo fixture generation tests do not test generation for hub and their children. This PR enables tests for `smartcam` hubs and their child fixtures. It does not enable support for `smart` hub fixtures as not all the fixtures currently have the required info. This can be addressed in a subsequent PR.
This commit is contained in:
parent
debcff9f9b
commit
2e3b1bc376
@ -48,13 +48,18 @@ class FakeSmartTransport(BaseTransport):
|
|||||||
),
|
),
|
||||||
)
|
)
|
||||||
self.fixture_name = fixture_name
|
self.fixture_name = fixture_name
|
||||||
|
|
||||||
|
# When True verbatim will bypass any extra processing of missing
|
||||||
|
# methods and is used to test the fixture creation itself.
|
||||||
|
self.verbatim = verbatim
|
||||||
|
|
||||||
# Don't copy the dict if the device is a child so that updates on the
|
# Don't copy the dict if the device is a child so that updates on the
|
||||||
# child are then still reflected on the parent's lis of child device in
|
# child are then still reflected on the parent's lis of child device in
|
||||||
if not is_child:
|
if not is_child:
|
||||||
self.info = copy.deepcopy(info)
|
self.info = copy.deepcopy(info)
|
||||||
if get_child_fixtures:
|
if get_child_fixtures:
|
||||||
self.child_protocols = self._get_child_protocols(
|
self.child_protocols = self._get_child_protocols(
|
||||||
self.info, self.fixture_name, "get_child_device_list"
|
self.info, self.fixture_name, "get_child_device_list", self.verbatim
|
||||||
)
|
)
|
||||||
else:
|
else:
|
||||||
self.info = info
|
self.info = info
|
||||||
@ -67,9 +72,6 @@ class FakeSmartTransport(BaseTransport):
|
|||||||
self.warn_fixture_missing_methods = warn_fixture_missing_methods
|
self.warn_fixture_missing_methods = warn_fixture_missing_methods
|
||||||
self.fix_incomplete_fixture_lists = fix_incomplete_fixture_lists
|
self.fix_incomplete_fixture_lists = fix_incomplete_fixture_lists
|
||||||
|
|
||||||
# When True verbatim will bypass any extra processing of missing
|
|
||||||
# methods and is used to test the fixture creation itself.
|
|
||||||
self.verbatim = verbatim
|
|
||||||
if verbatim:
|
if verbatim:
|
||||||
self.warn_fixture_missing_methods = False
|
self.warn_fixture_missing_methods = False
|
||||||
self.fix_incomplete_fixture_lists = False
|
self.fix_incomplete_fixture_lists = False
|
||||||
@ -124,7 +126,7 @@ class FakeSmartTransport(BaseTransport):
|
|||||||
},
|
},
|
||||||
),
|
),
|
||||||
"get_auto_update_info": (
|
"get_auto_update_info": (
|
||||||
"firmware",
|
("firmware", 2),
|
||||||
{"enable": True, "random_range": 120, "time": 180},
|
{"enable": True, "random_range": 120, "time": 180},
|
||||||
),
|
),
|
||||||
"get_alarm_configure": (
|
"get_alarm_configure": (
|
||||||
@ -169,6 +171,30 @@ class FakeSmartTransport(BaseTransport):
|
|||||||
),
|
),
|
||||||
}
|
}
|
||||||
|
|
||||||
|
def _missing_result(self, method):
|
||||||
|
"""Check the FIXTURE_MISSING_MAP for responses.
|
||||||
|
|
||||||
|
Fixtures generated prior to a query being supported by dump_devinfo
|
||||||
|
do not have the response so this method checks whether the component
|
||||||
|
is supported and fills in the missing response.
|
||||||
|
If the first value of the lookup value is a tuple it will also check
|
||||||
|
the version, i.e. (component_name, component_version).
|
||||||
|
"""
|
||||||
|
if not (missing := self.FIXTURE_MISSING_MAP.get(method)):
|
||||||
|
return None
|
||||||
|
condition = missing[0]
|
||||||
|
if (
|
||||||
|
isinstance(condition, tuple)
|
||||||
|
and (version := self.components.get(condition[0]))
|
||||||
|
and version >= condition[1]
|
||||||
|
):
|
||||||
|
return copy.deepcopy(missing[1])
|
||||||
|
|
||||||
|
if condition in self.components:
|
||||||
|
return copy.deepcopy(missing[1])
|
||||||
|
|
||||||
|
return None
|
||||||
|
|
||||||
async def send(self, request: str):
|
async def send(self, request: str):
|
||||||
request_dict = json_loads(request)
|
request_dict = json_loads(request)
|
||||||
method = request_dict["method"]
|
method = request_dict["method"]
|
||||||
@ -189,7 +215,7 @@ class FakeSmartTransport(BaseTransport):
|
|||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def _get_child_protocols(
|
def _get_child_protocols(
|
||||||
parent_fixture_info, parent_fixture_name, child_devices_key
|
parent_fixture_info, parent_fixture_name, child_devices_key, verbatim
|
||||||
):
|
):
|
||||||
child_infos = parent_fixture_info.get(child_devices_key, {}).get(
|
child_infos = parent_fixture_info.get(child_devices_key, {}).get(
|
||||||
"child_device_list", []
|
"child_device_list", []
|
||||||
@ -251,7 +277,7 @@ class FakeSmartTransport(BaseTransport):
|
|||||||
)
|
)
|
||||||
# Replace parent child infos with the infos from the child fixtures so
|
# Replace parent child infos with the infos from the child fixtures so
|
||||||
# that updates update both
|
# that updates update both
|
||||||
if child_infos and found_child_fixture_infos:
|
if not verbatim and child_infos and found_child_fixture_infos:
|
||||||
parent_fixture_info[child_devices_key]["child_device_list"] = (
|
parent_fixture_info[child_devices_key]["child_device_list"] = (
|
||||||
found_child_fixture_infos
|
found_child_fixture_infos
|
||||||
)
|
)
|
||||||
@ -318,13 +344,11 @@ class FakeSmartTransport(BaseTransport):
|
|||||||
elif child_method in child_device_calls:
|
elif child_method in child_device_calls:
|
||||||
result = copy.deepcopy(child_device_calls[child_method])
|
result = copy.deepcopy(child_device_calls[child_method])
|
||||||
return {"result": result, "error_code": 0}
|
return {"result": result, "error_code": 0}
|
||||||
elif (
|
elif missing_result := self._missing_result(child_method):
|
||||||
# FIXTURE_MISSING is for service calls not in place when
|
# FIXTURE_MISSING is for service calls not in place when
|
||||||
# SMART fixtures started to be generated
|
# SMART fixtures started to be generated
|
||||||
missing_result := self.FIXTURE_MISSING_MAP.get(child_method)
|
|
||||||
) and missing_result[0] in self.components:
|
|
||||||
# Copy to info so it will work with update methods
|
# Copy to info so it will work with update methods
|
||||||
child_device_calls[child_method] = copy.deepcopy(missing_result[1])
|
child_device_calls[child_method] = missing_result
|
||||||
result = copy.deepcopy(info[child_method])
|
result = copy.deepcopy(info[child_method])
|
||||||
retval = {"result": result, "error_code": 0}
|
retval = {"result": result, "error_code": 0}
|
||||||
return retval
|
return retval
|
||||||
@ -529,13 +553,11 @@ class FakeSmartTransport(BaseTransport):
|
|||||||
"method": method,
|
"method": method,
|
||||||
}
|
}
|
||||||
|
|
||||||
if (
|
if missing_result := self._missing_result(method):
|
||||||
# FIXTURE_MISSING is for service calls not in place when
|
# FIXTURE_MISSING is for service calls not in place when
|
||||||
# SMART fixtures started to be generated
|
# SMART fixtures started to be generated
|
||||||
missing_result := self.FIXTURE_MISSING_MAP.get(method)
|
|
||||||
) and missing_result[0] in self.components:
|
|
||||||
# Copy to info so it will work with update methods
|
# Copy to info so it will work with update methods
|
||||||
info[method] = copy.deepcopy(missing_result[1])
|
info[method] = missing_result
|
||||||
result = copy.deepcopy(info[method])
|
result = copy.deepcopy(info[method])
|
||||||
retval = {"result": result, "error_code": 0}
|
retval = {"result": result, "error_code": 0}
|
||||||
elif (
|
elif (
|
||||||
|
@ -57,11 +57,11 @@ class FakeSmartCamTransport(BaseTransport):
|
|||||||
# lists
|
# lists
|
||||||
if get_child_fixtures:
|
if get_child_fixtures:
|
||||||
self.child_protocols = FakeSmartTransport._get_child_protocols(
|
self.child_protocols = FakeSmartTransport._get_child_protocols(
|
||||||
self.info, self.fixture_name, "getChildDeviceList"
|
self.info, self.fixture_name, "getChildDeviceList", self.verbatim
|
||||||
)
|
)
|
||||||
else:
|
else:
|
||||||
self.info = info
|
self.info = info
|
||||||
# self.child_protocols = self._get_child_protocols()
|
|
||||||
self.list_return_size = list_return_size
|
self.list_return_size = list_return_size
|
||||||
|
|
||||||
# Setting this flag allows tests to create dummy transports without
|
# Setting this flag allows tests to create dummy transports without
|
||||||
|
@ -77,7 +77,7 @@ def idgenerator(paramtuple: FixtureInfo):
|
|||||||
return None
|
return None
|
||||||
|
|
||||||
|
|
||||||
def get_fixture_info() -> list[FixtureInfo]:
|
def get_fixture_infos() -> list[FixtureInfo]:
|
||||||
"""Return raw discovery file contents as JSON. Used for discovery tests."""
|
"""Return raw discovery file contents as JSON. Used for discovery tests."""
|
||||||
fixture_data = []
|
fixture_data = []
|
||||||
for file, protocol in SUPPORTED_DEVICES:
|
for file, protocol in SUPPORTED_DEVICES:
|
||||||
@ -99,7 +99,7 @@ def get_fixture_info() -> list[FixtureInfo]:
|
|||||||
return fixture_data
|
return fixture_data
|
||||||
|
|
||||||
|
|
||||||
FIXTURE_DATA: list[FixtureInfo] = get_fixture_info()
|
FIXTURE_DATA: list[FixtureInfo] = get_fixture_infos()
|
||||||
|
|
||||||
|
|
||||||
def filter_fixtures(
|
def filter_fixtures(
|
||||||
|
@ -1,5 +1,7 @@
|
|||||||
"""Module for dump_devinfo tests."""
|
"""Module for dump_devinfo tests."""
|
||||||
|
|
||||||
|
import copy
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
from devtools.dump_devinfo import get_legacy_fixture, get_smart_fixtures
|
from devtools.dump_devinfo import get_legacy_fixture, get_smart_fixtures
|
||||||
@ -11,6 +13,7 @@ from kasa.smartcam import SmartCamDevice
|
|||||||
from .conftest import (
|
from .conftest import (
|
||||||
FixtureInfo,
|
FixtureInfo,
|
||||||
get_device_for_fixture,
|
get_device_for_fixture,
|
||||||
|
get_fixture_info,
|
||||||
parametrize,
|
parametrize,
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -64,22 +67,54 @@ async def test_smart_fixtures(fixture_info: FixtureInfo):
|
|||||||
assert fixture_info.data == fixture_result.data
|
assert fixture_info.data == fixture_result.data
|
||||||
|
|
||||||
|
|
||||||
|
def _normalize_child_device_ids(info: dict):
|
||||||
|
"""Scrubbed child device ids in hubs may not match ids in child fixtures.
|
||||||
|
|
||||||
|
Different hub fixtures could create the same child fixture so we scrub
|
||||||
|
them again for the purpose of the test.
|
||||||
|
"""
|
||||||
|
if dev_info := info.get("get_device_info"):
|
||||||
|
dev_info["device_id"] = "SCRUBBED"
|
||||||
|
elif (
|
||||||
|
dev_info := info.get("getDeviceInfo", {})
|
||||||
|
.get("device_info", {})
|
||||||
|
.get("basic_info")
|
||||||
|
):
|
||||||
|
dev_info["dev_id"] = "SCRUBBED"
|
||||||
|
|
||||||
|
|
||||||
@smartcam_fixtures
|
@smartcam_fixtures
|
||||||
async def test_smartcam_fixtures(fixture_info: FixtureInfo):
|
async def test_smartcam_fixtures(fixture_info: FixtureInfo):
|
||||||
"""Test that smartcam fixtures are created the same."""
|
"""Test that smartcam fixtures are created the same."""
|
||||||
dev = await get_device_for_fixture(fixture_info, verbatim=True)
|
dev = await get_device_for_fixture(fixture_info, verbatim=True)
|
||||||
assert isinstance(dev, SmartCamDevice)
|
assert isinstance(dev, SmartCamDevice)
|
||||||
if dev.children:
|
|
||||||
pytest.skip("Test not currently implemented for devices with children.")
|
created_fixtures = await get_smart_fixtures(
|
||||||
fixtures = await get_smart_fixtures(
|
|
||||||
dev.protocol,
|
dev.protocol,
|
||||||
discovery_info=fixture_info.data.get("discovery_result"),
|
discovery_info=fixture_info.data.get("discovery_result"),
|
||||||
batch_size=5,
|
batch_size=5,
|
||||||
)
|
)
|
||||||
fixture_result = fixtures[0]
|
fixture_result = created_fixtures.pop(0)
|
||||||
|
|
||||||
assert fixture_info.data == fixture_result.data
|
assert fixture_info.data == fixture_result.data
|
||||||
|
|
||||||
|
for created_child_fixture in created_fixtures:
|
||||||
|
child_fixture_info = get_fixture_info(
|
||||||
|
created_child_fixture.filename + ".json",
|
||||||
|
created_child_fixture.protocol_suffix,
|
||||||
|
)
|
||||||
|
|
||||||
|
assert child_fixture_info
|
||||||
|
|
||||||
|
_normalize_child_device_ids(created_child_fixture.data)
|
||||||
|
|
||||||
|
saved_fixture_data = copy.deepcopy(child_fixture_info.data)
|
||||||
|
_normalize_child_device_ids(saved_fixture_data)
|
||||||
|
saved_fixture_data = {
|
||||||
|
key: val for key, val in saved_fixture_data.items() if val != -1001
|
||||||
|
}
|
||||||
|
assert saved_fixture_data == created_child_fixture.data
|
||||||
|
|
||||||
|
|
||||||
@iot_fixtures
|
@iot_fixtures
|
||||||
async def test_iot_fixtures(fixture_info: FixtureInfo):
|
async def test_iot_fixtures(fixture_info: FixtureInfo):
|
||||||
|
Loading…
Reference in New Issue
Block a user