Skip to content

Migrate iot cloud module to mashumaro #1282

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

Merged
merged 3 commits into from
Nov 20, 2024
Merged

Migrate iot cloud module to mashumaro #1282

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
e6d0e96
Migrate iot cloud module to mashumaro
sdb9696 Nov 20, 2024
7ee01c3
Merge remote-tracking branch 'upstream/master' into janitor/mashu_cloud
sdb9696 Nov 20, 2024
87c1b16
Update cloudinfo names and use cld_connection for status
sdb9696 Nov 20, 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
29 changes: 17 additions & 12 deletions kasa/iot/modules/cloud.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,23 +1,28 @@
"""Cloud module implementation."""

from pydantic.v1 import BaseModel
from dataclasses import dataclass
from typing import Annotated

from mashumaro import DataClassDictMixin
from mashumaro.types import Alias

from ...feature import Feature
from ..iotmodule import IotModule


class CloudInfo(BaseModel):
@dataclass
class CloudInfo(DataClassDictMixin):
"""Container for cloud settings."""

binded: bool
cld_connection: int
fwDlPage: str
fwNotifyType: int
illegalType: int
provisioned: Annotated[int, Alias("binded")]
cloud_connected: Annotated[int, Alias("cld_connection")]
firmware_download_page: Annotated[str, Alias("fwDlPage")]
firmware_notify_type: Annotated[int, Alias("fwNotifyType")]
illegal_type: Annotated[int, Alias("illegalType")]
server: str
stopConnect: int
tcspInfo: str
tcspStatus: int
stop_connect: Annotated[int, Alias("stopConnect")]
tcsp_info: Annotated[str, Alias("tcspInfo")]
tcsp_status: Annotated[int, Alias("tcspStatus")]
username: str


Expand All @@ -42,7 +47,7 @@ def _initialize_features(self) -> None:
@property
def is_connected(self) -> bool:
"""Return true if device is connected to the cloud."""
return self.info.binded
return bool(self.info.cloud_connected)

def query(self) -> dict:
"""Request cloud connectivity info."""
Expand All @@ -51,7 +56,7 @@ def query(self) -> dict:
@property
def info(self) -> CloudInfo:
"""Return information about the cloud connectivity."""
return CloudInfo.parse_obj(self.data["get_info"])
return CloudInfo.from_dict(self.data["get_info"])

def get_available_firmwares(self) -> dict:
"""Return list of available firmwares."""
Expand Down
1 change: 1 addition & 0 deletions tests/fakeprotocol_iot.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -125,6 +125,7 @@ def success(res):
"username": "",
"server": "devs.tplinkcloud.com",
"binded": 0,
"err_code": 0,
"cld_connection": 0,
"illegalType": -1,
"stopConnect": -1,
Expand Down
13 changes: 13 additions & 0 deletions tests/iot/modules/test_cloud.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
from kasa import Device, Module

from ...device_fixtures import device_iot


@device_iot
def test_cloud(dev: Device):
cloud = dev.modules.get(Module.IotCloud)
assert cloud
info = cloud.info
assert info
assert isinstance(info.provisioned, int)
assert cloud.is_connected == bool(info.cloud_connected)
Loading