Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

chore(cli): download without device_type and discovery add get_sn #343

Merged
merged 6 commits into from
Dec 24, 2024
Merged

chore(cli): download without device_type and discovery add get_sn #343

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
00fdcbb
chore(cli): download without device_type and discovery add get_sn
wuwentao Dec 23, 2024
c41274a
Merge remote-tracking branch 'origin/main' into improve_cli
wuwentao Dec 23, 2024
48c3a4a
chore(cli): download without device_type and discovery add get_sn
wuwentao Dec 23, 2024
602be4a
chore(cli): fix cli testcase error
wuwentao Dec 23, 2024
4269038
chore(cli): addd cli test case
wuwentao Dec 23, 2024
7d7c829
chore(cli): add cli testcase
wuwentao Dec 23, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 37 additions & 6 deletions midealocal/cli.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@
import json
import logging
import sys
from argparse import ArgumentParser, Namespace
from argparse import ArgumentParser, BooleanOptionalAction, Namespace
from pathlib import Path
from typing import Any, NoReturn

Expand All @@ -28,7 +28,7 @@
NoSupportedProtocol,
)
from midealocal.devices import device_selector
from midealocal.discover import discover
from midealocal.discover import SERIAL_TYPE1_LENGTH, discover
from midealocal.exceptions import SocketException
from midealocal.version import __version__

Expand Down Expand Up @@ -86,15 +86,20 @@ async def _get_keys(self, device_id: int) -> dict[int, dict[str, Any]]:

async def discover(self) -> list[MideaDevice]:
"""Discover device information."""
device_list: list[MideaDevice] = []

devices = discover(ip_address=self.namespace.host)

device_list: list[MideaDevice] = []
if len(devices) == 0:
_LOGGER.error("No devices found.")
return device_list

# Dump only basic device info from the base class
_LOGGER.info("Found %d devices.", len(devices))
# get sn
if self.namespace.get_sn:
_LOGGER.info("Found devices: %s", devices)
return device_list
for device in devices.values():
keys = (
{0: {"token": "", "key": ""}}
Expand Down Expand Up @@ -171,10 +176,11 @@ def save(self) -> None:

async def download(self) -> None:
"""Download lua from cloud."""
device_type = int.from_bytes(self.namespace.device_type or bytearray())
device_sn = str(self.namespace.device_sn)
# model and device_type will be get from SN or host
model: str | None = None
device_type: int = 0

# download with host ip
if self.namespace.host:
devices = discover(ip_address=self.namespace.host)

Expand All @@ -186,14 +192,33 @@ async def download(self) -> None:
device_type = device["type"]
device_sn = device["sn"]
model = device["model"]
# download with SN
elif self.namespace.device_sn:
device_sn = str(self.namespace.device_sn)
# manual input device_type exist
if self.namespace.device_type:
device_type = int.from_bytes(self.namespace.device_type or bytearray())
# no device type input, parse device_type from SN
elif len(device_sn) == SERIAL_TYPE1_LENGTH:
device_type = int.from_bytes(bytes.fromhex(device_sn[4:6]))
# parse model from SN
model = str(device_sn[9:17])
else:
_LOGGER.error("host or sn is mandatory")
return

cloud = await self._get_cloud()
_LOGGER.debug("Try to authenticate to the cloud.")
if not await cloud.login():
_LOGGER.error("Failed to authenticate to the cloud.")
return

_LOGGER.debug("Download lua file for %s [%s]", device_sn, hex(device_type))
_LOGGER.debug(
"Download lua file for %s [%s] %s",
device_sn,
hex(device_type),
model,
)
lua = await cloud.download_lua(str(Path()), device_type, device_sn, model)
_LOGGER.info("Downloaded lua file: %s", lua)

Expand Down Expand Up @@ -332,6 +357,12 @@ def main() -> NoReturn:
help="Hostname or IP address of a single device to discover.",
default=None,
)
discover_parser.add_argument(
"--get_sn",
help="Get device SN with host ip.",
default=False,
action=BooleanOptionalAction,
)
discover_parser.set_defaults(func=cli.discover)

decode_msg_parser = subparsers.add_parser(
Expand Down
94 changes: 88 additions & 6 deletions tests/cli_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@ def setUp(self) -> None:
device_sn="",
user=False,
debug=True,
get_sn=False,
attribute="power",
value="0",
attr_type="bool",
Expand Down Expand Up @@ -108,8 +109,8 @@ async def test_discover(self) -> None:
"type": "AC",
"ip_address": "192.168.0.2",
"port": 6444,
"model": "AC123",
"sn": "AC123",
"model": "AC123000",
"sn": "0000AC12300000000000000000000000",
}
mock_cloud_instance = AsyncMock()
mock_device_instance = MagicMock()
Expand Down Expand Up @@ -146,6 +147,14 @@ async def test_discover(self) -> None:
99: {"token": "token", "key": "key"},
}

# test V3 device get_sn
self.namespace.get_sn = True
await self.cli.discover() # test V3 device get_sn
mock_discover.assert_called()
# set get_sn to default False after test done
self.namespace.get_sn = False

# test V3 device
await self.cli.discover() # V3 device
authenticate_mock.assert_called()
refresh_status_mock.assert_called_with(True)
Expand Down Expand Up @@ -226,14 +235,18 @@ async def test_download(self) -> None:
"type": 0xAC,
"ip_address": "192.168.0.2",
"port": 6444,
"model": "AC123",
"sn": "AC123",
"model": "ABCD1234",
"sn": "0000AC000ABCD1234000000000000000",
}
mock_cloud_instance = AsyncMock()
with (
patch(
"midealocal.cli.discover",
side_effect=[{}, {1: mock_device}, {1: mock_device}],
side_effect=[
{}, # test no device
{1: mock_device}, # test cloud login failed
{1: mock_device}, # test download lua with host ip
],
) as mock_discover,
patch.object(
self.cli,
Expand All @@ -242,25 +255,94 @@ async def test_download(self) -> None:
),
):
await self.cli.download() # No device found
# default is discover with host ip, test result is None
mock_discover.assert_called_once_with(ip_address=self.namespace.host)
mock_discover.reset_mock()

mock_cloud_instance.login.side_effect = [False, True]
mock_cloud_instance.login.side_effect = [
False, # test cloud login failed
True, # test download lua with host ip
True, # test download lua with SN
True, # test download lua with SN
]
await self.cli.download() # Cloud login failed
# default is discover with host ip
mock_discover.assert_called_once_with(ip_address=self.namespace.host)
mock_discover.reset_mock()
# test cloud login failed
mock_cloud_instance.login.assert_called_once()
mock_cloud_instance.login.reset_mock()

# download lua with host (default is host)
await self.cli.download()
# default is discover with host ip
mock_discover.assert_called_once_with(ip_address=self.namespace.host)
mock_discover.reset_mock()
# cloud login pass
mock_cloud_instance.login.assert_called_once()
mock_cloud_instance.login.reset_mock()
mock_cloud_instance.download_lua.assert_called_once_with(
str(Path()),
mock_device["type"],
mock_device["sn"],
mock_device["model"],
)
mock_cloud_instance.download_lua.reset_mock()
mock_cloud_instance.download_plugin.assert_called_once_with(
str(Path()),
mock_device["type"],
mock_device["sn"],
)
mock_cloud_instance.download_plugin.reset_mock()

# download lua with SN (set host to None and match elif)
self.namespace.host = None
self.namespace.device_sn = mock_device["sn"]
await self.cli.download()
# skip discover and cloud login pass
mock_cloud_instance.login.assert_called_once()
mock_cloud_instance.login.reset_mock()
mock_cloud_instance.download_lua.assert_called_once_with(
str(Path()),
mock_device["type"],
mock_device["sn"],
mock_device["model"],
)
mock_cloud_instance.download_lua.reset_mock()
mock_cloud_instance.download_plugin.assert_called_once_with(
str(Path()),
mock_device["type"],
mock_device["sn"],
)
mock_cloud_instance.download_plugin.reset_mock()

# download lua with SN and device_type
self.namespace.host = None
self.namespace.device_sn = mock_device["sn"]
self.namespace.device_type = bytes.fromhex("AC")
await self.cli.download()
# skip discover and cloud login pass
mock_cloud_instance.login.assert_called_once()
mock_cloud_instance.login.reset_mock()
mock_cloud_instance.download_lua.assert_called_once_with(
str(Path()),
mock_device["type"],
mock_device["sn"],
mock_device["model"],
)
mock_cloud_instance.download_lua.reset_mock()
mock_cloud_instance.download_plugin.assert_called_once_with(
str(Path()),
mock_device["type"],
mock_device["sn"],
)
mock_cloud_instance.download_plugin.reset_mock()

# test no host and no device_sn
self.namespace.host = None
self.namespace.device_sn = None
# result is None
await self.cli.download()

async def test_set_attribute(self) -> None:
"""Test set attribute."""
Expand Down
Loading