Mirrored_Repos/python-kasa
1
0
Fork 0
mirror of https://github.com/python-kasa/python-kasa.git synced 2025-11-03 22:22:06 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

cli: print model, https, and lv for discover list (#1339)

Browse Source 
```
kasa --target 192.168.xx.xx discover list

HOST            MODEL     DEVICE FAMILY        ENCRYPT HTTPS LV  ALIAS
192.168.xxx.xxx KP115(EU) IOT.SMARTPLUGSWITCH  XOR     0     -   Fridge
192.168.xxx.xxx L900-5    SMART.TAPOBULB       KLAP    0     2   L900
192.168.xxx.xxx P115      SMART.TAPOPLUG       AES     0     2   Nightdesk
192.168.xxx.xxx TC65      SMART.IPCAMERA       AES     1     2   Tapo_TC65_B593

```

Also handles `TimeoutError` and `Exception` during `update()`

---------

Co-authored-by: Steven B <51370195+sdb9696@users.noreply.github.com>
This commit is contained in:
Teemu R.
2024-12-17 21:09:17 +01:00
committed by GitHub
parent 14d5629de1
commit 37ef7b0463
3 changed files with 64 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
kasa/cli/discover.py
View File

@@ -123,14 +123,19 @@ async def list(ctx):
async def print_discovered(dev: Device):
cparams = dev.config.connection_type
infostr = (
f"{dev.host:<15} {cparams.device_family.value:<20} "
f"{cparams.encryption_type.value:<7}"
f"{dev.host:<15} {dev.model:<9} {cparams.device_family.value:<20} "
f"{cparams.encryption_type.value:<7} {cparams.https:<5} "
f"{cparams.login_version or '-':<3}"
)
async with sem:
try:
await dev.update()
except AuthenticationError:
echo(f"{infostr} - Authentication failed")
except TimeoutError:
echo(f"{infostr} - Timed out")
except Exception as ex:
echo(f"{infostr} - Error: {ex}")
else:
echo(f"{infostr} {dev.alias}")
@@ -138,7 +143,10 @@ async def list(ctx):
if host := unsupported_exception.host:
echo(f"{host:<15} UNSUPPORTED DEVICE")
echo(f"{'HOST':<15} {'DEVICE FAMILY':<20} {'ENCRYPT':<7} {'ALIAS'}")
echo(
f"{'HOST':<15} {'MODEL':<9} {'DEVICE FAMILY':<20} {'ENCRYPT':<7} "
f"{'HTTPS':<5} {'LV':<3} {'ALIAS'}"
)
return await _discover(
ctx,
print_discovered=print_discovered,

16
tests/discovery_fixtures.py
View File

@@ -160,6 +160,17 @@ def create_discovery_mock(ip: str, fixture_data: dict):
login_version: int | None = None
port_override: int | None = None
@property
def model(self) -> str:
dd = self.discovery_data
model_region = (
dd["result"]["device_model"]
if self.discovery_port == 20002
else dd["system"]["get_sysinfo"]["model"]
)
model, _, _ = model_region.partition("(")
return model
@property
def _datagram(self) -> bytes:
if self.default_port == 9999:
@@ -178,7 +189,10 @@ def create_discovery_mock(ip: str, fixture_data: dict):
"encrypt_type", discovery_result.get("encrypt_info", {}).get("sym_schm")
)
login_version = discovery_result["mgt_encrypt_schm"].get("lv")
if not (login_version := discovery_result["mgt_encrypt_schm"].get("lv")) and (
et := discovery_result.get("encrypt_type")
):
login_version = max([int(i) for i in et])
https = discovery_result["mgt_encrypt_schm"]["is_support_https"]
dm = _DiscoveryMock(
ip,

47
tests/test_cli.py
View File

@@ -122,8 +122,15 @@ async def test_list_devices(discovery_mock, runner):
catch_exceptions=False,
)
assert res.exit_code == 0
header = f"{'HOST':<15} {'DEVICE FAMILY':<20} {'ENCRYPT':<7} {'ALIAS'}"
row = f"{discovery_mock.ip:<15} {discovery_mock.device_type:<20} {discovery_mock.encrypt_type:<7}"
header = (
f"{'HOST':<15} {'MODEL':<9} {'DEVICE FAMILY':<20} {'ENCRYPT':<7} "
f"{'HTTPS':<5} {'LV':<3} {'ALIAS'}"
)
row = (
f"{discovery_mock.ip:<15} {discovery_mock.model:<9} {discovery_mock.device_type:<20} "
f"{discovery_mock.encrypt_type:<7} {discovery_mock.https:<5} "
f"{discovery_mock.login_version or '-':<3}"
)
assert header in res.output
assert row in res.output
@@ -158,14 +165,26 @@ async def test_discover_raw(discovery_mock, runner, mocker):
redact_spy.assert_called()
@pytest.mark.parametrize(
("exception", "expected"),
[
pytest.param(
AuthenticationError("Failed to authenticate"),
"Authentication failed",
id="auth",
),
pytest.param(TimeoutError(), "Timed out", id="timeout"),
pytest.param(Exception("Foobar"), "Error: Foobar", id="other-error"),
],
)
@new_discovery
async def test_list_auth_failed(discovery_mock, mocker, runner):
async def test_list_update_failed(discovery_mock, mocker, runner, exception, expected):
"""Test that device update is called on main."""
device_class = Discover._get_device_class(discovery_mock.discovery_data)
mocker.patch.object(
device_class,
"update",
side_effect=AuthenticationError("Failed to authenticate"),
side_effect=exception,
)
res = await runner.invoke(
cli,
@@ -173,10 +192,17 @@ async def test_list_auth_failed(discovery_mock, mocker, runner):
catch_exceptions=False,
)
assert res.exit_code == 0
header = f"{'HOST':<15} {'DEVICE FAMILY':<20} {'ENCRYPT':<7} {'ALIAS'}"
row = f"{discovery_mock.ip:<15} {discovery_mock.device_type:<20} {discovery_mock.encrypt_type:<7} - Authentication failed"
assert header in res.output
assert row in res.output
header = (
f"{'HOST':<15} {'MODEL':<9} {'DEVICE FAMILY':<20} {'ENCRYPT':<7} "
f"{'HTTPS':<5} {'LV':<3} {'ALIAS'}"
)
row = (
f"{discovery_mock.ip:<15} {discovery_mock.model:<9} {discovery_mock.device_type:<20} "
f"{discovery_mock.encrypt_type:<7} {discovery_mock.https:<5} "
f"{discovery_mock.login_version or '-':<3} - {expected}"
)
assert header in res.output.replace("\n", "")
assert row in res.output.replace("\n", "")
async def test_list_unsupported(unsupported_device_info, runner):
@@ -187,7 +213,10 @@ async def test_list_unsupported(unsupported_device_info, runner):
catch_exceptions=False,
)
assert res.exit_code == 0
header = f"{'HOST':<15} {'DEVICE FAMILY':<20} {'ENCRYPT':<7} {'ALIAS'}"
header = (
f"{'HOST':<15} {'MODEL':<9} {'DEVICE FAMILY':<20} {'ENCRYPT':<7} "
f"{'HTTPS':<5} {'LV':<3} {'ALIAS'}"
)
row = f"{'127.0.0.1':<15} UNSUPPORTED DEVICE"
assert header in res.output
assert row in res.output