diff --git a/tests/transports/test_aestransport.py b/tests/transports/test_aestransport.py index d8d7909b..7f69a666 100644 --- a/tests/transports/test_aestransport.py +++ b/tests/transports/test_aestransport.py @@ -6,6 +6,7 @@ import logging import random import string import time +from contextlib import AbstractContextManager from contextlib import nullcontext as does_not_raise from json import dumps as json_dumps from json import loads as json_loads @@ -17,10 +18,11 @@ from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric import padding as asymmetric_padding from cryptography.hazmat.primitives.asymmetric.rsa import RSAPublicKey from freezegun.api import FrozenDateTimeFactory +from pytest_mock import MockerFixture from yarl import URL from kasa.credentials import Credentials -from kasa.deviceconfig import DeviceConfig +from kasa.deviceconfig import DeviceConfig, KeyPairDict from kasa.exceptions import ( AuthenticationError, KasaException, @@ -43,7 +45,7 @@ iv = b"9=\xf8\x1bS\xcd0\xb5\x89i\xba\xfd^9\x9f\xfa" KEY_IV = key + iv -def test_encrypt(): +def test_encrypt() -> None: encryption_session = AesEncyptionSession(KEY_IV[:16], KEY_IV[16:]) d = json.dumps({"foo": 1, "bar": 2}) @@ -69,8 +71,12 @@ status_parameters = pytest.mark.parametrize( @status_parameters async def test_handshake( - mocker, status_code, error_code, inner_error_code, expectation -): + mocker: MockerFixture, + status_code: int, + error_code: int, + inner_error_code: int, + expectation: AbstractContextManager, +) -> None: host = "127.0.0.1" mock_aes_device = MockAesDevice(host, status_code, error_code, inner_error_code) mocker.patch.object(aiohttp.ClientSession, "post", side_effect=mock_aes_device.post) @@ -87,12 +93,12 @@ async def test_handshake( assert transport._state is TransportState.LOGIN_REQUIRED -async def test_handshake_with_keys(mocker): +async def test_handshake_with_keys(mocker: MockerFixture) -> None: host = "127.0.0.1" mock_aes_device = MockAesDevice(host) mocker.patch.object(aiohttp.ClientSession, "post", side_effect=mock_aes_device.post) - test_keys = { + test_keys: KeyPairDict = { "private": "MIICdQIBADANBgkqhkiG9w0BAQEFAASCAl8wggJbAgEAAoGBAMo/JQpXIbP2M3bLOKyfEVCURFCxHIXv4HDME8J58AL4BwGDXf0oQycgj9nV+T/MzgEd/4iVysYuYfLuIEKXADP7Lby6AfA/dbcinZZ7bLUNMNa7TaylIvVKtSfR0LV8AmG0jdQYkr4cTzLAEd+AEs/wG3nMQNEcoQRVY+svLPDjAgMBAAECgYBCsDOch0KbvrEVmMklUoY5Fcq4+M249HIDf6d8VwznTbWxsAmL8nzCKCCG6eF4QiYjhCrAdPQaCS1PF2oXywbLhngid/9W9gz4CKKDJChs1X8KvLi+TLg1jgJUXvq9yVNh1CB+lS2ho4gdDDCbVmiVOZR5TDfEf0xeJ+Zz3zlUEQJBAPkhuNdc3yRue8huFZbrWwikURQPYBxLOYfVTDsfV9mZGSkGoWS1FPDsxrqSXugTmcTRuw+lrXKDabJ72kqywA8CQQDP0oaGh5r7F12Xzcwb7X9JkTvyr+rO8YgVtKNBaNVOPabAzysNwOlvH/sNCVQcRj8rn5LNXitgLx6T+Q5uqa3tAkA7J0elUzbkhps7ju/vYri9x448zh3K+g2R9BJio2GPmCuCM0HVEK4FOqNBH4oLXsQPGKFq6LLTUuKg74l4XRL/AkBHBO6r8pNn0yhMxCtIL/UbsuIFoVBgv/F9WWmg5K5gOnlN0n4oCRC8xPUKE3IG54qW4cVNIS05hWCxuJ7R+nJRAkByt/+kX1nQxis2wIXj90fztXG3oSmoVaieYxaXPxlWvX3/Q5kslFF5UsGy9gcK0v2PXhqjTbhud3/X0Er6YP4v", "public": "MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQDKPyUKVyGz9jN2yzisnxFQlERQsRyF7+BwzBPCefAC+AcBg139KEMnII/Z1fk/zM4BHf+IlcrGLmHy7iBClwAz+y28ugHwP3W3Ip2We2y1DTDWu02spSL1SrUn0dC1fAJhtI3UGJK+HE8ywBHfgBLP8Bt5zEDRHKEEVWPrLyzw4wIDAQAB", } @@ -106,12 +112,19 @@ async def test_handshake_with_keys(mocker): assert transport._state is TransportState.HANDSHAKE_REQUIRED await transport.perform_handshake() + assert transport._key_pair is not None assert transport._key_pair.private_key_der_b64 == test_keys["private"] assert transport._key_pair.public_key_der_b64 == test_keys["public"] @status_parameters -async def test_login(mocker, status_code, error_code, inner_error_code, expectation): +async def test_login( + mocker: MockerFixture, + status_code: int, + error_code: int, + inner_error_code: int, + expectation: AbstractContextManager, +) -> None: host = "127.0.0.1" mock_aes_device = MockAesDevice(host, status_code, error_code, inner_error_code) mocker.patch.object(aiohttp.ClientSession, "post", side_effect=mock_aes_device.post) @@ -157,7 +170,12 @@ async def test_login(mocker, status_code, error_code, inner_error_code, expectat "LOGIN_ERROR-SESSION_TIMEOUT_ERROR", ), ) -async def test_login_errors(mocker, inner_error_codes, expectation, call_count): +async def test_login_errors( + mocker: MockerFixture, + inner_error_codes: list[int], + expectation: AbstractContextManager, + call_count: int, +) -> None: host = "127.0.0.1" mock_aes_device = MockAesDevice(host, 200, 0, inner_error_codes) post_mock = mocker.patch.object( @@ -190,7 +208,13 @@ async def test_login_errors(mocker, inner_error_codes, expectation, call_count): @status_parameters -async def test_send(mocker, status_code, error_code, inner_error_code, expectation): +async def test_send( + mocker: MockerFixture, + status_code: int, + error_code: int, + inner_error_code: int, + expectation: AbstractContextManager, +) -> None: host = "127.0.0.1" mock_aes_device = MockAesDevice(host, status_code, error_code, inner_error_code) mocker.patch.object(aiohttp.ClientSession, "post", side_effect=mock_aes_device.post) @@ -198,7 +222,7 @@ async def test_send(mocker, status_code, error_code, inner_error_code, expectati transport = AesTransport( config=DeviceConfig(host, credentials=Credentials("foo", "bar")) ) - transport._handshake_done = True + 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( @@ -218,7 +242,9 @@ async def test_send(mocker, status_code, error_code, inner_error_code, expectati @pytest.mark.xdist_group(name="caplog") -async def test_unencrypted_response(mocker, caplog): +async def test_unencrypted_response( + mocker: MockerFixture, caplog: pytest.LogCaptureFixture +) -> None: 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) @@ -249,7 +275,9 @@ async def test_unencrypted_response(mocker, caplog): ) -async def test_unencrypted_response_invalid_json(mocker, caplog): +async def test_unencrypted_response_invalid_json( + mocker: MockerFixture, caplog: pytest.LogCaptureFixture +) -> None: host = "127.0.0.1" mock_aes_device = MockAesDevice( host, 200, 0, 0, do_not_encrypt_response=True, send_response=b"Foobar" @@ -283,14 +311,16 @@ ERRORS = [e for e in SmartErrorCode if e != 0] @pytest.mark.parametrize("error_code", ERRORS, ids=lambda e: e.name) -async def test_passthrough_errors(mocker, error_code): +async def test_passthrough_errors( + mocker: MockerFixture, error_code: SmartErrorCode +) -> None: host = "127.0.0.1" mock_aes_device = MockAesDevice(host, 200, error_code, 0) mocker.patch.object(aiohttp.ClientSession, "post", side_effect=mock_aes_device.post) config = DeviceConfig(host, credentials=Credentials("foo", "bar")) transport = AesTransport(config=config) - transport._handshake_done = True + 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( @@ -309,14 +339,14 @@ async def test_passthrough_errors(mocker, error_code): @pytest.mark.parametrize("error_code", [-13333, 13333]) -async def test_unknown_errors(mocker, error_code): +async def test_unknown_errors(mocker: MockerFixture, error_code: int) -> None: host = "127.0.0.1" mock_aes_device = MockAesDevice(host, 200, error_code, 0) mocker.patch.object(aiohttp.ClientSession, "post", side_effect=mock_aes_device.post) config = DeviceConfig(host, credentials=Credentials("foo", "bar")) transport = AesTransport(config=config) - transport._handshake_done = True + 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( @@ -335,7 +365,7 @@ async def test_unknown_errors(mocker, error_code): assert res is SmartErrorCode.INTERNAL_UNKNOWN_ERROR -async def test_port_override(): +async def test_port_override() -> None: """Test that port override sets the app_url.""" host = "127.0.0.1" config = DeviceConfig( @@ -356,11 +386,11 @@ async def test_port_override(): ], ) async def test_device_closes_connection( - mocker, + mocker: MockerFixture, freezer: FrozenDateTimeFactory, - device_delay_required, - should_error, - should_succeed, + device_delay_required: float, + should_error: bool, + should_succeed: bool, ): """Test the delay logic in http client to deal with devices that close connections after each request. diff --git a/tests/transports/test_klaptransport.py b/tests/transports/test_klaptransport.py index 7fd56cf5..b1931b99 100644 --- a/tests/transports/test_klaptransport.py +++ b/tests/transports/test_klaptransport.py @@ -1,12 +1,17 @@ +from __future__ import annotations + import json import logging import re import secrets import time +from collections.abc import Callable +from contextlib import AbstractContextManager from contextlib import nullcontext as does_not_raise import aiohttp import pytest +from pytest_mock import MockerFixture from yarl import URL from kasa.credentials import DEFAULT_CREDENTIALS, Credentials, get_default_credentials @@ -63,8 +68,13 @@ class _mock_response: @pytest.mark.parametrize("protocol_class", [IotProtocol, SmartProtocol]) @pytest.mark.parametrize("retry_count", [1, 3, 5]) async def test_protocol_retries_via_client_session( - mocker, retry_count, protocol_class, transport_class, error, retry_expectation -): + mocker: MockerFixture, + retry_count: int, + protocol_class: type, + transport_class: type, + error: Exception, + retry_expectation: bool, +) -> None: host = "127.0.0.1" conn = mocker.patch.object(aiohttp.ClientSession, "post", side_effect=error) @@ -91,8 +101,13 @@ async def test_protocol_retries_via_client_session( @pytest.mark.parametrize("protocol_class", [IotProtocol, SmartProtocol]) @pytest.mark.parametrize("retry_count", [1, 3, 5]) async def test_protocol_retries_via_httpclient( - mocker, retry_count, protocol_class, transport_class, error, retry_expectation -): + mocker: MockerFixture, + retry_count: int, + protocol_class: type, + transport_class: type, + error: Exception, + retry_expectation: bool, +) -> None: host = "127.0.0.1" conn = mocker.patch.object(HttpClient, "post", side_effect=error) mocker.patch.object(protocol_class, "BACKOFF_SECONDS_AFTER_TIMEOUT", 0) @@ -110,8 +125,8 @@ async def test_protocol_retries_via_httpclient( @pytest.mark.parametrize("transport_class", [AesTransport, KlapTransport]) @pytest.mark.parametrize("protocol_class", [IotProtocol, SmartProtocol]) async def test_protocol_no_retry_on_connection_error( - mocker, protocol_class, transport_class -): + mocker: MockerFixture, protocol_class: type, transport_class: type +) -> None: host = "127.0.0.1" conn = mocker.patch.object( aiohttp.ClientSession, @@ -131,8 +146,8 @@ async def test_protocol_no_retry_on_connection_error( @pytest.mark.parametrize("transport_class", [AesTransport, KlapTransport]) @pytest.mark.parametrize("protocol_class", [IotProtocol, SmartProtocol]) async def test_protocol_retry_recoverable_error( - mocker, protocol_class, transport_class -): + mocker: MockerFixture, protocol_class: type, transport_class: type +) -> None: host = "127.0.0.1" conn = mocker.patch.object( aiohttp.ClientSession, @@ -152,7 +167,9 @@ async def test_protocol_retry_recoverable_error( @pytest.mark.parametrize("transport_class", [AesTransport, KlapTransport]) @pytest.mark.parametrize("protocol_class", [IotProtocol, SmartProtocol]) @pytest.mark.parametrize("retry_count", [1, 3, 5]) -async def test_protocol_reconnect(mocker, retry_count, protocol_class, transport_class): +async def test_protocol_reconnect( + mocker: MockerFixture, retry_count: int, protocol_class: type, transport_class: type +) -> None: host = "127.0.0.1" remaining = retry_count mock_response = {"result": {"great": "success"}, "error_code": 0} @@ -185,12 +202,13 @@ async def test_protocol_reconnect(mocker, retry_count, protocol_class, transport @pytest.mark.parametrize("log_level", [logging.WARNING, logging.DEBUG]) @pytest.mark.xdist_group(name="caplog") -async def test_protocol_logging(mocker, caplog, log_level): +async def test_protocol_logging( + mocker: MockerFixture, caplog: pytest.LogCaptureFixture, log_level: int +) -> None: caplog.set_level(log_level) logging.getLogger("kasa").setLevel(log_level) def _return_encrypted(*_, **__): - nonlocal encryption_session # Do the encrypt just before returning the value so the incrementing sequence number is correct encrypted, seq = encryption_session.encrypt('{"great":"success"}') return 200, encrypted @@ -200,11 +218,12 @@ async def test_protocol_logging(mocker, caplog, log_level): encryption_session = KlapEncryptionSession(seed, seed, auth_hash) config = DeviceConfig("127.0.0.1") - protocol = IotProtocol(transport=KlapTransport(config=config)) + transport = KlapTransport(config=config) + protocol = IotProtocol(transport=transport) - protocol._transport._handshake_done = True - protocol._transport._session_expire_at = time.time() + 86400 - protocol._transport._encryption_session = encryption_session + transport._handshake_done = True + transport._session_expire_at = time.time() + 86400 + transport._encryption_session = encryption_session mocker.patch.object(HttpClient, "post", side_effect=_return_encrypted) response = await protocol.query({}) @@ -215,7 +234,7 @@ async def test_protocol_logging(mocker, caplog, log_level): assert "success" not in caplog.text -def test_encrypt(): +def test_encrypt() -> None: d = json.dumps({"foo": 1, "bar": 2}) seed = secrets.token_bytes(16) @@ -227,7 +246,7 @@ def test_encrypt(): assert d == encryption_session.decrypt(encrypted) -def test_encrypt_unicode(): +def test_encrypt_unicode() -> None: d = "{'snowman': '\u2603'}" seed = secrets.token_bytes(16) @@ -241,7 +260,7 @@ def test_encrypt_unicode(): assert d == decrypted -async def test_transport_decrypt(mocker): +async def test_transport_decrypt(mocker: MockerFixture) -> None: """Test transport decryption.""" d = {"great": "success"} @@ -255,6 +274,7 @@ async def test_transport_decrypt(mocker): transport._encryption_session = encryption_session async def _return_response(url: URL, params=None, data=None, *_, **__): + assert transport._encryption_session is not None encryption_session = KlapEncryptionSession( transport._encryption_session.local_seed, transport._encryption_session.remote_seed, @@ -272,7 +292,9 @@ async def test_transport_decrypt(mocker): assert d == resp -async def test_transport_decrypt_error(mocker, caplog): +async def test_transport_decrypt_error( + mocker: MockerFixture, caplog: pytest.LogCaptureFixture +) -> None: """Test that a decryption error raises a kasa exception.""" d = {"great": "success"} @@ -286,6 +308,7 @@ async def test_transport_decrypt_error(mocker, caplog): transport._encryption_session = encryption_session async def _return_response(url: URL, params=None, data=None, *_, **__): + assert transport._encryption_session is not None encryption_session = KlapEncryptionSession( secrets.token_bytes(16), transport._encryption_session.remote_seed, @@ -330,10 +353,19 @@ async def test_transport_decrypt_error(mocker, caplog): ], ) async def test_handshake1( - mocker, device_credentials, expectation, transport_class, seed_auth_hash_calc -): + mocker: MockerFixture, + device_credentials: Credentials, + expectation: AbstractContextManager, + transport_class: type[KlapTransport], + seed_auth_hash_calc: Callable[..., bytes], +) -> None: + client_seed = None + server_seed = secrets.token_bytes(16) + client_credentials = Credentials("foo", "bar") + device_auth_hash = transport_class.generate_auth_hash(device_credentials) + async def _return_handshake1_response(url, params=None, data=None, *_, **__): - nonlocal client_seed, server_seed, device_auth_hash + nonlocal client_seed client_seed = data seed_auth_hash = _sha256( @@ -341,24 +373,20 @@ async def test_handshake1( ) return _mock_response(200, server_seed + seed_auth_hash) - client_seed = None - server_seed = secrets.token_bytes(16) - client_credentials = Credentials("foo", "bar") - device_auth_hash = transport_class.generate_auth_hash(device_credentials) - mocker.patch.object( aiohttp.ClientSession, "post", side_effect=_return_handshake1_response ) config = DeviceConfig("127.0.0.1", credentials=client_credentials) - protocol = IotProtocol(transport=transport_class(config=config)) + transport = transport_class(config=config) + protocol = IotProtocol(transport=transport) with expectation: ( local_seed, device_remote_seed, auth_hash, - ) = await protocol._transport.perform_handshake1() + ) = await transport.perform_handshake1() assert local_seed == client_seed assert device_remote_seed == server_seed @@ -381,15 +409,18 @@ async def test_handshake1( ], ) async def test_handshake( - mocker, transport_class, seed_auth_hash_calc1, seed_auth_hash_calc2 -): + mocker: MockerFixture, + transport_class: type[KlapTransport], + seed_auth_hash_calc1: Callable[..., bytes], + seed_auth_hash_calc2: Callable[..., bytes], +) -> None: client_seed = None server_seed = secrets.token_bytes(16) client_credentials = Credentials("foo", "bar") device_auth_hash = transport_class.generate_auth_hash(client_credentials) async def _return_handshake_response(url: URL, params=None, data=None, *_, **__): - nonlocal client_seed, server_seed, device_auth_hash + nonlocal client_seed if url == URL("http://127.0.0.1:80/app/handshake1"): client_seed = data @@ -410,20 +441,21 @@ async def test_handshake( ) config = DeviceConfig("127.0.0.1", credentials=client_credentials) - protocol = IotProtocol(transport=transport_class(config=config)) + transport = transport_class(config=config) + protocol = IotProtocol(transport=transport) response_status = 200 - await protocol._transport.perform_handshake() - assert protocol._transport._handshake_done is True + await transport.perform_handshake() + assert transport._handshake_done is True response_status = 403 with pytest.raises(KasaException): - await protocol._transport.perform_handshake() - assert protocol._transport._handshake_done is False + await transport.perform_handshake() + assert transport._handshake_done is False await protocol.close() -async def test_query(mocker): +async def test_query(mocker: MockerFixture) -> None: client_seed = None last_seq = None seq = None @@ -432,7 +464,7 @@ async def test_query(mocker): device_auth_hash = KlapTransport.generate_auth_hash(client_credentials) async def _return_response(url: URL, params=None, data=None, *_, **__): - nonlocal client_seed, server_seed, device_auth_hash, seq + nonlocal client_seed, seq if url == URL("http://127.0.0.1:80/app/handshake1"): client_seed = data @@ -442,10 +474,11 @@ async def test_query(mocker): elif url == URL("http://127.0.0.1:80/app/handshake2"): return _mock_response(200, b"") elif url == URL("http://127.0.0.1:80/app/request"): + assert transport._encryption_session is not None encryption_session = KlapEncryptionSession( - protocol._transport._encryption_session.local_seed, - protocol._transport._encryption_session.remote_seed, - protocol._transport._encryption_session.user_hash, + transport._encryption_session.local_seed, + transport._encryption_session.remote_seed, + transport._encryption_session.user_hash, ) seq = params.get("seq") encryption_session._seq = seq - 1 @@ -456,7 +489,8 @@ async def test_query(mocker): mocker.patch.object(aiohttp.ClientSession, "post", side_effect=_return_response) config = DeviceConfig("127.0.0.1", credentials=client_credentials) - protocol = IotProtocol(transport=KlapTransport(config=config)) + transport = KlapTransport(config=config) + protocol = IotProtocol(transport=transport) for _ in range(10): resp = await protocol.query({}) @@ -508,8 +542,11 @@ async def test_query(mocker): ], ) async def test_authentication_failures( - mocker, response_status, credentials_match, expectation -): + mocker: MockerFixture, + response_status: tuple[int, ...], + credentials_match: bool | bytes, + expectation: AbstractContextManager, +) -> None: client_seed = None server_seed = secrets.token_bytes(16) @@ -520,12 +557,7 @@ async def test_authentication_failures( device_auth_hash = KlapTransport.generate_auth_hash(device_credentials) async def _return_response(url: URL, params=None, data=None, *_, **__): - nonlocal \ - client_seed, \ - server_seed, \ - device_auth_hash, \ - response_status, \ - credentials_match + nonlocal client_seed if url == URL("http://127.0.0.1:80/app/handshake1"): client_seed = data @@ -553,7 +585,7 @@ async def test_authentication_failures( await protocol.query({}) -async def test_port_override(): +async def test_port_override() -> None: """Test that port override sets the app_url.""" host = "127.0.0.1" config = DeviceConfig( diff --git a/tests/transports/test_linkietransport.py b/tests/transports/test_linkietransport.py index 1ac8dba5..a907513a 100644 --- a/tests/transports/test_linkietransport.py +++ b/tests/transports/test_linkietransport.py @@ -3,6 +3,7 @@ from unittest.mock import ANY import aiohttp import pytest +from pytest_mock import MockerFixture from yarl import URL from kasa.credentials import DEFAULT_CREDENTIALS, Credentials, get_default_credentials @@ -18,7 +19,7 @@ KASACAM_RESPONSE_ERROR = '{"smartlife.cam.ipcamera.cloud": {"get_inf": {"err_cod KASA_DEFAULT_CREDENTIALS_HASH = "YWRtaW46MjEyMzJmMjk3YTU3YTVhNzQzODk0YTBlNGE4MDFmYzM=" -async def test_working(mocker): +async def test_working(mocker: MockerFixture) -> None: """No errors with an expected request/response.""" host = "127.0.0.1" mock_linkie_device = MockLinkieDevice(host) @@ -35,7 +36,7 @@ async def test_working(mocker): } -async def test_credentials_hash(mocker): +async def test_credentials_hash(mocker: MockerFixture) -> None: """Ensure the default credentials are always passed as Basic Auth.""" # Test without credentials input @@ -92,7 +93,9 @@ async def test_credentials_hash(mocker): (200, KASACAM_RESPONSE_ERROR, "Unsupported API call"), ], ) -async def test_exceptions(mocker, return_status, return_data, expected): +async def test_exceptions( + mocker: MockerFixture, return_status: int, return_data: str, expected: str +) -> None: """Test a variety of possible responses from the device.""" host = "127.0.0.1" transport = LinkieTransportV2(config=DeviceConfig(host)) @@ -107,7 +110,7 @@ async def test_exceptions(mocker, return_status, return_data, expected): await transport.send(KASACAM_REQUEST_PLAINTEXT) -def _generate_kascam_basic_auth(): +def _generate_kascam_basic_auth() -> str: creds = get_default_credentials(DEFAULT_CREDENTIALS["KASACAMERA"]) creds_combined = f"{creds.username}:{creds.password}" return base64.b64encode(creds_combined.encode()).decode() diff --git a/tests/transports/test_sslaestransport.py b/tests/transports/test_sslaestransport.py index 0105205d..fe408195 100644 --- a/tests/transports/test_sslaestransport.py +++ b/tests/transports/test_sslaestransport.py @@ -3,6 +3,7 @@ from __future__ import annotations import base64 import logging import secrets +from contextlib import AbstractContextManager from contextlib import nullcontext as does_not_raise from json import dumps as json_dumps from json import loads as json_loads @@ -10,6 +11,7 @@ from typing import Any import aiohttp import pytest +from pytest_mock import MockerFixture from yarl import URL from kasa.credentials import DEFAULT_CREDENTIALS, Credentials, get_default_credentials @@ -107,14 +109,14 @@ MOCK_UNENCRYPTED_PASSTHROUGH_STOK = "32charLowerCaseHexStok" ], ) async def test_handshake( - mocker, - status_code, - username, - password, - wants_default_user, - digest_password_fail, - expectation, -): + mocker: MockerFixture, + status_code: int, + username: str, + password: str, + wants_default_user: bool, + digest_password_fail: bool, + expectation: AbstractContextManager, +) -> None: host = "127.0.0.1" mock_ssl_aes_device = MockSslAesDevice( host, @@ -142,7 +144,9 @@ async def test_handshake( ("wants_default_user"), [pytest.param(False, id="username"), pytest.param(True, id="default")], ) -async def test_credentials_hash(mocker, wants_default_user): +async def test_credentials_hash( + mocker: MockerFixture, wants_default_user: bool +) -> None: host = "127.0.0.1" mock_ssl_aes_device = MockSslAesDevice( host, want_default_username=wants_default_user @@ -167,7 +171,7 @@ async def test_credentials_hash(mocker, wants_default_user): assert transport.credentials_hash == creds_hash -async def test_send(mocker): +async def test_send(mocker: MockerFixture) -> None: host = "127.0.0.1" mock_ssl_aes_device = MockSslAesDevice(host, want_default_username=False) mocker.patch.object( @@ -187,7 +191,9 @@ async def test_send(mocker): @pytest.mark.xdist_group(name="caplog") -async def test_unencrypted_response(mocker, caplog): +async def test_unencrypted_response( + mocker: MockerFixture, caplog: pytest.LogCaptureFixture +) -> None: host = "127.0.0.1" mock_ssl_aes_device = MockSslAesDevice(host, do_not_encrypt_response=True) mocker.patch.object( @@ -213,7 +219,9 @@ async def test_unencrypted_response(mocker, caplog): @pytest.mark.parametrize(("want_default"), [True, False]) @pytest.mark.xdist_group(name="caplog") -async def test_unencrypted_passthrough(mocker, caplog, want_default): +async def test_unencrypted_passthrough( + mocker: MockerFixture, caplog: pytest.LogCaptureFixture, want_default: bool +) -> None: host = "127.0.0.1" mock_ssl_aes_device = MockSslAesDevice( host, unencrypted_passthrough=True, want_default_username=want_default @@ -240,7 +248,9 @@ async def test_unencrypted_passthrough(mocker, caplog, want_default): @pytest.mark.parametrize(("want_default"), [True, False]) @pytest.mark.xdist_group(name="caplog") -async def test_unencrypted_passthrough_errors(mocker, caplog, want_default): +async def test_unencrypted_passthrough_errors( + mocker: MockerFixture, caplog: pytest.LogCaptureFixture, want_default: bool +) -> None: host = "127.0.0.1" request = { "method": "getDeviceInfo", @@ -329,7 +339,7 @@ async def test_unencrypted_passthrough_errors(mocker, caplog, want_default): await transport.send(json_dumps(request)) -async def test_device_blocked_response(mocker): +async def test_device_blocked_response(mocker: MockerFixture) -> None: host = "127.0.0.1" mock_ssl_aes_device = MockSslAesDevice(host, device_blocked=True) mocker.patch.object( @@ -360,7 +370,9 @@ async def test_device_blocked_response(mocker): ), ], ) -async def test_device_500_error(mocker, response, expected_msg): +async def test_device_500_error( + mocker: MockerFixture, response: dict | bytes, expected_msg: str +) -> None: """Test 500 error raises retryable exception.""" host = "127.0.0.1" mock_ssl_aes_device = MockSslAesDevice(host) @@ -387,7 +399,7 @@ async def test_device_500_error(mocker, response, expected_msg): await transport.send(json_dumps(request)) -async def test_port_override(): +async def test_port_override() -> None: """Test that port override sets the app_url.""" host = "127.0.0.1" port_override = 12345 @@ -420,8 +432,8 @@ async def test_port_override(): ], ) async def test_login_version_default_credentials( - mocker, login_version, expected_password_b64 -): + mocker: MockerFixture, login_version: int | None, expected_password_b64: str +) -> None: """Test that login_version=3 uses TAPOCAMERA_LV3 credentials while other versions use TAPOCAMERA.""" host = "127.0.0.1" tapo_family = DeviceFamily.SmartIpCamera diff --git a/tests/transports/test_ssltransport.py b/tests/transports/test_ssltransport.py index 37b79725..939e2ea6 100644 --- a/tests/transports/test_ssltransport.py +++ b/tests/transports/test_ssltransport.py @@ -2,11 +2,13 @@ from __future__ import annotations import logging from base64 import b64encode +from contextlib import AbstractContextManager from contextlib import nullcontext as does_not_raise from typing import Any import aiohttp import pytest +from pytest_mock import MockerFixture from yarl import URL from kasa.credentials import DEFAULT_CREDENTIALS, Credentials, get_default_credentials @@ -122,13 +124,13 @@ _LOGGER = logging.getLogger(__name__) ], ) async def test_login( - mocker, - status_code, - error_code, - username, - password, - expectation, -): + mocker: MockerFixture, + status_code: int, + error_code: SmartErrorCode | list[SmartErrorCode], + username: str, + password: str, + expectation: AbstractContextManager, +) -> None: host = "127.0.0.1" mock_ssl_aes_device = MockSslDevice( host, @@ -151,7 +153,7 @@ async def test_login( await transport.close() -async def test_credentials_hash(mocker): +async def test_credentials_hash(mocker: MockerFixture) -> None: host = "127.0.0.1" mock_ssl_aes_device = MockSslDevice(host) mocker.patch.object( @@ -174,7 +176,7 @@ async def test_credentials_hash(mocker): await transport.close() -async def test_send(mocker): +async def test_send(mocker: MockerFixture) -> None: host = "127.0.0.1" mock_ssl_aes_device = MockSslDevice(host, send_error_code=SmartErrorCode.SUCCESS) mocker.patch.object( @@ -203,7 +205,7 @@ async def test_send(mocker): await transport.close() -async def test_no_credentials(mocker): +async def test_no_credentials(mocker: MockerFixture) -> None: """Test transport without credentials.""" host = "127.0.0.1" mock_ssl_aes_device = MockSslDevice( @@ -225,7 +227,7 @@ async def test_no_credentials(mocker): await transport.close() -async def test_reset(mocker): +async def test_reset(mocker: MockerFixture) -> None: """Test that transport state adjusts correctly for reset.""" host = "127.0.0.1" mock_ssl_aes_device = MockSslDevice(host, send_error_code=SmartErrorCode.SUCCESS) @@ -249,7 +251,7 @@ async def test_reset(mocker): assert str(transport._app_url) == "https://127.0.0.1:4433/app" -async def test_port_override(): +async def test_port_override() -> None: """Test that port override sets the app_url.""" host = "127.0.0.1" port_override = 12345