mirror of
https://github.com/python-kasa/python-kasa.git
synced 2025-08-09 20:24:02 +00:00
Merge remote-tracking branch 'upstream/master' into feat/dev_descriptors
This commit is contained in:
Teemu Rytilahti
173
kasa/tests/fixtures/smart/P100_1.0.0_1.1.3.json
vendored
Normal file
173
kasa/tests/fixtures/smart/P100_1.0.0_1.1.3.json
vendored
Normal file
@@ -0,0 +1,173 @@
|
||||
{
|
||||
"component_nego": {
|
||||
"component_list": [
|
||||
{
|
||||
"id": "device",
|
||||
"ver_code": 1
|
||||
},
|
||||
{
|
||||
"id": "firmware",
|
||||
"ver_code": 1
|
||||
},
|
||||
{
|
||||
"id": "quick_setup",
|
||||
"ver_code": 1
|
||||
},
|
||||
{
|
||||
"id": "time",
|
||||
"ver_code": 1
|
||||
},
|
||||
{
|
||||
"id": "wireless",
|
||||
"ver_code": 1
|
||||
},
|
||||
{
|
||||
"id": "schedule",
|
||||
"ver_code": 1
|
||||
},
|
||||
{
|
||||
"id": "countdown",
|
||||
"ver_code": 1
|
||||
},
|
||||
{
|
||||
"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
|
||||
}
|
||||
]
|
||||
},
|
||||
"discovery_result": {
|
||||
"device_id": "00000000000000000000000000000000",
|
||||
"device_model": "P100",
|
||||
"device_type": "SMART.TAPOPLUG",
|
||||
"factory_default": false,
|
||||
"ip": "127.0.0.123",
|
||||
"mac": "1C-3B-F3-00-00-00",
|
||||
"mgt_encrypt_schm": {
|
||||
"encrypt_type": "AES",
|
||||
"http_port": 80,
|
||||
"is_support_https": false
|
||||
},
|
||||
"owner": "00000000000000000000000000000000"
|
||||
},
|
||||
"get_antitheft_rules": {
|
||||
"antitheft_rule_max_count": 1,
|
||||
"enable": false,
|
||||
"rule_list": []
|
||||
},
|
||||
"get_connect_cloud_state": {
|
||||
"status": -1001
|
||||
},
|
||||
"get_countdown_rules": {
|
||||
"countdown_rule_max_count": 1,
|
||||
"enable": false,
|
||||
"rule_list": []
|
||||
},
|
||||
"get_device_info": {
|
||||
"avatar": "plug",
|
||||
"device_id": "0000000000000000000000000000000000000000",
|
||||
"device_on": true,
|
||||
"fw_id": "00000000000000000000000000000000",
|
||||
"fw_ver": "1.1.3 Build 20191017 Rel. 57937",
|
||||
"has_set_location_info": true,
|
||||
"hw_id": "00000000000000000000000000000000",
|
||||
"hw_ver": "1.0.0",
|
||||
"ip": "127.0.0.123",
|
||||
"latitude": 0,
|
||||
"location": "hallway",
|
||||
"longitude": 0,
|
||||
"mac": "1C-3B-F3-00-00-00",
|
||||
"model": "P100",
|
||||
"nickname": "I01BU0tFRF9OQU1FIw==",
|
||||
"oem_id": "00000000000000000000000000000000",
|
||||
"on_time": 6868,
|
||||
"overheated": false,
|
||||
"signal_level": 2,
|
||||
"specs": "US",
|
||||
"ssid": "I01BU0tFRF9TU0lEIw==",
|
||||
"time_usage_past30": 114,
|
||||
"time_usage_past7": 114,
|
||||
"time_usage_today": 114,
|
||||
"type": "SMART.TAPOPLUG"
|
||||
},
|
||||
"get_device_time": {
|
||||
"region": "Europe/London",
|
||||
"time_diff": 0,
|
||||
"timestamp": 1707905077
|
||||
},
|
||||
"get_fw_download_state": {
|
||||
"download_progress": 0,
|
||||
"reboot_time": 10,
|
||||
"status": 0,
|
||||
"upgrade_time": 0
|
||||
},
|
||||
"get_latest_fw": {
|
||||
"fw_size": 786432,
|
||||
"fw_ver": "1.3.7 Build 20230711 Rel.61904",
|
||||
"hw_id": "00000000000000000000000000000000",
|
||||
"need_to_upgrade": true,
|
||||
"oem_id": "00000000000000000000000000000000",
|
||||
"release_date": "2023-07-26",
|
||||
"release_note": "Modifications and Bug fixes:\nEnhanced device security.",
|
||||
"type": 3
|
||||
},
|
||||
"get_led_info": {
|
||||
"led_rule": "always",
|
||||
"led_status": true
|
||||
},
|
||||
"get_next_event": {
|
||||
"action": -1,
|
||||
"e_time": 0,
|
||||
"id": "0",
|
||||
"s_time": 0,
|
||||
"type": 0
|
||||
},
|
||||
"get_schedule_rules": {
|
||||
"enable": false,
|
||||
"rule_list": [],
|
||||
"schedule_rule_max_count": 20,
|
||||
"start_index": 0,
|
||||
"sum": 0
|
||||
},
|
||||
"get_wireless_scan_info": {
|
||||
"ap_list": [],
|
||||
"start_index": 0,
|
||||
"sum": 0,
|
||||
"wep_supported": false
|
||||
},
|
||||
"qs_component_nego": {
|
||||
"component_list": [
|
||||
{
|
||||
"id": "quick_setup",
|
||||
"ver_code": 1
|
||||
},
|
||||
{
|
||||
"id": "sunrise_sunset",
|
||||
"ver_code": 1
|
||||
}
|
||||
],
|
||||
"extra_info": {
|
||||
"device_model": "P100",
|
||||
"device_type": "SMART.TAPOPLUG"
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,5 +1,6 @@
|
||||
import base64
|
||||
import json
|
||||
import logging
|
||||
import random
|
||||
import string
|
||||
import time
|
||||
@@ -180,6 +181,67 @@ async def test_send(mocker, status_code, error_code, inner_error_code, expectati
|
||||
assert "result" in res
|
||||
|
||||
|
||||
async def test_unencrypted_response(mocker, caplog):
|
||||
host = "127.0.0.1"
|
||||
mock_aes_device = MockAesDevice(host, 200, 0, 0, do_not_encrypt_response=True)
|
||||
mocker.patch.object(aiohttp.ClientSession, "post", side_effect=mock_aes_device.post)
|
||||
|
||||
transport = AesTransport(
|
||||
config=DeviceConfig(host, credentials=Credentials("foo", "bar"))
|
||||
)
|
||||
transport._state = TransportState.ESTABLISHED
|
||||
transport._session_expire_at = time.time() + 86400
|
||||
transport._encryption_session = mock_aes_device.encryption_session
|
||||
transport._token_url = transport._app_url.with_query(
|
||||
f"token={mock_aes_device.token}"
|
||||
)
|
||||
|
||||
request = {
|
||||
"method": "get_device_info",
|
||||
"params": None,
|
||||
"request_time_milis": round(time.time() * 1000),
|
||||
"requestID": 1,
|
||||
"terminal_uuid": "foobar",
|
||||
}
|
||||
caplog.set_level(logging.DEBUG)
|
||||
res = await transport.send(json_dumps(request))
|
||||
assert "result" in res
|
||||
assert (
|
||||
"Received unencrypted response over secure passthrough from 127.0.0.1"
|
||||
in caplog.text
|
||||
)
|
||||
|
||||
|
||||
async def test_unencrypted_response_invalid_json(mocker, caplog):
|
||||
host = "127.0.0.1"
|
||||
mock_aes_device = MockAesDevice(
|
||||
host, 200, 0, 0, do_not_encrypt_response=True, send_response=b"Foobar"
|
||||
)
|
||||
mocker.patch.object(aiohttp.ClientSession, "post", side_effect=mock_aes_device.post)
|
||||
|
||||
transport = AesTransport(
|
||||
config=DeviceConfig(host, credentials=Credentials("foo", "bar"))
|
||||
)
|
||||
transport._state = TransportState.ESTABLISHED
|
||||
transport._session_expire_at = time.time() + 86400
|
||||
transport._encryption_session = mock_aes_device.encryption_session
|
||||
transport._token_url = transport._app_url.with_query(
|
||||
f"token={mock_aes_device.token}"
|
||||
)
|
||||
|
||||
request = {
|
||||
"method": "get_device_info",
|
||||
"params": None,
|
||||
"request_time_milis": round(time.time() * 1000),
|
||||
"requestID": 1,
|
||||
"terminal_uuid": "foobar",
|
||||
}
|
||||
caplog.set_level(logging.DEBUG)
|
||||
msg = f"Unable to decrypt response from {host}, error: Incorrect padding, response: Foobar"
|
||||
with pytest.raises(SmartDeviceException, match=msg):
|
||||
await transport.send(json_dumps(request))
|
||||
|
||||
|
||||
ERRORS = [e for e in SmartErrorCode if e != 0]
|
||||
|
||||
|
||||
@@ -233,15 +295,28 @@ class MockAesDevice:
|
||||
pass
|
||||
|
||||
async def read(self):
|
||||
return json_dumps(self._json).encode()
|
||||
if isinstance(self._json, dict):
|
||||
return json_dumps(self._json).encode()
|
||||
return self._json
|
||||
|
||||
encryption_session = AesEncyptionSession(KEY_IV[:16], KEY_IV[16:])
|
||||
|
||||
def __init__(self, host, status_code=200, error_code=0, inner_error_code=0):
|
||||
def __init__(
|
||||
self,
|
||||
host,
|
||||
status_code=200,
|
||||
error_code=0,
|
||||
inner_error_code=0,
|
||||
*,
|
||||
do_not_encrypt_response=False,
|
||||
send_response=None,
|
||||
):
|
||||
self.host = host
|
||||
self.status_code = status_code
|
||||
self.error_code = error_code
|
||||
self._inner_error_code = inner_error_code
|
||||
self.do_not_encrypt_response = do_not_encrypt_response
|
||||
self.send_response = send_response
|
||||
self.http_client = HttpClient(DeviceConfig(self.host))
|
||||
self.inner_call_count = 0
|
||||
self.token = "".join(random.choices(string.ascii_uppercase, k=32)) # noqa: S311
|
||||
@@ -289,13 +364,15 @@ class MockAesDevice:
|
||||
decrypted_request_dict = json_loads(decrypted_request)
|
||||
decrypted_response = await self._post(url, decrypted_request_dict)
|
||||
async with decrypted_response:
|
||||
response_data = await decrypted_response.read()
|
||||
decrypted_response_dict = json_loads(response_data.decode())
|
||||
encrypted_response = self.encryption_session.encrypt(
|
||||
json_dumps(decrypted_response_dict).encode()
|
||||
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
|
||||
else encrypted_response
|
||||
)
|
||||
result = {
|
||||
"result": {"response": encrypted_response.decode()},
|
||||
"result": {"response": response.decode()},
|
||||
"error_code": self.error_code,
|
||||
}
|
||||
return self._mock_response(self.status_code, result)
|
||||
@@ -310,5 +387,6 @@ class MockAesDevice:
|
||||
|
||||
async def _return_send_response(self, url: URL, json: Dict[str, Any]):
|
||||
result = {"result": {"method": None}, "error_code": self.inner_error_code}
|
||||
response = self.send_response if self.send_response else result
|
||||
self.inner_call_count += 1
|
||||
return self._mock_response(self.status_code, result)
|
||||
return self._mock_response(self.status_code, response)
|
||||
|
||||
@@ -7,6 +7,7 @@ from asyncclick.testing import CliRunner
|
||||
|
||||
from kasa import (
|
||||
AuthenticationException,
|
||||
Credentials,
|
||||
Device,
|
||||
EmeterStatus,
|
||||
SmartDeviceException,
|
||||
@@ -351,7 +352,9 @@ async def test_credentials(discovery_mock, mocker):
|
||||
async def test_without_device_type(dev, mocker):
|
||||
"""Test connecting without the device type."""
|
||||
runner = CliRunner()
|
||||
mocker.patch("kasa.discover.Discover.discover_single", return_value=dev)
|
||||
discovery_mock = mocker.patch(
|
||||
"kasa.discover.Discover.discover_single", return_value=dev
|
||||
)
|
||||
# These will mock the features to avoid accessing non-existing
|
||||
mocker.patch("kasa.device.Device.features", return_value={})
|
||||
mocker.patch("kasa.iot.iotdevice.IotDevice.features", return_value={})
|
||||
@@ -365,9 +368,18 @@ async def test_without_device_type(dev, mocker):
|
||||
"foo",
|
||||
"--password",
|
||||
"bar",
|
||||
"--discovery-timeout",
|
||||
"7",
|
||||
],
|
||||
)
|
||||
assert res.exit_code == 0
|
||||
discovery_mock.assert_called_once_with(
|
||||
"127.0.0.1",
|
||||
port=None,
|
||||
credentials=Credentials("foo", "bar"),
|
||||
timeout=5,
|
||||
discovery_timeout=7,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("auth_param", ["--username", "--password"])
|
||||
|
||||
@@ -53,7 +53,7 @@ async def test_connect(
|
||||
host=host, credentials=Credentials("foor", "bar"), connection_type=ctype
|
||||
)
|
||||
protocol_class = get_protocol(config).__class__
|
||||
|
||||
close_mock = mocker.patch.object(protocol_class, "close")
|
||||
dev = await connect(
|
||||
config=config,
|
||||
)
|
||||
@@ -61,8 +61,9 @@ async def test_connect(
|
||||
assert isinstance(dev.protocol, protocol_class)
|
||||
|
||||
assert dev.config == config
|
||||
|
||||
assert close_mock.call_count == 0
|
||||
await dev.disconnect()
|
||||
assert close_mock.call_count == 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize("custom_port", [123, None])
|
||||
@@ -116,8 +117,12 @@ async def test_connect_query_fails(all_fixture_data: dict, mocker):
|
||||
config = DeviceConfig(
|
||||
host=host, credentials=Credentials("foor", "bar"), connection_type=ctype
|
||||
)
|
||||
protocol_class = get_protocol(config).__class__
|
||||
close_mock = mocker.patch.object(protocol_class, "close")
|
||||
assert close_mock.call_count == 0
|
||||
with pytest.raises(SmartDeviceException):
|
||||
await connect(config=config)
|
||||
assert close_mock.call_count == 1
|
||||
|
||||
|
||||
async def test_connect_http_client(all_fixture_data, mocker):
|
||||
|
||||
Reference in New Issue
Block a user