Mirrored_Repos/python-kasa
1
0
Fork 0
mirror of https://github.com/python-kasa/python-kasa.git synced 2026-05-01 04:57:44 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'upstream/master' into feat/light_module_feats

Browse Source
This commit is contained in:
Steven B
2024-12-11 13:21:46 +00:00
parent 2b34b9efc6 032cd5d2cc
commit feed5d18a1
78 changed files with 5065 additions and 854 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
tests/device_fixtures.py
View File

@@ -98,6 +98,7 @@ PLUGS = {
SWITCHES_IOT = {
"HS200",
"HS210",
"KS200",
"KS200M",
}
SWITCHES_SMART = {
@@ -217,6 +218,9 @@ no_emeter = parametrize(
model_filter=ALL_DEVICES - WITH_EMETER,
protocol_filter={"SMART", "IOT"},
)
has_emeter_smart = parametrize(
"has emeter smart", model_filter=WITH_EMETER_SMART, protocol_filter={"SMART"}
)
has_emeter_iot = parametrize(
"has emeter iot", model_filter=WITH_EMETER_IOT, protocol_filter={"IOT"}
)

28
tests/discovery_fixtures.py
View File

@@ -22,6 +22,29 @@ class DiscoveryResponse(TypedDict):
error_code: int
UNSUPPORTED_HOMEWIFISYSTEM = {
"error_code": 0,
"result": {
"channel_2g": "10",
"channel_5g": "44",
"device_id": "REDACTED_51f72a752213a6c45203530",
"device_model": "X20",
"device_type": "HOMEWIFISYSTEM",
"factory_default": False,
"group_id": "REDACTED_07d902da02fa9beab8a64",
"group_name": "I01BU0tFRF9TU0lEIw==", # '#MASKED_SSID#'
"hardware_version": "3.0",
"ip": "192.168.1.192",
"mac": "24:2F:D0:00:00:00",
"master_device_id": "REDACTED_51f72a752213a6c45203530",
"need_account_digest": True,
"owner": "REDACTED_341c020d7e8bda184e56a90",
"role": "master",
"tmp_port": [20001],
},
}
def _make_unsupported(
device_family,
encrypt_type,
@@ -75,13 +98,14 @@ UNSUPPORTED_DEVICES = {
"unable_to_parse": _make_unsupported(
"SMART.TAPOBULB",
"FOO",
omit_keys={"mgt_encrypt_schm": None},
omit_keys={"device_id": None},
),
"invalidinstance": _make_unsupported(
"IOT.SMARTPLUGSWITCH",
"KLAP",
https=True,
),
"homewifi": UNSUPPORTED_HOMEWIFISYSTEM,
}
@@ -106,6 +130,8 @@ new_discovery = parametrize_discovery(
"new discovery", data_root_filter="discovery_result"
)
smart_discovery = parametrize_discovery("smart discovery", protocol_filter={"SMART"})
@pytest.fixture(
params=filter_fixtures("discoverable", protocol_filter={"SMART", "IOT"}),

13
tests/fakeprotocol_smart.py
View File

@@ -449,6 +449,17 @@ class FakeSmartTransport(BaseTransport):
info["get_preset_rules"]["states"][params["index"]] = params["state"]
return {"error_code": 0}
def _set_temperature_unit(self, info, params):
"""Set or remove values as per the device behaviour."""
unit = params["temp_unit"]
if unit not in {"celsius", "fahrenheit"}:
raise ValueError(f"Invalid value for temperature unit {unit}")
if "temp_unit" not in info["get_device_info"]:
return {"error_code": SmartErrorCode.UNKNOWN_METHOD_ERROR}
else:
info["get_device_info"]["temp_unit"] = unit
return {"error_code": 0}
def _update_sysinfo_key(self, info: dict, key: str, value: str) -> dict:
"""Update a single key in the main system info.
@@ -551,6 +562,8 @@ class FakeSmartTransport(BaseTransport):
return self._set_preset_rules(info, params)
elif method == "edit_preset_rules":
return self._edit_preset_rules(info, params)
elif method == "set_temperature_unit":
return self._set_temperature_unit(info, params)
elif method == "set_on_off_gradually_info":
return self._set_on_off_gradually_info(info, params)
elif method == "set_child_protection":

63
tests/fixtures/iot/KS200(US)_1.0_1.0.8.json vendored Normal file
View File

@@ -0,0 +1,63 @@
{
"cnCloud": {
"get_info": {
"binded": 1,
"cld_connection": 1,
"err_code": 0,
"fwDlPage": "",
"fwNotifyType": -1,
"illegalType": 0,
"server": "n-devs.tplinkcloud.com",
"stopConnect": 0,
"tcspInfo": "",
"tcspStatus": 1,
"username": "#MASKED_NAME#"
},
"get_intl_fw_list": {
"err_code": 0,
"fw_list": []
}
},
"schedule": {
"get_next_action": {
"err_code": 0,
"type": -1
},
"get_rules": {
"enable": 1,
"err_code": 0,
"rule_list": [],
"version": 2
}
},
"system": {
"get_sysinfo": {
"active_mode": "none",
"alias": "#MASKED_NAME#",
"dev_name": "Smart Wi-Fi Light Switch",
"deviceId": "0000000000000000000000000000000000000000",
"err_code": 0,
"feature": "TIM",
"hwId": "00000000000000000000000000000000",
"hw_ver": "1.0",
"icon_hash": "",
"latitude_i": 0,
"led_off": 0,
"longitude_i": 0,
"mac": "A8:42:A1:00:00:00",
"mic_type": "IOT.SMARTPLUGSWITCH",
"model": "KS200(US)",
"next_action": {
"type": -1
},
"obd_src": "tplink",
"oemId": "00000000000000000000000000000000",
"on_time": 0,
"relay_state": 0,
"rssi": -46,
"status": "new",
"sw_ver": "1.0.8 Build 240424 Rel.101842",
"updating": 0
}
}
}

86
tests/fixtures/iotcam/EC60(US)_4.0_2.3.22.json vendored Normal file
View File

@@ -0,0 +1,86 @@
{
"emeter": {
"get_realtime": {
"err_code": -10008,
"err_msg": "Unsupported API call."
}
},
"smartlife.iot.LAS": {
"get_current_brt": {
"err_code": -10008,
"err_msg": "Unsupported API call."
}
},
"smartlife.iot.PIR": {
"get_config": {
"err_code": -10008,
"err_msg": "Unsupported API call."
}
},
"smartlife.iot.common.emeter": {
"get_realtime": {
"err_code": -10008,
"err_msg": "Unsupported API call."
}
},
"smartlife.iot.dimmer": {
"get_dimmer_parameters": {
"err_code": -10008,
"err_msg": "Unsupported API call."
}
},
"smartlife.iot.smartbulb.lightingservice": {
"get_light_state": {
"err_code": -10008,
"err_msg": "Unsupported API call."
}
},
"system": {
"get_sysinfo": {
"err_code": 0,
"system": {
"a_type": 2,
"alias": "#MASKED_NAME#",
"bind_status": false,
"c_opt": [
0,
1
],
"camera_switch": "on",
"dev_name": "Kasa Spot, 24/7 Recording",
"deviceId": "0000000000000000000000000000000000000000",
"f_list": [
1,
2
],
"hwId": "00000000000000000000000000000000",
"hw_ver": "4.0",
"is_cal": 1,
"last_activity_timestamp": 0,
"latitude": 0,
"led_status": "on",
"longitude": 0,
"mac": "74:FE:CE:00:00:00",
"mic_mac": "74FECE000000",
"model": "EC60(US)",
"new_feature": [
2,
3,
4,
5,
7,
9
],
"oemId": "00000000000000000000000000000000",
"resolution": "720P",
"rssi": -28,
"status": "new",
"stream_version": 2,
"sw_ver": "2.3.22 Build 20230731 rel.69808",
"system_time": 1690827820,
"type": "IOT.IPCAMERA",
"updating": false
}
}
}
}

614
tests/fixtures/smart/P110M(EU)_1.0_1.2.3.json vendored Normal file
View File

@@ -0,0 +1,614 @@
{
"component_nego": {
"component_list": [
{
"id": "device",
"ver_code": 2
},
{
"id": "firmware",
"ver_code": 2
},
{
"id": "quick_setup",
"ver_code": 3
},
{
"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": "auto_off",
"ver_code": 2
},
{
"id": "localSmart",
"ver_code": 1
},
{
"id": "energy_monitoring",
"ver_code": 2
},
{
"id": "power_protection",
"ver_code": 1
},
{
"id": "charging_protection",
"ver_code": 2
},
{
"id": "matter",
"ver_code": 2
},
{
"id": "current_protection",
"ver_code": 1
}
]
},
"discovery_result": {
"device_id": "00000000000000000000000000000000",
"device_model": "P110M(EU)",
"device_type": "SMART.TAPOPLUG",
"factory_default": false,
"ip": "127.0.0.123",
"is_support_iot_cloud": true,
"mac": "F0-A7-31-00-00-00",
"mgt_encrypt_schm": {
"encrypt_type": "KLAP",
"http_port": 80,
"is_support_https": false,
"lv": 2
},
"obd_src": "matter",
"owner": ""
},
"get_antitheft_rules": {
"antitheft_rule_max_count": 1,
"enable": false,
"rule_list": []
},
"get_auto_off_config": {
"delay_min": 120,
"enable": false
},
"get_auto_update_info": {
"enable": false,
"random_range": 120,
"time": 180
},
"get_connect_cloud_state": {
"status": 1
},
"get_countdown_rules": {
"countdown_rule_max_count": 1,
"enable": false,
"rule_list": []
},
"get_current_power": {
"current_power": 0
},
"get_device_info": {
"auto_off_remain_time": 0,
"auto_off_status": "off",
"avatar": "",
"charging_status": "normal",
"default_states": {
"state": {},
"type": "last_states"
},
"device_id": "0000000000000000000000000000000000000000",
"device_on": false,
"fw_id": "00000000000000000000000000000000",
"fw_ver": "1.2.3 Build 240617 Rel.153525",
"has_set_location_info": false,
"hw_id": "00000000000000000000000000000000",
"hw_ver": "1.0",
"ip": "127.0.0.123",
"lang": "",
"latitude": 0,
"longitude": 0,
"mac": "F0-A7-31-00-00-00",
"model": "P110M",
"nickname": "I01BU0tFRF9OQU1FIw==",
"oem_id": "00000000000000000000000000000000",
"on_time": 0,
"overcurrent_status": "normal",
"overheat_status": "normal",
"power_protection_status": "normal",
"region": "CET",
"rssi": -33,
"signal_level": 3,
"specs": "",
"ssid": "I01BU0tFRF9TU0lEIw==",
"time_diff": 60,
"type": "SMART.TAPOPLUG"
},
"get_device_time": {
"region": "CET",
"time_diff": 60,
"timestamp": 1732361090
},
"get_device_usage": {
"power_usage": {
"past30": 7892,
"past7": 1549,
"today": 0
},
"saved_power": {
"past30": 9381,
"past7": 1362,
"today": 0
},
"time_usage": {
"past30": 17273,
"past7": 2911,
"today": 0
}
},
"get_electricity_price_config": {
"constant_price": 0,
"time_of_use_config": {
"summer": {
"midpeak": 0,
"offpeak": 0,
"onpeak": 0,
"period": [
0,
0,
0,
0
],
"weekday_config": [
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1
],
"weekend_config": [
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1
]
},
"winter": {
"midpeak": 0,
"offpeak": 0,
"onpeak": 0,
"period": [
0,
0,
0,
0
],
"weekday_config": [
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1
],
"weekend_config": [
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1
]
}
},
"type": "constant"
},
"get_emeter_data": {
"current_ma": 0,
"energy_wh": 1469,
"power_mw": 0,
"voltage_mv": 233509
},
"get_emeter_vgain_igain": {
"igain": 11299,
"vgain": 124300
},
"get_energy_usage": {
"current_power": 0,
"electricity_charge": [
0,
0,
0
],
"local_time": "2024-11-23 12:24:51",
"month_energy": 6266,
"month_runtime": 12705,
"today_energy": 0,
"today_runtime": 0
},
"get_fw_download_state": {
"auto_upgrade": false,
"download_progress": 0,
"reboot_time": 5,
"status": 0,
"upgrade_time": 5
},
"get_led_info": {
"bri_config": {
"bri_type": "overall",
"overall_bri": 50
},
"led_rule": "always",
"led_status": false,
"night_mode": {
"end_time": 420,
"night_mode_type": "sunrise_sunset",
"start_time": 1140,
"sunrise_offset": 0,
"sunset_offset": 0
}
},
"get_matter_setup_info": {
"setup_code": "00000000000",
"setup_payload": "00:000000-000000000000"
},
"get_max_power": {
"max_power": 3896
},
"get_next_event": {},
"get_protection_power": {
"enabled": false,
"protection_power": 0
},
"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": 2,
"ssid": "I01BU0tFRF9TU0lEIw=="
},
{
"bssid": "000000000000",
"channel": 0,
"cipher_type": 0,
"key_type": "none",
"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": 0,
"key_type": "none",
"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": 0,
"key_type": "none",
"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": 0,
"key_type": "none",
"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=="
},
{
"bssid": "000000000000",
"channel": 0,
"cipher_type": 2,
"key_type": "wpa2_psk",
"signal_level": 1,
"ssid": "I01BU0tFRF9TU0lEIw=="
}
],
"start_index": 0,
"sum": 22,
"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": "P110M",
"device_type": "SMART.TAPOPLUG",
"is_klap": true
}
}
}

640
tests/fixtures/smart/P115(US)_1.0_1.1.3.json vendored Normal file
View File

@@ -0,0 +1,640 @@
{
"component_nego": {
"component_list": [
{
"id": "device",
"ver_code": 2
},
{
"id": "firmware",
"ver_code": 2
},
{
"id": "quick_setup",
"ver_code": 3
},
{
"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": "auto_off",
"ver_code": 2
},
{
"id": "localSmart",
"ver_code": 1
},
{
"id": "energy_monitoring",
"ver_code": 2
},
{
"id": "power_protection",
"ver_code": 1
},
{
"id": "charging_protection",
"ver_code": 2
},
{
"id": "overheat_protection",
"ver_code": 1
},
{
"id": "current_protection",
"ver_code": 1
}
]
},
"discovery_result": {
"device_id": "00000000000000000000000000000000",
"device_model": "P115(US)",
"device_type": "SMART.TAPOPLUG",
"factory_default": false,
"ip": "127.0.0.123",
"is_support_iot_cloud": true,
"mac": "B0-19-21-00-00-00",
"mgt_encrypt_schm": {
"encrypt_type": "KLAP",
"http_port": 80,
"is_support_https": false,
"lv": 2
},
"obd_src": "tplink",
"owner": "00000000000000000000000000000000"
},
"get_antitheft_rules": {
"antitheft_rule_max_count": 1,
"enable": false,
"rule_list": []
},
"get_auto_off_config": {
"delay_min": 120,
"enable": false
},
"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_current_power": {
"current_power": 0
},
"get_device_info": {
"auto_off_remain_time": 0,
"auto_off_status": "off",
"avatar": "plug",
"charging_status": "normal",
"default_states": {
"state": {},
"type": "last_states"
},
"device_id": "0000000000000000000000000000000000000000",
"device_on": false,
"fw_id": "00000000000000000000000000000000",
"fw_ver": "1.1.3 Build 240523 Rel.175054",
"has_set_location_info": true,
"hw_id": "00000000000000000000000000000000",
"hw_ver": "1.0",
"ip": "127.0.0.123",
"lang": "en_US",
"latitude": 0,
"longitude": 0,
"mac": "B0-19-21-00-00-00",
"model": "P115",
"nickname": "I01BU0tFRF9OQU1FIw==",
"oem_id": "00000000000000000000000000000000",
"on_time": 0,
"overcurrent_status": "normal",
"overheat_status": "normal",
"power_protection_status": "normal",
"region": "America/Indiana/Indianapolis",
"rssi": -54,
"signal_level": 2,
"specs": "US",
"ssid": "I01BU0tFRF9TU0lEIw==",
"time_diff": -300,
"type": "SMART.TAPOPLUG"
},
"get_device_time": {
"region": "America/Indiana/Indianapolis",
"time_diff": -300,
"timestamp": 1733673137
},
"get_device_usage": {
"power_usage": {
"past30": 4376,
"past7": 1879,
"today": 0
},
"saved_power": {
"past30": 8618,
"past7": 69,
"today": 0
},
"time_usage": {
"past30": 12994,
"past7": 1948,
"today": 0
}
},
"get_electricity_price_config": {
"constant_price": 0,
"time_of_use_config": {
"summer": {
"midpeak": 0,
"offpeak": 0,
"onpeak": 0,
"period": [
0,
0,
0,
0
],
"weekday_config": [
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1
],
"weekend_config": [
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1
]
},
"winter": {
"midpeak": 0,
"offpeak": 0,
"onpeak": 0,
"period": [
0,
0,
0,
0
],
"weekday_config": [
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1
],
"weekend_config": [
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1,
1
]
}
},
"type": "constant"
},
"get_emeter_data": {
"current_ma": 30,
"energy_wh": 1465,
"power_mw": 0,
"voltage_mv": 122133
},
"get_emeter_vgain_igain": {
"igain": 11101,
"vgain": 125071
},
"get_energy_usage": {
"current_power": 0,
"electricity_charge": [
0,
0,
0
],
"local_time": "2024-12-08 10:52:19",
"month_energy": 2532,
"month_runtime": 2630,
"today_energy": 0,
"today_runtime": 0
},
"get_fw_download_state": {
"auto_upgrade": false,
"download_progress": 0,
"reboot_time": 5,
"status": 0,
"upgrade_time": 5
},
"get_latest_fw": {
"fw_size": 0,
"fw_ver": "1.1.3 Build 240523 Rel.175054",
"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": false,
"night_mode": {
"end_time": 476,
"night_mode_type": "sunrise_sunset",
"start_time": 1040,
"sunrise_offset": 0,
"sunset_offset": 0
}
},
"get_max_power": {
"max_power": 1934
},
"get_next_event": {},
"get_protection_power": {
"enabled": false,
"protection_power": 0
},
"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": 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": 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": 0,
"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=="
},
{
"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": 0,
"key_type": "none",
"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=="
},
{
"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": 25,
"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": "iot_cloud",
"ver_code": 1
},
{
"id": "inherit",
"ver_code": 1
},
{
"id": "firmware",
"ver_code": 2
}
],
"extra_info": {
"device_model": "P115",
"device_type": "SMART.TAPOPLUG",
"is_klap": true
}
}
}

1026
tests/fixtures/smartcam/C520WS(US)_1.0_1.2.8.json vendored Normal file
View File

File diff suppressed because it is too large Load Diff

96
tests/test_emeter.py → tests/iot/modules/test_emeter.py
View File

@@ -12,25 +12,23 @@ from voluptuous import (
from kasa import Device, DeviceType, EmeterStatus, Module
from kasa.interfaces.energy import Energy
from kasa.iot import IotDevice, IotStrip
from kasa.iot import IotStrip
from kasa.iot.modules.emeter import Emeter
from kasa.smart import SmartDevice
from kasa.smart.modules import Energy as SmartEnergyModule
from kasa.smart.smartmodule import SmartModule
from .conftest import has_emeter, has_emeter_iot, no_emeter
from tests.conftest import has_emeter_iot, no_emeter_iot
CURRENT_CONSUMPTION_SCHEMA = Schema(
Any(
{
"voltage": Any(All(float, Range(min=0, max=300)), None),
"power": Any(Coerce(float), None),
"total": Any(Coerce(float), None),
"current": Any(All(float), None),
"voltage_mv": Any(All(float, Range(min=0, max=300000)), int, None),
"power_mw": Any(Coerce(float), None),
"total_wh": Any(Coerce(float), None),
"current_ma": Any(All(float), int, None),
"energy_wh": Any(Coerce(float), None),
"total_wh": Any(Coerce(float), None),
"voltage": Any(All(float, Range(min=0, max=300)), None),
"power": Any(Coerce(float), None),
"current": Any(All(float), None),
"total": Any(Coerce(float), None),
"energy": Any(Coerce(float), None),
"slot_id": Any(Coerce(int), None),
},
None,
@@ -38,33 +36,30 @@ CURRENT_CONSUMPTION_SCHEMA = Schema(
)
@no_emeter
@no_emeter_iot
async def test_no_emeter(dev):
assert not dev.has_emeter
with pytest.raises(AttributeError):
await dev.get_emeter_realtime()
# Only iot devices support the historical stats so other
# devices will not implement the methods below
if isinstance(dev, IotDevice):
with pytest.raises(AttributeError):
await dev.get_emeter_daily()
with pytest.raises(AttributeError):
await dev.get_emeter_monthly()
with pytest.raises(AttributeError):
await dev.erase_emeter_stats()
with pytest.raises(AttributeError):
await dev.get_emeter_daily()
with pytest.raises(AttributeError):
await dev.get_emeter_monthly()
with pytest.raises(AttributeError):
await dev.erase_emeter_stats()
@has_emeter
@has_emeter_iot
async def test_get_emeter_realtime(dev):
if isinstance(dev, SmartDevice):
mod = SmartEnergyModule(dev, str(Module.Energy))
if not await mod._check_supported():
pytest.skip(f"Energy module not supported for {dev}.")
emeter = dev.modules[Module.Energy]
current_emeter = await emeter.get_status()
# Check realtime query gets the same value as status property
# iot _query_helper strips out the error code from module responses.
# but it's not stripped out of the _modular_update queries.
assert current_emeter == {k: v for k, v in emeter.status.items() if k != "err_code"}
CURRENT_CONSUMPTION_SCHEMA(current_emeter)
@@ -130,7 +125,7 @@ async def test_emeter_status(dev):
@pytest.mark.skip("not clearing your stats..")
@has_emeter
@has_emeter_iot
async def test_erase_emeter_stats(dev):
emeter = dev.modules[Module.Energy]
@@ -185,37 +180,22 @@ async def test_emeter_daily():
assert emeter.consumption_today == 0.500
@has_emeter
@has_emeter_iot
async def test_supported(dev: Device):
if isinstance(dev, SmartDevice):
mod = SmartEnergyModule(dev, str(Module.Energy))
if not await mod._check_supported():
pytest.skip(f"Energy module not supported for {dev}.")
energy_module = dev.modules.get(Module.Energy)
assert energy_module
if isinstance(dev, IotDevice):
info = (
dev._last_update
if not isinstance(dev, IotStrip)
else dev.children[0].internal_state
)
emeter = info[energy_module._module]["get_realtime"]
has_total = "total" in emeter or "total_wh" in emeter
has_voltage_current = "voltage" in emeter or "voltage_mv" in emeter
assert (
energy_module.supports(Energy.ModuleFeature.CONSUMPTION_TOTAL) is has_total
)
assert (
energy_module.supports(Energy.ModuleFeature.VOLTAGE_CURRENT)
is has_voltage_current
)
assert energy_module.supports(Energy.ModuleFeature.PERIODIC_STATS) is True
else:
assert isinstance(energy_module, SmartModule)
assert energy_module.supports(Energy.ModuleFeature.CONSUMPTION_TOTAL) is False
assert energy_module.supports(Energy.ModuleFeature.PERIODIC_STATS) is False
if energy_module.supported_version < 2:
assert energy_module.supports(Energy.ModuleFeature.VOLTAGE_CURRENT) is False
else:
assert energy_module.supports(Energy.ModuleFeature.VOLTAGE_CURRENT) is True
info = (
dev._last_update
if not isinstance(dev, IotStrip)
else dev.children[0].internal_state
)
emeter = info[energy_module._module]["get_realtime"]
has_total = "total" in emeter or "total_wh" in emeter
has_voltage_current = "voltage" in emeter or "voltage_mv" in emeter
assert energy_module.supports(Energy.ModuleFeature.CONSUMPTION_TOTAL) is has_total
assert (
energy_module.supports(Energy.ModuleFeature.VOLTAGE_CURRENT)
is has_voltage_current
)
assert energy_module.supports(Energy.ModuleFeature.PERIODIC_STATS) is True

0
tests/test_usage.py → tests/iot/modules/test_usage.py
View File

320
tests/iot/test_iotbulb.py Normal file
View File

@@ -0,0 +1,320 @@
from __future__ import annotations
import re
import pytest
from voluptuous import (
All,
Boolean,
Optional,
Range,
Schema,
)
from kasa import Device, IotLightPreset, KasaException, LightState, Module
from kasa.iot import IotBulb, IotDimmer
from kasa.iot.modules import LightPreset as IotLightPresetModule
from tests.conftest import (
bulb_iot,
color_bulb_iot,
dimmable_iot,
handle_turn_on,
non_dimmable_iot,
turn_on,
variable_temp_iot,
)
from tests.iot.test_iotdevice import SYSINFO_SCHEMA
@bulb_iot
async def test_bulb_sysinfo(dev: Device):
assert dev.sys_info is not None
SYSINFO_SCHEMA_BULB(dev.sys_info)
assert dev.model is not None
@bulb_iot
async def test_light_state_without_update(dev: IotBulb, monkeypatch):
monkeypatch.setitem(dev._last_update["system"]["get_sysinfo"], "light_state", None)
with pytest.raises(KasaException):
print(dev.light_state)
@bulb_iot
async def test_get_light_state(dev: IotBulb):
LIGHT_STATE_SCHEMA(await dev.get_light_state())
@color_bulb_iot
async def test_set_hsv_transition(dev: IotBulb, mocker):
set_light_state = mocker.patch("kasa.iot.IotBulb._set_light_state")
light = dev.modules.get(Module.Light)
assert light
await light.set_hsv(10, 10, 100, transition=1000)
set_light_state.assert_called_with(
{"hue": 10, "saturation": 10, "brightness": 100, "color_temp": 0},
transition=1000,
)
@bulb_iot
async def test_light_set_state(dev: IotBulb, mocker):
"""Testing setting LightState on the light module."""
light = dev.modules.get(Module.Light)
assert light
set_light_state = mocker.spy(dev, "_set_light_state")
state = LightState(light_on=True)
await light.set_state(state)
set_light_state.assert_called_with({"on_off": 1}, transition=None)
state = LightState(light_on=False)
await light.set_state(state)
set_light_state.assert_called_with({"on_off": 0}, transition=None)
@variable_temp_iot
async def test_set_color_temp_transition(dev: IotBulb, mocker):
set_light_state = mocker.patch("kasa.iot.IotBulb._set_light_state")
light = dev.modules.get(Module.Light)
assert light
await light.set_color_temp(2700, transition=100)
set_light_state.assert_called_with({"color_temp": 2700}, transition=100)
@variable_temp_iot
@pytest.mark.xdist_group(name="caplog")
async def test_unknown_temp_range(dev: IotBulb, monkeypatch, caplog):
monkeypatch.setitem(dev._sys_info, "model", "unknown bulb")
light = dev.modules.get(Module.Light)
assert light
assert light.valid_temperature_range == (2700, 5000)
assert "Unknown color temperature range, fallback to 2700-5000" in caplog.text
@dimmable_iot
@turn_on
async def test_dimmable_brightness(dev: IotBulb, turn_on):
assert isinstance(dev, IotBulb | IotDimmer)
light = dev.modules.get(Module.Light)
assert light
await handle_turn_on(dev, turn_on)
assert dev._is_dimmable
await light.set_brightness(50)
await dev.update()
assert light.brightness == 50
await light.set_brightness(10)
await dev.update()
assert light.brightness == 10
with pytest.raises(TypeError, match="Brightness must be an integer"):
await light.set_brightness("foo") # type: ignore[arg-type]
@bulb_iot
async def test_turn_on_transition(dev: IotBulb, mocker):
set_light_state = mocker.patch("kasa.iot.IotBulb._set_light_state")
await dev.turn_on(transition=1000)
set_light_state.assert_called_with({"on_off": 1}, transition=1000)
await dev.turn_off(transition=100)
set_light_state.assert_called_with({"on_off": 0}, transition=100)
@bulb_iot
async def test_dimmable_brightness_transition(dev: IotBulb, mocker):
set_light_state = mocker.patch("kasa.iot.IotBulb._set_light_state")
light = dev.modules.get(Module.Light)
assert light
await light.set_brightness(10, transition=1000)
set_light_state.assert_called_with({"brightness": 10, "on_off": 1}, transition=1000)
@dimmable_iot
async def test_invalid_brightness(dev: IotBulb):
assert dev._is_dimmable
light = dev.modules.get(Module.Light)
assert light
with pytest.raises(
ValueError,
match=re.escape("Invalid brightness value: 110 (valid range: 0-100%)"),
):
await light.set_brightness(110)
with pytest.raises(
ValueError,
match=re.escape("Invalid brightness value: -100 (valid range: 0-100%)"),
):
await light.set_brightness(-100)
@non_dimmable_iot
async def test_non_dimmable(dev: IotBulb):
assert not dev._is_dimmable
light = dev.modules.get(Module.Light)
assert light
with pytest.raises(KasaException):
assert light.brightness == 0
with pytest.raises(KasaException):
await light.set_brightness(100)
@bulb_iot
async def test_ignore_default_not_set_without_color_mode_change_turn_on(
dev: IotBulb, mocker
):
query_helper = mocker.patch("kasa.iot.IotBulb._query_helper")
# When turning back without settings, ignore default to restore the state
await dev.turn_on()
args, kwargs = query_helper.call_args_list[0]
assert args[2] == {"on_off": 1, "ignore_default": 0}
await dev.turn_off()
args, kwargs = query_helper.call_args_list[1]
assert args[2] == {"on_off": 0, "ignore_default": 1}
@bulb_iot
async def test_list_presets(dev: IotBulb):
light_preset = dev.modules.get(Module.LightPreset)
assert light_preset
assert isinstance(light_preset, IotLightPresetModule)
presets = light_preset._deprecated_presets
# Light strip devices may list some light effects along with normal presets but these
# are handled by the LightEffect module so exclude preferred states with id
raw_presets = [
pstate for pstate in dev.sys_info["preferred_state"] if "id" not in pstate
]
assert len(presets) == len(raw_presets)
for preset, raw in zip(presets, raw_presets, strict=False):
assert preset.index == raw["index"]
assert preset.brightness == raw["brightness"]
assert preset.hue == raw["hue"]
assert preset.saturation == raw["saturation"]
assert preset.color_temp == raw["color_temp"]
@bulb_iot
async def test_modify_preset(dev: IotBulb, mocker):
"""Verify that modifying preset calls the and exceptions are raised properly."""
if (
not (light_preset := dev.modules.get(Module.LightPreset))
or not light_preset._deprecated_presets
):
pytest.skip("Some strips do not support presets")
assert isinstance(light_preset, IotLightPresetModule)
data: dict[str, int | None] = {
"index": 0,
"brightness": 10,
"hue": 0,
"saturation": 0,
"color_temp": 0,
}
preset = IotLightPreset(**data) # type: ignore[call-arg, arg-type]
assert preset.index == 0
assert preset.brightness == 10
assert preset.hue == 0
assert preset.saturation == 0
assert preset.color_temp == 0
await light_preset._deprecated_save_preset(preset)
await dev.update()
assert light_preset._deprecated_presets[0].brightness == 10
with pytest.raises(KasaException):
await light_preset._deprecated_save_preset(
IotLightPreset(index=5, hue=0, brightness=0, saturation=0, color_temp=0) # type: ignore[call-arg]
)
@bulb_iot
@pytest.mark.parametrize(
("preset", "payload"),
[
(
IotLightPreset(index=0, hue=0, brightness=1, saturation=0), # type: ignore[call-arg]
{"index": 0, "hue": 0, "brightness": 1, "saturation": 0},
),
(
IotLightPreset(index=0, brightness=1, id="testid", mode=2, custom=0), # type: ignore[call-arg]
{"index": 0, "brightness": 1, "id": "testid", "mode": 2, "custom": 0},
),
],
)
async def test_modify_preset_payloads(dev: IotBulb, preset, payload, mocker):
"""Test that modify preset payloads ignore none values."""
if (
not (light_preset := dev.modules.get(Module.LightPreset))
or not light_preset._deprecated_presets
):
pytest.skip("Some strips do not support presets")
query_helper = mocker.patch("kasa.iot.IotBulb._query_helper")
await light_preset._deprecated_save_preset(preset)
query_helper.assert_called_with(dev.LIGHT_SERVICE, "set_preferred_state", payload)
LIGHT_STATE_SCHEMA = Schema(
{
"brightness": All(int, Range(min=0, max=100)),
"color_temp": int,
"hue": All(int, Range(min=0, max=360)),
"mode": str,
"on_off": Boolean,
"saturation": All(int, Range(min=0, max=100)),
"length": Optional(int),
"transition": Optional(int),
"dft_on_state": Optional(
{
"brightness": All(int, Range(min=0, max=100)),
"color_temp": All(int, Range(min=0, max=9000)),
"hue": All(int, Range(min=0, max=360)),
"mode": str,
"saturation": All(int, Range(min=0, max=100)),
"groups": Optional(list[int]),
}
),
"err_code": int,
}
)
SYSINFO_SCHEMA_BULB = SYSINFO_SCHEMA.extend(
{
"ctrl_protocols": Optional(dict),
"description": Optional(str), # Seen on LBxxx, similar to dev_name
"dev_state": str,
"disco_ver": str,
"heapsize": int,
"is_color": Boolean,
"is_dimmable": Boolean,
"is_factory": Boolean,
"is_variable_color_temp": Boolean,
"light_state": LIGHT_STATE_SCHEMA,
"preferred_state": [
{
"brightness": All(int, Range(min=0, max=100)),
"color_temp": int,
"hue": All(int, Range(min=0, max=360)),
"index": int,
"saturation": All(int, Range(min=0, max=100)),
}
],
}
)
@bulb_iot
async def test_turn_on_behaviours(dev: IotBulb):
behavior = await dev.get_turn_on_behavior()
assert behavior

7
tests/test_iotdevice.py → tests/iot/test_iotdevice.py
View File

@@ -19,10 +19,9 @@ from voluptuous import (
from kasa import DeviceType, KasaException, Module
from kasa.iot import IotDevice
from kasa.iot.iotmodule import _merge_dict
from .conftest import get_device_for_fixture_protocol, handle_turn_on, turn_on
from .device_fixtures import device_iot, has_emeter_iot, no_emeter_iot
from .fakeprotocol_iot import FakeIotProtocol
from tests.conftest import get_device_for_fixture_protocol, handle_turn_on, turn_on
from tests.device_fixtures import device_iot, has_emeter_iot, no_emeter_iot
from tests.fakeprotocol_iot import FakeIotProtocol
TZ_SCHEMA = Schema(
{"zone_str": str, "dst_offset": int, "index": All(int, Range(min=0)), "tz_str": str}

3
tests/test_dimmer.py → tests/iot/test_iotdimmer.py
View File

@@ -2,8 +2,7 @@ import pytest
from kasa import DeviceType, Module
from kasa.iot import IotDimmer
from .conftest import dimmer_iot, handle_turn_on, turn_on
from tests.conftest import dimmer_iot, handle_turn_on, turn_on
@dimmer_iot

3
tests/test_lightstrip.py → tests/iot/test_iotlightstrip.py
View File

@@ -3,8 +3,7 @@ import pytest
from kasa import DeviceType, Module
from kasa.iot import IotLightStrip
from kasa.iot.modules import LightEffect
from .conftest import lightstrip_iot
from tests.conftest import lightstrip_iot
@lightstrip_iot

0
tests/smartcamera/__init__.py → tests/protocols/__init__.py
View File

4
tests/test_protocol.py → tests/protocols/test_iotprotocol.py
View File

@@ -29,8 +29,8 @@ from kasa.transports.basetransport import BaseTransport
from kasa.transports.klaptransport import KlapTransport, KlapTransportV2
from kasa.transports.xortransport import XorEncryption, XorTransport
from .conftest import device_iot
from .fakeprotocol_iot import FakeIotTransport
from ..conftest import device_iot
from ..fakeprotocol_iot import FakeIotTransport
@pytest.mark.parametrize(

4
tests/test_smartprotocol.py → tests/protocols/test_smartprotocol.py
View File

@@ -12,8 +12,8 @@ from kasa.exceptions import (
from kasa.protocols.smartprotocol import SmartProtocol, _ChildProtocolWrapper
from kasa.smart import SmartDevice
from .conftest import device_smart
from .fakeprotocol_smart import FakeSmartTransport
from ..conftest import device_smart
from ..fakeprotocol_smart import FakeSmartTransport
DUMMY_QUERY = {"foobar": {"foo": "bar", "bar": "foo"}}
DUMMY_MULTIPLE_QUERY = {

5
tests/smart/modules/test_autooff.py
View File

@@ -1,6 +1,5 @@
from __future__ import annotations
import sys
from datetime import datetime
import pytest
@@ -25,10 +24,6 @@ autooff = parametrize(
("auto_off_at", "auto_off_at", datetime | None),
],
)
@pytest.mark.skipif(
sys.version_info < (3, 10),
reason="Subscripted generics cannot be used with class and instance checks",
)
async def test_autooff_features(
dev: SmartDevice, feature: str, prop_name: str, type: type
):

21
tests/smart/modules/test_energy.py Normal file
View File

@@ -0,0 +1,21 @@
import pytest
from kasa import Module, SmartDevice
from kasa.interfaces.energy import Energy
from kasa.smart.modules import Energy as SmartEnergyModule
from tests.conftest import has_emeter_smart
@has_emeter_smart
async def test_supported(dev: SmartDevice):
energy_module = dev.modules.get(Module.Energy)
if not energy_module:
pytest.skip(f"Energy module not supported for {dev}.")
assert isinstance(energy_module, SmartEnergyModule)
assert energy_module.supports(Energy.ModuleFeature.CONSUMPTION_TOTAL) is False
assert energy_module.supports(Energy.ModuleFeature.PERIODIC_STATS) is False
if energy_module.supported_version < 2:
assert energy_module.supports(Energy.ModuleFeature.VOLTAGE_CURRENT) is False
else:
assert energy_module.supports(Energy.ModuleFeature.VOLTAGE_CURRENT) is True

99
tests/test_smartdevice.py → tests/smart/test_smartdevice.py
View File

@@ -2,6 +2,7 @@
from __future__ import annotations
import copy
import logging
import time
from typing import Any, cast
@@ -11,18 +12,20 @@ import pytest
from freezegun.api import FrozenDateTimeFactory
from pytest_mock import MockerFixture
from kasa import Device, KasaException, Module
from kasa import Device, DeviceType, KasaException, Module
from kasa.exceptions import DeviceError, SmartErrorCode
from kasa.protocols.smartprotocol import _ChildProtocolWrapper
from kasa.smart import SmartDevice
from kasa.smart.modules.energy import Energy
from kasa.smart.smartmodule import SmartModule
from .conftest import (
from tests.conftest import (
DISCOVERY_MOCK_IP,
device_smart,
get_device_for_fixture_protocol,
get_parent_and_child_modules,
smart_discovery,
)
from tests.device_fixtures import variable_temp_smart
@device_smart
@@ -51,6 +54,31 @@ async def test_update_no_device_info(dev: SmartDevice, mocker: MockerFixture):
await dev.update()
@smart_discovery
async def test_device_type_no_update(discovery_mock, caplog: pytest.LogCaptureFixture):
"""Test device type and repr when device not updated."""
dev = SmartDevice(DISCOVERY_MOCK_IP)
assert dev.device_type is DeviceType.Unknown
assert repr(dev) == f"<DeviceType.Unknown at {DISCOVERY_MOCK_IP} - update() needed>"
discovery_result = copy.deepcopy(discovery_mock.discovery_data["result"])
dev.update_from_discover_info(discovery_result)
assert dev.device_type is DeviceType.Unknown
assert (
repr(dev)
== f"<DeviceType.Unknown at {DISCOVERY_MOCK_IP} - None (None) - update() needed>"
)
discovery_result["device_type"] = "SMART.FOOBAR"
dev.update_from_discover_info(discovery_result)
dev._components = {"dummy": 1}
assert dev.device_type is DeviceType.Plug
assert (
repr(dev)
== f"<DeviceType.Plug at {DISCOVERY_MOCK_IP} - None (None) - update() needed>"
)
assert "Unknown device type, falling back to plug" in caplog.text
@device_smart
async def test_initial_update(dev: SmartDevice, mocker: MockerFixture):
"""Test the initial update cycle."""
@@ -435,3 +463,68 @@ async def test_smartdevice_cloud_connection(dev: SmartDevice, mocker: MockerFixt
):
await new_dev.update()
assert new_dev.is_cloud_connected is False
@variable_temp_smart
async def test_smart_temp_range(dev: Device):
light = dev.modules.get(Module.Light)
assert light
assert light.valid_temperature_range
@device_smart
async def test_initialize_modules_sysinfo_lookup_keys(
dev: SmartDevice, mocker: MockerFixture
):
"""Test that matching modules using SYSINFO_LOOKUP_KEYS are initialized correctly."""
class AvailableKey(SmartModule):
SYSINFO_LOOKUP_KEYS = ["device_id"]
class NonExistingKey(SmartModule):
SYSINFO_LOOKUP_KEYS = ["this_does_not_exist"]
# The __init_subclass__ hook in smartmodule checks the path,
# so we have to manually add these for testing.
mocker.patch.dict(
"kasa.smart.smartmodule.SmartModule.REGISTERED_MODULES",
{
AvailableKey._module_name(): AvailableKey,
NonExistingKey._module_name(): NonExistingKey,
},
)
# We have an already initialized device, so we try to initialize the modules again
await dev._initialize_modules()
assert "AvailableKey" in dev.modules
assert "NonExistingKey" not in dev.modules
@device_smart
async def test_initialize_modules_required_component(
dev: SmartDevice, mocker: MockerFixture
):
"""Test that matching modules using REQUIRED_COMPONENT are initialized correctly."""
class AvailableComponent(SmartModule):
REQUIRED_COMPONENT = "device"
class NonExistingComponent(SmartModule):
REQUIRED_COMPONENT = "this_does_not_exist"
# The __init_subclass__ hook in smartmodule checks the path,
# so we have to manually add these for testing.
mocker.patch.dict(
"kasa.smart.smartmodule.SmartModule.REGISTERED_MODULES",
{
AvailableComponent._module_name(): AvailableComponent,
NonExistingComponent._module_name(): NonExistingComponent,
},
)
# We have an already initialized device, so we try to initialize the modules again
await dev._initialize_modules()
assert "AvailableComponent" in dev.modules
assert "NonExistingComponent" not in dev.modules

0
tests/smartcamera/modules/__init__.py → tests/smartcam/__init__.py
View File

0
tests/smartcam/modules/__init__.py Normal file
View File

0
tests/smartcamera/modules/test_alarm.py → tests/smartcam/modules/test_alarm.py
View File

67
tests/smartcamera/test_smartcamera.py → tests/smartcam/modules/test_camera.py
View File

@@ -4,15 +4,13 @@ from __future__ import annotations
import base64
import json
from datetime import UTC, datetime
from unittest.mock import patch
import pytest
from freezegun.api import FrozenDateTimeFactory
from kasa import Credentials, Device, DeviceType, Module
from kasa import Credentials, Device, DeviceType, Module, StreamResolution
from ..conftest import camera_smartcam, device_smartcam, hub_smartcam
from ...conftest import camera_smartcam, device_smartcam
@device_smartcam
@@ -37,6 +35,16 @@ async def test_stream_rtsp_url(dev: Device):
url = camera_module.stream_rtsp_url(Credentials("foo", "bar"))
assert url == "rtsp://foo:bar@127.0.0.123:554/stream1"
url = camera_module.stream_rtsp_url(
Credentials("foo", "bar"), stream_resolution=StreamResolution.HD
)
assert url == "rtsp://foo:bar@127.0.0.123:554/stream1"
url = camera_module.stream_rtsp_url(
Credentials("foo", "bar"), stream_resolution=StreamResolution.SD
)
assert url == "rtsp://foo:bar@127.0.0.123:554/stream2"
with patch.object(dev.config, "credentials", Credentials("bar", "foo")):
url = camera_module.stream_rtsp_url()
assert url == "rtsp://bar:foo@127.0.0.123:554/stream1"
@@ -75,49 +83,12 @@ async def test_stream_rtsp_url(dev: Device):
url = camera_module.stream_rtsp_url()
assert url is None
# Test with camera off
await camera_module.set_state(False)
await dev.update()
url = camera_module.stream_rtsp_url(Credentials("foo", "bar"))
assert url is None
with patch.object(dev.config, "credentials", Credentials("bar", "foo")):
url = camera_module.stream_rtsp_url()
assert url is None
@camera_smartcam
async def test_onvif_url(dev: Device):
"""Test the onvif url."""
camera_module = dev.modules.get(Module.Camera)
assert camera_module
@device_smartcam
async def test_alias(dev):
test_alias = "TEST1234"
original = dev.alias
assert isinstance(original, str)
await dev.set_alias(test_alias)
await dev.update()
assert dev.alias == test_alias
await dev.set_alias(original)
await dev.update()
assert dev.alias == original
@hub_smartcam
async def test_hub(dev):
assert dev.children
for child in dev.children:
assert "Cloud" in child.modules
assert child.modules["Cloud"].data
assert child.alias
await child.update()
assert "Time" not in child.modules
assert child.time
@device_smartcam
async def test_device_time(dev: Device, freezer: FrozenDateTimeFactory):
"""Test a child device gets the time from it's parent module."""
fallback_time = datetime.now(UTC).astimezone().replace(microsecond=0)
assert dev.time != fallback_time
module = dev.modules[Module.Time]
await module.set_time(fallback_time)
await dev.update()
assert dev.time == fallback_time
url = camera_module.onvif_url()
assert url == "http://127.0.0.123:2020/onvif/device_service"

61
tests/smartcam/test_smartcamdevice.py Normal file
View File

@@ -0,0 +1,61 @@
"""Tests for smart camera devices."""
from __future__ import annotations
from datetime import UTC, datetime
import pytest
from freezegun.api import FrozenDateTimeFactory
from kasa import Device, DeviceType, Module
from ..conftest import device_smartcam, hub_smartcam
@device_smartcam
async def test_state(dev: Device):
if dev.device_type is DeviceType.Hub:
pytest.skip("Hubs cannot be switched on and off")
state = dev.is_on
await dev.set_state(not state)
await dev.update()
assert dev.is_on is not state
@device_smartcam
async def test_alias(dev):
test_alias = "TEST1234"
original = dev.alias
assert isinstance(original, str)
await dev.set_alias(test_alias)
await dev.update()
assert dev.alias == test_alias
await dev.set_alias(original)
await dev.update()
assert dev.alias == original
@hub_smartcam
async def test_hub(dev):
assert dev.children
for child in dev.children:
assert "Cloud" in child.modules
assert child.modules["Cloud"].data
assert child.alias
await child.update()
assert "Time" not in child.modules
assert child.time
@device_smartcam
async def test_device_time(dev: Device, freezer: FrozenDateTimeFactory):
"""Test a child device gets the time from it's parent module."""
fallback_time = datetime.now(UTC).astimezone().replace(microsecond=0)
assert dev.time != fallback_time
module = dev.modules[Module.Time]
await module.set_time(fallback_time)
await dev.update()
assert dev.time == fallback_time

327
tests/test_bulb.py
View File

@@ -1,44 +1,16 @@
from __future__ import annotations
import re
import pytest
from voluptuous import (
All,
Boolean,
Optional,
Range,
Schema,
)
from kasa import Device, DeviceType, IotLightPreset, KasaException, LightState, Module
from kasa.iot import IotBulb, IotDimmer
from kasa.iot.modules import LightPreset as IotLightPresetModule
from .conftest import (
from kasa import Device, DeviceType, KasaException, Module
from tests.conftest import handle_turn_on, turn_on
from tests.device_fixtures import (
bulb,
bulb_iot,
color_bulb,
color_bulb_iot,
dimmable_iot,
handle_turn_on,
non_color_bulb,
non_dimmable_iot,
non_variable_temp,
turn_on,
variable_temp,
variable_temp_iot,
variable_temp_smart,
)
from .test_iotdevice import SYSINFO_SCHEMA
@bulb_iot
async def test_bulb_sysinfo(dev: Device):
assert dev.sys_info is not None
SYSINFO_SCHEMA_BULB(dev.sys_info)
assert dev.model is not None
@bulb
@@ -47,18 +19,6 @@ async def test_state_attributes(dev: Device):
assert isinstance(dev.state_information["Cloud connection"], bool)
@bulb_iot
async def test_light_state_without_update(dev: IotBulb, monkeypatch):
monkeypatch.setitem(dev._last_update["system"]["get_sysinfo"], "light_state", None)
with pytest.raises(KasaException):
print(dev.light_state)
@bulb_iot
async def test_get_light_state(dev: IotBulb):
LIGHT_STATE_SCHEMA(await dev.get_light_state())
@color_bulb
@turn_on
async def test_hsv(dev: Device, turn_on):
@@ -81,35 +41,6 @@ async def test_hsv(dev: Device, turn_on):
assert brightness == 1
@color_bulb_iot
async def test_set_hsv_transition(dev: IotBulb, mocker):
set_light_state = mocker.patch("kasa.iot.IotBulb._set_light_state")
light = dev.modules.get(Module.Light)
assert light
await light.set_hsv(10, 10, 100, transition=1000)
set_light_state.assert_called_with(
{"hue": 10, "saturation": 10, "brightness": 100, "color_temp": 0},
transition=1000,
)
@bulb_iot
async def test_light_set_state(dev: IotBulb, mocker):
"""Testing setting LightState on the light module."""
light = dev.modules.get(Module.Light)
assert light
set_light_state = mocker.spy(dev, "_set_light_state")
state = LightState(light_on=True)
await light.set_state(state)
set_light_state.assert_called_with({"on_off": 1}, transition=None)
state = LightState(light_on=False)
await light.set_state(state)
set_light_state.assert_called_with({"on_off": 0}, transition=None)
@color_bulb
@turn_on
@pytest.mark.parametrize(
@@ -221,33 +152,6 @@ async def test_try_set_colortemp(dev: Device, turn_on):
assert light.color_temp == 2700
@variable_temp_iot
async def test_set_color_temp_transition(dev: IotBulb, mocker):
set_light_state = mocker.patch("kasa.iot.IotBulb._set_light_state")
light = dev.modules.get(Module.Light)
assert light
await light.set_color_temp(2700, transition=100)
set_light_state.assert_called_with({"color_temp": 2700}, transition=100)
@variable_temp_iot
@pytest.mark.xdist_group(name="caplog")
async def test_unknown_temp_range(dev: IotBulb, monkeypatch, caplog):
monkeypatch.setitem(dev._sys_info, "model", "unknown bulb")
light = dev.modules.get(Module.Light)
assert light
assert light.valid_temperature_range == (2700, 5000)
assert "Unknown color temperature range, fallback to 2700-5000" in caplog.text
@variable_temp_smart
async def test_smart_temp_range(dev: Device):
light = dev.modules.get(Module.Light)
assert light
assert light.valid_temperature_range
@variable_temp
async def test_out_of_range_temperature(dev: Device):
light = dev.modules.get(Module.Light)
@@ -276,231 +180,6 @@ async def test_non_variable_temp(dev: Device):
print(light.color_temp)
@dimmable_iot
@turn_on
async def test_dimmable_brightness(dev: IotBulb, turn_on):
assert isinstance(dev, IotBulb | IotDimmer)
light = dev.modules.get(Module.Light)
assert light
await handle_turn_on(dev, turn_on)
assert dev._is_dimmable
await light.set_brightness(50)
await dev.update()
assert light.brightness == 50
await light.set_brightness(10)
await dev.update()
assert light.brightness == 10
with pytest.raises(TypeError, match="Brightness must be an integer"):
await light.set_brightness("foo") # type: ignore[arg-type]
@bulb_iot
async def test_turn_on_transition(dev: IotBulb, mocker):
set_light_state = mocker.patch("kasa.iot.IotBulb._set_light_state")
await dev.turn_on(transition=1000)
set_light_state.assert_called_with({"on_off": 1}, transition=1000)
await dev.turn_off(transition=100)
set_light_state.assert_called_with({"on_off": 0}, transition=100)
@bulb_iot
async def test_dimmable_brightness_transition(dev: IotBulb, mocker):
set_light_state = mocker.patch("kasa.iot.IotBulb._set_light_state")
light = dev.modules.get(Module.Light)
assert light
await light.set_brightness(10, transition=1000)
set_light_state.assert_called_with({"brightness": 10, "on_off": 1}, transition=1000)
@dimmable_iot
async def test_invalid_brightness(dev: IotBulb):
assert dev._is_dimmable
light = dev.modules.get(Module.Light)
assert light
with pytest.raises(
ValueError,
match=re.escape("Invalid brightness value: 110 (valid range: 0-100%)"),
):
await light.set_brightness(110)
with pytest.raises(
ValueError,
match=re.escape("Invalid brightness value: -100 (valid range: 0-100%)"),
):
await light.set_brightness(-100)
@non_dimmable_iot
async def test_non_dimmable(dev: IotBulb):
assert not dev._is_dimmable
light = dev.modules.get(Module.Light)
assert light
with pytest.raises(KasaException):
assert light.brightness == 0
with pytest.raises(KasaException):
await light.set_brightness(100)
@bulb_iot
async def test_ignore_default_not_set_without_color_mode_change_turn_on(
dev: IotBulb, mocker
):
query_helper = mocker.patch("kasa.iot.IotBulb._query_helper")
# When turning back without settings, ignore default to restore the state
await dev.turn_on()
args, kwargs = query_helper.call_args_list[0]
assert args[2] == {"on_off": 1, "ignore_default": 0}
await dev.turn_off()
args, kwargs = query_helper.call_args_list[1]
assert args[2] == {"on_off": 0, "ignore_default": 1}
@bulb_iot
async def test_list_presets(dev: IotBulb):
light_preset = dev.modules.get(Module.LightPreset)
assert light_preset
assert isinstance(light_preset, IotLightPresetModule)
presets = light_preset._deprecated_presets
# Light strip devices may list some light effects along with normal presets but these
# are handled by the LightEffect module so exclude preferred states with id
raw_presets = [
pstate for pstate in dev.sys_info["preferred_state"] if "id" not in pstate
]
assert len(presets) == len(raw_presets)
for preset, raw in zip(presets, raw_presets, strict=False):
assert preset.index == raw["index"]
assert preset.brightness == raw["brightness"]
assert preset.hue == raw["hue"]
assert preset.saturation == raw["saturation"]
assert preset.color_temp == raw["color_temp"]
@bulb_iot
async def test_modify_preset(dev: IotBulb, mocker):
"""Verify that modifying preset calls the and exceptions are raised properly."""
if (
not (light_preset := dev.modules.get(Module.LightPreset))
or not light_preset._deprecated_presets
):
pytest.skip("Some strips do not support presets")
assert isinstance(light_preset, IotLightPresetModule)
data: dict[str, int | None] = {
"index": 0,
"brightness": 10,
"hue": 0,
"saturation": 0,
"color_temp": 0,
}
preset = IotLightPreset(**data) # type: ignore[call-arg, arg-type]
assert preset.index == 0
assert preset.brightness == 10
assert preset.hue == 0
assert preset.saturation == 0
assert preset.color_temp == 0
await light_preset._deprecated_save_preset(preset)
await dev.update()
assert light_preset._deprecated_presets[0].brightness == 10
with pytest.raises(KasaException):
await light_preset._deprecated_save_preset(
IotLightPreset(index=5, hue=0, brightness=0, saturation=0, color_temp=0) # type: ignore[call-arg]
)
@bulb_iot
@pytest.mark.parametrize(
("preset", "payload"),
[
(
IotLightPreset(index=0, hue=0, brightness=1, saturation=0), # type: ignore[call-arg]
{"index": 0, "hue": 0, "brightness": 1, "saturation": 0},
),
(
IotLightPreset(index=0, brightness=1, id="testid", mode=2, custom=0), # type: ignore[call-arg]
{"index": 0, "brightness": 1, "id": "testid", "mode": 2, "custom": 0},
),
],
)
async def test_modify_preset_payloads(dev: IotBulb, preset, payload, mocker):
"""Test that modify preset payloads ignore none values."""
if (
not (light_preset := dev.modules.get(Module.LightPreset))
or not light_preset._deprecated_presets
):
pytest.skip("Some strips do not support presets")
query_helper = mocker.patch("kasa.iot.IotBulb._query_helper")
await light_preset._deprecated_save_preset(preset)
query_helper.assert_called_with(dev.LIGHT_SERVICE, "set_preferred_state", payload)
LIGHT_STATE_SCHEMA = Schema(
{
"brightness": All(int, Range(min=0, max=100)),
"color_temp": int,
"hue": All(int, Range(min=0, max=360)),
"mode": str,
"on_off": Boolean,
"saturation": All(int, Range(min=0, max=100)),
"length": Optional(int),
"transition": Optional(int),
"dft_on_state": Optional(
{
"brightness": All(int, Range(min=0, max=100)),
"color_temp": All(int, Range(min=0, max=9000)),
"hue": All(int, Range(min=0, max=360)),
"mode": str,
"saturation": All(int, Range(min=0, max=100)),
"groups": Optional(list[int]),
}
),
"err_code": int,
}
)
SYSINFO_SCHEMA_BULB = SYSINFO_SCHEMA.extend(
{
"ctrl_protocols": Optional(dict),
"description": Optional(str), # Seen on LBxxx, similar to dev_name
"dev_state": str,
"disco_ver": str,
"heapsize": int,
"is_color": Boolean,
"is_dimmable": Boolean,
"is_factory": Boolean,
"is_variable_color_temp": Boolean,
"light_state": LIGHT_STATE_SCHEMA,
"preferred_state": [
{
"brightness": All(int, Range(min=0, max=100)),
"color_temp": int,
"hue": All(int, Range(min=0, max=360)),
"index": int,
"saturation": All(int, Range(min=0, max=100)),
}
],
}
)
@bulb
def test_device_type_bulb(dev: Device):
assert dev.device_type in {DeviceType.Bulb, DeviceType.LightStrip}
@bulb_iot
async def test_turn_on_behaviours(dev: IotBulb):
behavior = await dev.get_turn_on_behavior()
assert behavior

77
tests/test_cli.py
View File

@@ -2,7 +2,7 @@ import json
import os
import re
from datetime import datetime
from unittest.mock import ANY
from unittest.mock import ANY, PropertyMock, patch
from zoneinfo import ZoneInfo
import asyncclick as click
@@ -40,10 +40,11 @@ from kasa.cli.light import (
)
from kasa.cli.main import TYPES, _legacy_type_to_class, cli, cmd_command, raw_command
from kasa.cli.time import time
from kasa.cli.usage import emeter, energy
from kasa.cli.usage import energy
from kasa.cli.wifi import wifi
from kasa.discover import Discover, DiscoveryResult
from kasa.discover import Discover, DiscoveryResult, redact_data
from kasa.iot import IotDevice
from kasa.json import dumps as json_dumps
from kasa.smart import SmartDevice
from kasa.smartcam import SmartCamDevice
@@ -126,6 +127,36 @@ async def test_list_devices(discovery_mock, runner):
assert row in res.output
async def test_discover_raw(discovery_mock, runner, mocker):
"""Test the discover raw command."""
redact_spy = mocker.patch(
"kasa.protocols.protocol.redact_data", side_effect=redact_data
)
res = await runner.invoke(
cli,
["--username", "foo", "--password", "bar", "discover", "raw"],
catch_exceptions=False,
)
assert res.exit_code == 0
expected = {
"discovery_response": discovery_mock.discovery_data,
"meta": {"ip": "127.0.0.123", "port": discovery_mock.discovery_port},
}
assert res.output == json_dumps(expected, indent=True) + "\n"
redact_spy.assert_not_called()
res = await runner.invoke(
cli,
["--username", "foo", "--password", "bar", "discover", "raw", "--redact"],
catch_exceptions=False,
)
assert res.exit_code == 0
redact_spy.assert_called()
@new_discovery
async def test_list_auth_failed(discovery_mock, mocker, runner):
"""Test that device update is called on main."""
@@ -432,38 +463,45 @@ async def test_time_set(dev: Device, mocker, runner):
async def test_emeter(dev: Device, mocker, runner):
res = await runner.invoke(emeter, obj=dev)
mocker.patch("kasa.Discover.discover_single", return_value=dev)
base_cmd = ["--host", "dummy", "energy"]
res = await runner.invoke(cli, base_cmd, obj=dev)
if not (energy := dev.modules.get(Module.Energy)):
assert "Device has no energy module." in res.output
return
assert "== Emeter ==" in res.output
assert "== Energy ==" in res.output
if dev.device_type is not DeviceType.Strip:
res = await runner.invoke(emeter, ["--index", "0"], obj=dev)
res = await runner.invoke(cli, [*base_cmd, "--index", "0"], obj=dev)
assert f"Device: {dev.host} does not have children" in res.output
res = await runner.invoke(emeter, ["--name", "mock"], obj=dev)
res = await runner.invoke(cli, [*base_cmd, "--name", "mock"], obj=dev)
assert f"Device: {dev.host} does not have children" in res.output
if dev.device_type is DeviceType.Strip and len(dev.children) > 0:
child_energy = dev.children[0].modules.get(Module.Energy)
assert child_energy
realtime_emeter = mocker.patch.object(child_energy, "get_status")
realtime_emeter.return_value = EmeterStatus({"voltage_mv": 122066})
res = await runner.invoke(emeter, ["--index", "0"], obj=dev)
assert "Voltage: 122.066 V" in res.output
realtime_emeter.assert_called()
assert realtime_emeter.call_count == 1
with patch.object(
type(child_energy), "status", new_callable=PropertyMock
) as child_status:
child_status.return_value = EmeterStatus({"voltage_mv": 122066})
res = await runner.invoke(emeter, ["--name", dev.children[0].alias], obj=dev)
assert "Voltage: 122.066 V" in res.output
assert realtime_emeter.call_count == 2
res = await runner.invoke(cli, [*base_cmd, "--index", "0"], obj=dev)
assert "Voltage: 122.066 V" in res.output
child_status.assert_called()
assert child_status.call_count == 1
res = await runner.invoke(
cli, [*base_cmd, "--name", dev.children[0].alias], obj=dev
)
assert "Voltage: 122.066 V" in res.output
assert child_status.call_count == 2
if isinstance(dev, IotDevice):
monthly = mocker.patch.object(energy, "get_monthly_stats")
monthly.return_value = {1: 1234}
res = await runner.invoke(emeter, ["--year", "1900"], obj=dev)
res = await runner.invoke(cli, [*base_cmd, "--year", "1900"], obj=dev)
if not isinstance(dev, IotDevice):
assert "Device does not support historical statistics" in res.output
return
@@ -474,7 +512,7 @@ async def test_emeter(dev: Device, mocker, runner):
if isinstance(dev, IotDevice):
daily = mocker.patch.object(energy, "get_daily_stats")
daily.return_value = {1: 1234}
res = await runner.invoke(emeter, ["--month", "1900-12"], obj=dev)
res = await runner.invoke(cli, [*base_cmd, "--month", "1900-12"], obj=dev)
if not isinstance(dev, IotDevice):
assert "Device has no historical statistics" in res.output
return
@@ -685,6 +723,8 @@ async def test_credentials(discovery_mock, mocker, runner):
dr.device_type,
"--encrypt-type",
dr.mgt_encrypt_schm.encrypt_type,
"--login-version",
dr.mgt_encrypt_schm.lv or 1,
],
)
assert res.exit_code == 0
@@ -722,6 +762,7 @@ async def test_without_device_type(dev, mocker, runner):
timeout=5,
discovery_timeout=7,
on_unsupported=ANY,
on_discovered_raw=ANY,
)

41
tests/test_common_modules.py
View File

@@ -4,7 +4,7 @@ from zoneinfo import ZoneInfo
import pytest
from pytest_mock import MockerFixture
from kasa import Device, LightState, Module
from kasa import Device, LightState, Module, ThermostatState
from .device_fixtures import (
bulb_iot,
@@ -57,6 +57,12 @@ light_preset = parametrize_combine([light_preset_smart, bulb_iot])
light = parametrize_combine([bulb_smart, bulb_iot, dimmable])
temp_control_smart = parametrize(
"has temp control smart",
component_filter="temp_control",
protocol_filter={"SMART.CHILD"},
)
@led
async def test_led_module(dev: Device, mocker: MockerFixture):
@@ -325,6 +331,39 @@ async def test_light_preset_save(dev: Device, mocker: MockerFixture):
assert new_preset_state.color_temp == new_preset.color_temp
@temp_control_smart
async def test_thermostat(dev: Device, mocker: MockerFixture):
"""Test saving a new preset value."""
therm_mod = next(get_parent_and_child_modules(dev, Module.Thermostat))
assert therm_mod
await therm_mod.set_state(False)
await dev.update()
assert therm_mod.state is False
assert therm_mod.mode is ThermostatState.Off
await therm_mod.set_target_temperature(10)
await dev.update()
assert therm_mod.state is True
assert therm_mod.mode is ThermostatState.Heating
assert therm_mod.target_temperature == 10
target_temperature_feature = therm_mod.get_feature(therm_mod.set_target_temperature)
temp_control = dev.modules.get(Module.TemperatureControl)
assert temp_control
allowed_range = temp_control.allowed_temperature_range
assert target_temperature_feature.minimum_value == allowed_range[0]
assert target_temperature_feature.maximum_value == allowed_range[1]
await therm_mod.set_temperature_unit("celsius")
await dev.update()
assert therm_mod.temperature_unit == "celsius"
await therm_mod.set_temperature_unit("fahrenheit")
await dev.update()
assert therm_mod.temperature_unit == "fahrenheit"
async def test_set_time(dev: Device):
"""Test setting the device time."""
time_mod = dev.modules[Module.Time]

7
tests/test_device.py
View File

@@ -16,6 +16,7 @@ import kasa
from kasa import Credentials, Device, DeviceConfig, DeviceType, KasaException, Module
from kasa.iot import (
IotBulb,
IotCamera,
IotDevice,
IotDimmer,
IotLightStrip,
@@ -55,6 +56,11 @@ device_classes = pytest.mark.parametrize(
)
async def test_device_id(dev: Device):
"""Test all devices have a device id."""
assert dev.device_id
async def test_alias(dev):
test_alias = "TEST1234"
original = dev.alias
@@ -113,6 +119,7 @@ async def test_device_class_repr(device_class_name_obj):
IotStrip: DeviceType.Strip,
IotWallSwitch: DeviceType.WallSwitch,
IotLightStrip: DeviceType.LightStrip,
IotCamera: DeviceType.Camera,
SmartChildDevice: DeviceType.Unknown,
SmartDevice: DeviceType.Unknown,
SmartCamDevice: DeviceType.Camera,

5
tests/test_device_factory.py
View File

@@ -47,7 +47,10 @@ def _get_connection_type_device_class(discovery_info):
dr = DiscoveryResult.from_dict(discovery_info["result"])
connection_type = DeviceConnectionParameters.from_values(
dr.device_type, dr.mgt_encrypt_schm.encrypt_type
dr.device_type,
dr.mgt_encrypt_schm.encrypt_type,
dr.mgt_encrypt_schm.lv,
dr.mgt_encrypt_schm.is_support_https,
)
else:
connection_type = DeviceConnectionParameters.from_values(

2
tests/test_plug.py
View File

@@ -1,9 +1,9 @@
import pytest
from kasa import DeviceType
from tests.iot.test_iotdevice import SYSINFO_SCHEMA
from .conftest import plug, plug_iot, plug_smart, switch_smart, wallswitch_iot
from .test_iotdevice import SYSINFO_SCHEMA
# these schemas should go to the mainlib as
# they can be useful when adding support for new features/devices

0
tests/transports/__init__.py Normal file
View File

0
tests/test_aestransport.py → tests/transports/test_aestransport.py
View File

0
tests/test_klapprotocol.py → tests/transports/test_klaptransport.py
View File

144
tests/transports/test_linkietransport.py Normal file
View File

@@ -0,0 +1,144 @@
import base64
from unittest.mock import ANY
import aiohttp
import pytest
from yarl import URL
from kasa.credentials import DEFAULT_CREDENTIALS, Credentials, get_default_credentials
from kasa.deviceconfig import DeviceConfig
from kasa.exceptions import KasaException
from kasa.httpclient import HttpClient
from kasa.json import dumps as json_dumps
from kasa.transports.linkietransport import LinkieTransportV2
KASACAM_REQUEST_PLAINTEXT = '{"smartlife.cam.ipcamera.dateTime":{"get_status":{}}}'
KASACAM_RESPONSE_ENCRYPTED = "0PKG74LnnfKc+dvhw5bCgaycqZOjk7Gdv96syaiKsJLTvtupwKPC7aPGse632KrB48/tiPiX9JzDsNW2lK6fqZCgmKuZoZGh3A=="
KASACAM_RESPONSE_ERROR = '{"smartlife.cam.ipcamera.cloud": {"get_inf": {"err_code": -10008, "err_msg": "Unsupported API call."}}}'
KASA_DEFAULT_CREDENTIALS_HASH = "YWRtaW46MjEyMzJmMjk3YTU3YTVhNzQzODk0YTBlNGE4MDFmYzM="
async def test_working(mocker):
"""No errors with an expected request/response."""
host = "127.0.0.1"
mock_linkie_device = MockLinkieDevice(host)
mocker.patch.object(
aiohttp.ClientSession, "post", side_effect=mock_linkie_device.post
)
transport_no_creds = LinkieTransportV2(config=DeviceConfig(host))
response = await transport_no_creds.send(KASACAM_REQUEST_PLAINTEXT)
assert response == {
"timezone": "UTC-05:00",
"area": "America/New_York",
"epoch_sec": 1690832800,
}
async def test_credentials_hash(mocker):
"""Ensure the default credentials are always passed as Basic Auth."""
# Test without credentials input
host = "127.0.0.1"
mock_linkie_device = MockLinkieDevice(host)
mock_post = mocker.patch.object(
aiohttp.ClientSession, "post", side_effect=mock_linkie_device.post
)
transport_no_creds = LinkieTransportV2(config=DeviceConfig(host))
await transport_no_creds.send(KASACAM_REQUEST_PLAINTEXT)
mock_post.assert_called_once_with(
URL(f"https://{host}:10443/data/LINKIE2.json"),
params=None,
data=ANY,
json=None,
timeout=ANY,
cookies=None,
headers={
"Authorization": "Basic " + _generate_kascam_basic_auth(),
"Content-Type": "application/x-www-form-urlencoded",
},
ssl=ANY,
)
assert transport_no_creds.credentials_hash == KASA_DEFAULT_CREDENTIALS_HASH
# Test with credentials input
transport_with_creds = LinkieTransportV2(
config=DeviceConfig(host, credentials=Credentials("Admin", "password"))
)
mock_post.reset_mock()
await transport_with_creds.send(KASACAM_REQUEST_PLAINTEXT)
mock_post.assert_called_once_with(
URL(f"https://{host}:10443/data/LINKIE2.json"),
params=None,
data=ANY,
json=None,
timeout=ANY,
cookies=None,
headers={
"Authorization": "Basic " + _generate_kascam_basic_auth(),
"Content-Type": "application/x-www-form-urlencoded",
},
ssl=ANY,
)
@pytest.mark.parametrize(
("return_status", "return_data", "expected"),
[
(500, KASACAM_RESPONSE_ENCRYPTED, "500"),
(200, "AAAAAAAAAAAAAAAAAAAAAAAA", "Unable to read response"),
(200, KASACAM_RESPONSE_ERROR, "Unsupported API call"),
],
)
async def test_exceptions(mocker, return_status, return_data, expected):
"""Test a variety of possible responses from the device."""
host = "127.0.0.1"
transport = LinkieTransportV2(config=DeviceConfig(host))
mock_linkie_device = MockLinkieDevice(
host, status_code=return_status, response=return_data
)
mocker.patch.object(
aiohttp.ClientSession, "post", side_effect=mock_linkie_device.post
)
with pytest.raises(KasaException, match=expected):
await transport.send(KASACAM_REQUEST_PLAINTEXT)
def _generate_kascam_basic_auth():
creds = get_default_credentials(DEFAULT_CREDENTIALS["KASACAMERA"])
creds_combined = f"{creds.username}:{creds.password}"
return base64.b64encode(creds_combined.encode()).decode()
class MockLinkieDevice:
"""Based on MockSslDevice."""
class _mock_response:
def __init__(self, status, request: dict):
self.status = status
self._json = request
async def __aenter__(self):
return self
async def __aexit__(self, exc_t, exc_v, exc_tb):
pass
async def read(self):
if isinstance(self._json, dict):
return json_dumps(self._json).encode()
return self._json
def __init__(self, host, *, status_code=200, response=KASACAM_RESPONSE_ENCRYPTED):
self.host = host
self.http_client = HttpClient(DeviceConfig(self.host))
self.status_code = status_code
self.response = response
async def post(
self, url: URL, *, headers=None, params=None, json=None, data=None, **__
):
return self._mock_response(self.status_code, self.response)

0
tests/test_sslaestransport.py → tests/transports/test_sslaestransport.py
View File

374
tests/transports/test_ssltransport.py Normal file
View File

@@ -0,0 +1,374 @@
from __future__ import annotations
import logging
from base64 import b64encode
from contextlib import nullcontext as does_not_raise
from typing import Any
import aiohttp
import pytest
from yarl import URL
from kasa.credentials import DEFAULT_CREDENTIALS, Credentials, get_default_credentials
from kasa.deviceconfig import DeviceConfig
from kasa.exceptions import (
AuthenticationError,
DeviceError,
KasaException,
SmartErrorCode,
_RetryableError,
)
from kasa.httpclient import HttpClient
from kasa.json import dumps as json_dumps
from kasa.json import loads as json_loads
from kasa.transports import SslTransport
from kasa.transports.ssltransport import TransportState, _md5_hash
# Transport tests are not designed for real devices
pytestmark = [pytest.mark.requires_dummy]
MOCK_PWD = "correct_pwd" # noqa: S105
MOCK_USER = "mock@example.com"
MOCK_BAD_USER_OR_PWD = "foobar" # noqa: S105
MOCK_TOKEN = "abcdefghijklmnopqrstuvwxyz1234)(" # noqa: S105
DEFAULT_CREDS = get_default_credentials(DEFAULT_CREDENTIALS["TAPO"])
_LOGGER = logging.getLogger(__name__)
@pytest.mark.parametrize(
(
"status_code",
"error_code",
"username",
"password",
"expectation",
),
[
pytest.param(
200,
SmartErrorCode.SUCCESS,
MOCK_USER,
MOCK_PWD,
does_not_raise(),
id="success",
),
pytest.param(
200,
SmartErrorCode.UNSPECIFIC_ERROR,
MOCK_USER,
MOCK_PWD,
pytest.raises(_RetryableError),
id="test retry",
),
pytest.param(
200,
SmartErrorCode.DEVICE_BLOCKED,
MOCK_USER,
MOCK_PWD,
pytest.raises(DeviceError),
id="test regular error",
),
pytest.param(
400,
SmartErrorCode.INTERNAL_UNKNOWN_ERROR,
MOCK_USER,
MOCK_PWD,
pytest.raises(KasaException),
id="400 error",
),
pytest.param(
200,
SmartErrorCode.LOGIN_ERROR,
MOCK_BAD_USER_OR_PWD,
MOCK_PWD,
pytest.raises(AuthenticationError),
id="bad-username",
),
pytest.param(
200,
[SmartErrorCode.LOGIN_ERROR, SmartErrorCode.SUCCESS],
MOCK_BAD_USER_OR_PWD,
"",
does_not_raise(),
id="working-fallback",
),
pytest.param(
200,
[SmartErrorCode.LOGIN_ERROR, SmartErrorCode.LOGIN_ERROR],
MOCK_BAD_USER_OR_PWD,
"",
pytest.raises(AuthenticationError),
id="fallback-fail",
),
pytest.param(
200,
SmartErrorCode.LOGIN_ERROR,
MOCK_USER,
MOCK_BAD_USER_OR_PWD,
pytest.raises(AuthenticationError),
id="bad-password",
),
pytest.param(
200,
SmartErrorCode.TRANSPORT_UNKNOWN_CREDENTIALS_ERROR,
MOCK_USER,
MOCK_PWD,
pytest.raises(AuthenticationError),
id="auth-error != login_error",
),
],
)
async def test_login(
mocker,
status_code,
error_code,
username,
password,
expectation,
):
host = "127.0.0.1"
mock_ssl_aes_device = MockSslDevice(
host,
status_code=status_code,
send_error_code=error_code,
)
mocker.patch.object(
aiohttp.ClientSession, "post", side_effect=mock_ssl_aes_device.post
)
transport = SslTransport(
config=DeviceConfig(host, credentials=Credentials(username, password))
)
assert transport._state is TransportState.LOGIN_REQUIRED
with expectation:
await transport.perform_login()
assert transport._state is TransportState.ESTABLISHED
await transport.close()
async def test_credentials_hash(mocker):
host = "127.0.0.1"
mock_ssl_aes_device = MockSslDevice(host)
mocker.patch.object(
aiohttp.ClientSession, "post", side_effect=mock_ssl_aes_device.post
)
creds = Credentials(MOCK_USER, MOCK_PWD)
data = {"password": _md5_hash(MOCK_PWD.encode()), "username": MOCK_USER}
creds_hash = b64encode(json_dumps(data).encode()).decode()
# Test with credentials input
transport = SslTransport(config=DeviceConfig(host, credentials=creds))
assert transport.credentials_hash == creds_hash
# Test with credentials_hash input
transport = SslTransport(config=DeviceConfig(host, credentials_hash=creds_hash))
assert transport.credentials_hash == creds_hash
await transport.close()
async def test_send(mocker):
host = "127.0.0.1"
mock_ssl_aes_device = MockSslDevice(host, send_error_code=SmartErrorCode.SUCCESS)
mocker.patch.object(
aiohttp.ClientSession, "post", side_effect=mock_ssl_aes_device.post
)
transport = SslTransport(
config=DeviceConfig(host, credentials=Credentials(MOCK_USER, MOCK_PWD))
)
try_login_spy = mocker.spy(transport, "try_login")
request = {
"method": "get_device_info",
"params": None,
}
assert transport._state is TransportState.LOGIN_REQUIRED
res = await transport.send(json_dumps(request))
assert "result" in res
try_login_spy.assert_called_once()
assert transport._state is TransportState.ESTABLISHED
# Second request does not
res = await transport.send(json_dumps(request))
try_login_spy.assert_called_once()
await transport.close()
async def test_no_credentials(mocker):
"""Test transport without credentials."""
host = "127.0.0.1"
mock_ssl_aes_device = MockSslDevice(
host, send_error_code=SmartErrorCode.LOGIN_ERROR
)
mocker.patch.object(
aiohttp.ClientSession, "post", side_effect=mock_ssl_aes_device.post
)
transport = SslTransport(config=DeviceConfig(host))
try_login_spy = mocker.spy(transport, "try_login")
with pytest.raises(AuthenticationError):
await transport.send('{"method": "dummy"}')
# We get called twice
assert try_login_spy.call_count == 2
await transport.close()
async def test_reset(mocker):
"""Test that transport state adjusts correctly for reset."""
host = "127.0.0.1"
mock_ssl_aes_device = MockSslDevice(host, send_error_code=SmartErrorCode.SUCCESS)
mocker.patch.object(
aiohttp.ClientSession, "post", side_effect=mock_ssl_aes_device.post
)
transport = SslTransport(
config=DeviceConfig(host, credentials=Credentials(MOCK_USER, MOCK_PWD))
)
assert transport._state is TransportState.LOGIN_REQUIRED
assert str(transport._app_url) == "https://127.0.0.1:4433/app"
await transport.perform_login()
assert transport._state is TransportState.ESTABLISHED
assert str(transport._app_url).startswith("https://127.0.0.1:4433/app?token=")
await transport.close()
assert transport._state is TransportState.LOGIN_REQUIRED
assert str(transport._app_url) == "https://127.0.0.1:4433/app"
async def test_port_override():
"""Test that port override sets the app_url."""
host = "127.0.0.1"
port_override = 12345
config = DeviceConfig(
host, credentials=Credentials("foo", "bar"), port_override=port_override
)
transport = SslTransport(config=config)
assert str(transport._app_url) == f"https://127.0.0.1:{port_override}/app"
await transport.close()
class MockSslDevice:
"""Based on MockAesSslDevice."""
class _mock_response:
def __init__(self, status, request: dict):
self.status = status
self._json = request
async def __aenter__(self):
return self
async def __aexit__(self, exc_t, exc_v, exc_tb):
pass
async def read(self):
if isinstance(self._json, dict):
return json_dumps(self._json).encode()
return self._json
def __init__(
self,
host,
*,
status_code=200,
send_error_code=SmartErrorCode.INTERNAL_UNKNOWN_ERROR,
):
self.host = host
self.http_client = HttpClient(DeviceConfig(self.host))
self._state = TransportState.LOGIN_REQUIRED
# test behaviour attributes
self.status_code = status_code
self.send_error_code = send_error_code
async def post(self, url: URL, params=None, json=None, data=None, *_, **__):
if data:
json = json_loads(data)
_LOGGER.debug("Request %s: %s", url, json)
res = self._post(url, json)
_LOGGER.debug("Response %s, data: %s", res, await res.read())
return res
def _post(self, url: URL, json: dict[str, Any]):
method = json["method"]
if method == "login":
if self._state is TransportState.LOGIN_REQUIRED:
assert json.get("token") is None
assert url == URL(f"https://{self.host}:4433/app")
return self._return_login_response(url, json)
else:
_LOGGER.warning("Received login although already logged in")
pytest.fail("non-handled re-login logic")
assert url == URL(f"https://{self.host}:4433/app?token={MOCK_TOKEN}")
return self._return_send_response(url, json)
def _return_login_response(self, url: URL, request: dict[str, Any]):
request_username = request["params"].get("username")
request_password = request["params"].get("password")
# Handle multiple error codes
if isinstance(self.send_error_code, list):
error_code = self.send_error_code.pop(0)
else:
error_code = self.send_error_code
_LOGGER.debug("Using error code %s", error_code)
def _return_login_error():
resp = {
"error_code": error_code.value,
"result": {"unknown": "payload"},
}
_LOGGER.debug("Returning login error with status %s", self.status_code)
return self._mock_response(self.status_code, resp)
if error_code is not SmartErrorCode.SUCCESS:
# Bad username
if request_username == MOCK_BAD_USER_OR_PWD:
return _return_login_error()
# Bad password
if request_password == _md5_hash(MOCK_BAD_USER_OR_PWD.encode()):
return _return_login_error()
# Empty password
if request_password == _md5_hash(b""):
return _return_login_error()
self._state = TransportState.ESTABLISHED
resp = {
"error_code": error_code.value,
"result": {
"token": MOCK_TOKEN,
},
}
_LOGGER.debug("Returning login success with status %s", self.status_code)
return self._mock_response(self.status_code, resp)
def _return_send_response(self, url: URL, json: dict[str, Any]):
method = json["method"]
result = {
"result": {method: {"dummy": "response"}},
"error_code": self.send_error_code.value,
}
return self._mock_response(self.status_code, result)