Mirrored_Repos/python-kasa
1
0
Fork 0
mirror of https://github.com/python-kasa/python-kasa.git synced 2025-08-09 20:24:02 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

Merge branch 'master' into janitor/smartcam_referer

Browse Source
This commit is contained in:
Steven B.
2024-12-23 09:27:56 +00:00
committed by GitHub
parent 69ea4ec728 63f4f82791
commit 322dda599b
17 changed files with 5584 additions and 296 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
tests/device_fixtures.py
View File

@@ -79,8 +79,6 @@ PLUGS_IOT = {
"KP125",
"KP401",
}
# P135 supports dimming, but its not currently support
# by the library
PLUGS_SMART = {
"P100",
"P110",
@@ -112,7 +110,7 @@ SWITCHES_SMART = {
}
SWITCHES = {*SWITCHES_IOT, *SWITCHES_SMART}
STRIPS_IOT = {"HS107", "HS300", "KP303", "KP200", "KP400", "EP40"}
STRIPS_SMART = {"P300", "P304M", "TP25", "EP40M"}
STRIPS_SMART = {"P300", "P304M", "TP25", "EP40M", "P210M", "P306"}
STRIPS = {*STRIPS_IOT, *STRIPS_SMART}
DIMMERS_IOT = {"ES20M", "HS220", "KS220", "KS220M", "KS230", "KP405"}

419
tests/fixtures/smart/P135(US)_1.0_1.2.0.json vendored Normal file
View File

@@ -0,0 +1,419 @@
{
"component_nego": {
"component_list": [
{
"id": "device",
"ver_code": 2
},
{
"id": "firmware",
"ver_code": 2
},
{
"id": "quick_setup",
"ver_code": 3
},
{
"id": "inherit",
"ver_code": 1
},
{
"id": "time",
"ver_code": 1
},
{
"id": "wireless",
"ver_code": 1
},
{
"id": "schedule",
"ver_code": 2
},
{
"id": "countdown",
"ver_code": 2
},
{
"id": "antitheft",
"ver_code": 1
},
{
"id": "account",
"ver_code": 1
},
{
"id": "synchronize",
"ver_code": 1
},
{
"id": "sunrise_sunset",
"ver_code": 1
},
{
"id": "led",
"ver_code": 1
},
{
"id": "cloud_connect",
"ver_code": 1
},
{
"id": "iot_cloud",
"ver_code": 1
},
{
"id": "device_local_time",
"ver_code": 1
},
{
"id": "default_states",
"ver_code": 1
},
{
"id": "brightness",
"ver_code": 1
},
{
"id": "preset",
"ver_code": 1
},
{
"id": "on_off_gradually",
"ver_code": 2
},
{
"id": "dimmer_calibration",
"ver_code": 1
},
{
"id": "localSmart",
"ver_code": 1
},
{
"id": "overheat_protection",
"ver_code": 1
},
{
"id": "matter",
"ver_code": 2
}
]
},
"discovery_result": {
"error_code": 0,
"result": {
"device_id": "00000000000000000000000000000000",
"device_model": "P135(US)",
"device_type": "SMART.TAPOPLUG",
"factory_default": false,
"ip": "127.0.0.123",
"is_support_iot_cloud": true,
"mac": "F0-09-0D-00-00-00",
"mgt_encrypt_schm": {
"encrypt_type": "KLAP",
"http_port": 80,
"is_support_https": false,
"lv": 2
},
"obd_src": "tplink",
"owner": "00000000000000000000000000000000",
"protocol_version": 1
}
},
"get_antitheft_rules": {
"antitheft_rule_max_count": 1,
"enable": false,
"rule_list": []
},
"get_auto_update_info": {
"enable": true,
"random_range": 120,
"time": 180
},
"get_connect_cloud_state": {
"status": 0
},
"get_countdown_rules": {
"countdown_rule_max_count": 1,
"enable": false,
"rule_list": []
},
"get_device_info": {
"accessory_at_low_battery": false,
"avatar": "plug",
"brightness": 100,
"default_states": {
"re_power_type": "always_off",
"re_power_type_capability": [
"last_states",
"always_on",
"always_off"
],
"type": "last_states"
},
"device_id": "0000000000000000000000000000000000000000",
"device_on": true,
"fw_id": "00000000000000000000000000000000",
"fw_ver": "1.2.0 Build 240415 Rel.171222",
"has_set_location_info": false,
"hw_id": "00000000000000000000000000000000",
"hw_ver": "1.0",
"ip": "127.0.0.123",
"lang": "",
"latitude": 0,
"longitude": 0,
"mac": "F0-09-0D-00-00-00",
"model": "P135",
"nickname": "I01BU0tFRF9OQU1FIw==",
"oem_id": "00000000000000000000000000000000",
"on_time": 3428,
"overheat_status": "normal",
"region": "America/Los_Angeles",
"rssi": -35,
"signal_level": 3,
"specs": "",
"ssid": "I01BU0tFRF9TU0lEIw==",
"time_diff": -480,
"type": "SMART.TAPOPLUG"
},
"get_device_time": {
"region": "America/Los_Angeles",
"time_diff": -480,
"timestamp": 1734735856
},
"get_device_usage": {
"time_usage": {
"past30": 57,
"past7": 57,
"today": 57
}
},
"get_fw_download_state": {
"auto_upgrade": false,
"download_progress": 0,
"reboot_time": 5,
"status": 0,
"upgrade_time": 5
},
"get_inherit_info": null,
"get_latest_fw": {
"fw_size": 0,
"fw_ver": "1.2.0 Build 240415 Rel.171222",
"hw_id": "",
"need_to_upgrade": false,
"oem_id": "",
"release_date": "",
"release_note": "",
"type": 0
},
"get_led_info": {
"bri_config": {
"bri_type": "overall",
"overall_bri": 50
},
"led_rule": "always",
"led_status": true,
"night_mode": {
"end_time": 420,
"night_mode_type": "custom",
"start_time": 1320
}
},
"get_matter_setup_info": {
"setup_code": "00000000000",
"setup_payload": "00:0000000000000000000"
},
"get_next_event": {},
"get_on_off_gradually_info": {
"off_state": {
"duration": 1,
"enable": true,
"max_duration": 60
},
"on_state": {
"duration": 1,
"enable": true,
"max_duration": 60
}
},
"get_preset_rules": {
"brightness": [
100,
75,
50,
25,
1
]
},
"get_schedule_rules": {
"enable": false,
"rule_list": [],
"schedule_rule_max_count": 32,
"start_index": 0,
"sum": 0
},
"get_wireless_scan_info": {
"ap_list": [
{
"bssid": "000000000000",
"channel": 0,
"cipher_type": 2,
"key_type": "wpa2_psk",
"signal_level": 3,
"ssid": "I01BU0tFRF9TU0lEIw=="
},
{
"bssid": "000000000000",
"channel": 0,
"cipher_type": 2,
"key_type": "wpa2_psk",
"signal_level": 3,
"ssid": "I01BU0tFRF9TU0lEIw=="
},
{
"bssid": "000000000000",
"channel": 0,
"cipher_type": 2,
"key_type": "wpa2_psk",
"signal_level": 3,
"ssid": "I01BU0tFRF9TU0lEIw=="
},
{
"bssid": "000000000000",
"channel": 0,
"cipher_type": 2,
"key_type": "wpa2_psk",
"signal_level": 3,
"ssid": "I01BU0tFRF9TU0lEIw=="
},
{
"bssid": "000000000000",
"channel": 0,
"cipher_type": 2,
"key_type": "wpa2_psk",
"signal_level": 2,
"ssid": "I01BU0tFRF9TU0lEIw=="
},
{
"bssid": "000000000000",
"channel": 0,
"cipher_type": 2,
"key_type": "wpa2_psk",
"signal_level": 2,
"ssid": "I01BU0tFRF9TU0lEIw=="
},
{
"bssid": "000000000000",
"channel": 0,
"cipher_type": 2,
"key_type": "wpa2_psk",
"signal_level": 2,
"ssid": "I01BU0tFRF9TU0lEIw=="
},
{
"bssid": "000000000000",
"channel": 0,
"cipher_type": 2,
"key_type": "wpa2_psk",
"signal_level": 2,
"ssid": "I01BU0tFRF9TU0lEIw=="
},
{
"bssid": "000000000000",
"channel": 0,
"cipher_type": 2,
"key_type": "wpa2_psk",
"signal_level": 2,
"ssid": "I01BU0tFRF9TU0lEIw=="
},
{
"bssid": "000000000000",
"channel": 0,
"cipher_type": 2,
"key_type": "wpa2_psk",
"signal_level": 1,
"ssid": "I01BU0tFRF9TU0lEIw=="
},
{
"bssid": "000000000000",
"channel": 0,
"cipher_type": 2,
"key_type": "wpa2_psk",
"signal_level": 1,
"ssid": "I01BU0tFRF9TU0lEIw=="
},
{
"bssid": "000000000000",
"channel": 0,
"cipher_type": 2,
"key_type": "wpa2_psk",
"signal_level": 1,
"ssid": "I01BU0tFRF9TU0lEIw=="
},
{
"bssid": "000000000000",
"channel": 0,
"cipher_type": 2,
"key_type": "wpa2_psk",
"signal_level": 1,
"ssid": "I01BU0tFRF9TU0lEIw=="
},
{
"bssid": "000000000000",
"channel": 0,
"cipher_type": 2,
"key_type": "wpa2_psk",
"signal_level": 1,
"ssid": "I01BU0tFRF9TU0lEIw=="
},
{
"bssid": "000000000000",
"channel": 0,
"cipher_type": 2,
"key_type": "wpa2_psk",
"signal_level": 1,
"ssid": "I01BU0tFRF9TU0lEIw=="
}
],
"start_index": 0,
"sum": 15,
"wep_supported": false
},
"qs_component_nego": {
"component_list": [
{
"id": "quick_setup",
"ver_code": 3
},
{
"id": "sunrise_sunset",
"ver_code": 1
},
{
"id": "ble_whole_setup",
"ver_code": 1
},
{
"id": "matter",
"ver_code": 2
},
{
"id": "iot_cloud",
"ver_code": 1
},
{
"id": "inherit",
"ver_code": 1
},
{
"id": "firmware",
"ver_code": 2
}
],
"extra_info": {
"device_model": "P135",
"device_type": "SMART.TAPOPLUG",
"is_klap": true
}
}
}

1585
tests/fixtures/smart/P210M(US)_1.0_1.0.3.json vendored Normal file
View File

File diff suppressed because it is too large Load Diff

1708
tests/fixtures/smart/P306(US)_1.0_1.1.2.json vendored Normal file
View File

File diff suppressed because it is too large Load Diff

1283
tests/fixtures/smartcam/C225(US)_2.0_1.0.11.json vendored Normal file
View File

File diff suppressed because it is too large Load Diff

3
tests/test_cli.py
View File

@@ -268,7 +268,8 @@ async def test_alias(dev, runner):
res = await runner.invoke(alias, obj=dev)
assert f"Alias: {new_alias}" in res.output
await dev.set_alias(old_alias)
# If alias is None set it back to empty string
await dev.set_alias(old_alias or "")
async def test_raw_command(dev, mocker, runner):

5
tests/test_device.py
View File

@@ -65,12 +65,13 @@ async def test_alias(dev):
test_alias = "TEST1234"
original = dev.alias
assert isinstance(original, str)
assert isinstance(original, str | None)
await dev.set_alias(test_alias)
await dev.update()
assert dev.alias == test_alias
await dev.set_alias(original)
# If alias is None set it back to empty string
await dev.set_alias(original or "")
await dev.update()
assert dev.alias == original

96
tests/transports/test_sslaestransport.py
View File

@@ -15,8 +15,10 @@ from kasa.credentials import DEFAULT_CREDENTIALS, Credentials, get_default_crede
from kasa.deviceconfig import DeviceConfig
from kasa.exceptions import (
AuthenticationError,
DeviceError,
KasaException,
SmartErrorCode,
_RetryableError,
)
from kasa.httpclient import HttpClient
from kasa.transports.aestransport import AesEncyptionSession
@@ -201,6 +203,64 @@ async def test_unencrypted_response(mocker, caplog):
)
async def test_device_blocked_response(mocker):
host = "127.0.0.1"
mock_ssl_aes_device = MockSslAesDevice(host, device_blocked=True)
mocker.patch.object(
aiohttp.ClientSession, "post", side_effect=mock_ssl_aes_device.post
)
transport = SslAesTransport(
config=DeviceConfig(host, credentials=Credentials(MOCK_USER, MOCK_PWD))
)
msg = "Device blocked for 1685 seconds"
with pytest.raises(DeviceError, match=msg):
await transport.perform_handshake()
@pytest.mark.parametrize(
("response", "expected_msg"),
[
pytest.param(
{"error_code": -1, "msg": "Check tapo tag failed"},
'{"error_code": -1, "msg": "Check tapo tag failed"}',
id="can-decrypt",
),
pytest.param(
b"12345678",
str({"result": {"response": "12345678"}, "error_code": 0}),
id="cannot-decrypt",
),
],
)
async def test_device_500_error(mocker, response, expected_msg):
"""Test 500 error raises retryable exception."""
host = "127.0.0.1"
mock_ssl_aes_device = MockSslAesDevice(host)
mocker.patch.object(
aiohttp.ClientSession, "post", side_effect=mock_ssl_aes_device.post
)
transport = SslAesTransport(
config=DeviceConfig(host, credentials=Credentials(MOCK_USER, MOCK_PWD))
)
request = {
"method": "getDeviceInfo",
"params": None,
}
await transport.perform_handshake()
mock_ssl_aes_device.put_next_response(response)
mock_ssl_aes_device.status_code = 500
msg = f"Device 127.0.0.1 replied with status 500 after handshake, response: {expected_msg}"
with pytest.raises(_RetryableError, match=msg):
await transport.send(json_dumps(request))
async def test_port_override():
"""Test that port override sets the app_url."""
host = "127.0.0.1"
@@ -236,6 +296,11 @@ class MockSslAesDevice:
},
}
DEVICE_BLOCKED_RESP = {
"data": {"code": SmartErrorCode.DEVICE_BLOCKED.value, "sec_left": 1685},
"error_code": SmartErrorCode.SESSION_EXPIRED.value,
}
class _mock_response:
def __init__(self, status, request: dict):
self.status = status
@@ -264,6 +329,7 @@ class MockSslAesDevice:
send_error_code=0,
secure_passthrough_error_code=0,
digest_password_fail=False,
device_blocked=False,
):
self.host = host
self.http_client = HttpClient(DeviceConfig(self.host))
@@ -278,6 +344,9 @@ class MockSslAesDevice:
self.do_not_encrypt_response = do_not_encrypt_response
self.want_default_username = want_default_username
self.digest_password_fail = digest_password_fail
self.device_blocked = device_blocked
self._next_responses: list[dict | bytes] = []
async def post(self, url: URL, params=None, json=None, data=None, *_, **__):
if data:
@@ -304,6 +373,9 @@ class MockSslAesDevice:
request_nonce = request["params"].get("cnonce")
request_username = request["params"].get("username")
if self.device_blocked:
return self._mock_response(self.status_code, self.DEVICE_BLOCKED_RESP)
if (self.want_default_username and request_username != MOCK_ADMIN_USER) or (
not self.want_default_username and request_username != MOCK_USER
):
@@ -360,11 +432,24 @@ class MockSslAesDevice:
assert self.encryption_session
decrypted_request = self.encryption_session.decrypt(encrypted_request.encode())
decrypted_request_dict = json_loads(decrypted_request)
decrypted_response = await self._post(url, decrypted_request_dict)
async with decrypted_response:
decrypted_response_data = await decrypted_response.read()
encrypted_response = self.encryption_session.encrypt(decrypted_response_data)
if self._next_responses:
next_response = self._next_responses.pop(0)
if isinstance(next_response, dict):
decrypted_response_data = json_dumps(next_response).encode()
encrypted_response = self.encryption_session.encrypt(
decrypted_response_data
)
else:
encrypted_response = next_response
else:
decrypted_response = await self._post(url, decrypted_request_dict)
async with decrypted_response:
decrypted_response_data = await decrypted_response.read()
encrypted_response = self.encryption_session.encrypt(
decrypted_response_data
)
response = (
decrypted_response_data
if self.do_not_encrypt_response
@@ -379,3 +464,6 @@ class MockSslAesDevice:
async def _return_send_response(self, url: URL, json: dict[str, Any]):
result = {"result": {"method": None}, "error_code": self.send_error_code}
return self._mock_response(self.status_code, result)
def put_next_response(self, request: dict | bytes) -> None:
self._next_responses.append(request)