From 291626e10afc9de8bfdb14a6ec9fc36972196d73 Mon Sep 17 00:00:00 2001 From: ZeliardM Date: Mon, 6 Apr 2026 22:22:28 -0400 Subject: [PATCH] Add type annotations to cli, protocols, and smartcam tests --- tests/cli/test_hub.py | 16 +- tests/cli/test_vacuum.py | 21 ++- tests/protocols/test_iotprotocol.py | 202 ++++++++++++++-------- tests/protocols/test_smartprotocol.py | 82 ++++++--- tests/smartcam/modules/test_alarm.py | 6 +- tests/smartcam/modules/test_battery.py | 14 +- tests/smartcam/modules/test_camera.py | 6 +- tests/smartcam/modules/test_childsetup.py | 6 +- tests/smartcam/modules/test_detections.py | 6 +- tests/smartcam/modules/test_pantilt.py | 20 +-- tests/smartcam/test_smartcamdevice.py | 14 +- 11 files changed, 255 insertions(+), 138 deletions(-) diff --git a/tests/cli/test_hub.py b/tests/cli/test_hub.py index 00c3645ed..c35308780 100644 --- a/tests/cli/test_hub.py +++ b/tests/cli/test_hub.py @@ -1,14 +1,20 @@ import pytest +from asyncclick.testing import CliRunner from pytest_mock import MockerFixture -from kasa import DeviceType, Module +from kasa import Device, DeviceType, Module from kasa.cli.hub import hub from ..device_fixtures import hubs, plug_iot @hubs -async def test_hub_pair(dev, mocker: MockerFixture, runner, caplog): +async def test_hub_pair( + dev: Device, + mocker: MockerFixture, + runner: CliRunner, + caplog: pytest.LogCaptureFixture, +) -> None: """Test that pair calls the expected methods.""" cs = dev.modules.get(Module.ChildSetup) # Patch if the device supports the module @@ -26,7 +32,9 @@ async def test_hub_pair(dev, mocker: MockerFixture, runner, caplog): @hubs -async def test_hub_unpair(dev, mocker: MockerFixture, runner): +async def test_hub_unpair( + dev: Device, mocker: MockerFixture, runner: CliRunner +) -> None: """Test that unpair calls the expected method.""" if not dev.children: pytest.skip("Cannot test without child devices") @@ -44,7 +52,7 @@ async def test_hub_unpair(dev, mocker: MockerFixture, runner): @plug_iot -async def test_non_hub(dev, mocker: MockerFixture, runner): +async def test_non_hub(dev: Device, mocker: MockerFixture, runner: CliRunner) -> None: """Test that hub commands return an error if executed on a non-hub.""" assert dev.device_type is not DeviceType.Hub res = await runner.invoke( diff --git a/tests/cli/test_vacuum.py b/tests/cli/test_vacuum.py index a790286e6..d45764369 100644 --- a/tests/cli/test_vacuum.py +++ b/tests/cli/test_vacuum.py @@ -1,6 +1,7 @@ +from asyncclick.testing import CliRunner from pytest_mock import MockerFixture -from kasa import DeviceType, Module +from kasa import Device, DeviceType, Module from kasa.cli.vacuum import vacuum from ..device_fixtures import plug_iot @@ -8,7 +9,9 @@ @vacuum_devices -async def test_vacuum_records_group(dev, mocker: MockerFixture, runner): +async def test_vacuum_records_group( + dev: Device, mocker: MockerFixture, runner: CliRunner +) -> None: """Test that vacuum records calls the expected methods.""" rec = dev.modules.get(Module.CleanRecords) assert rec @@ -26,7 +29,9 @@ async def test_vacuum_records_group(dev, mocker: MockerFixture, runner): @vacuum_devices -async def test_vacuum_records_list(dev, mocker: MockerFixture, runner): +async def test_vacuum_records_list( + dev: Device, mocker: MockerFixture, runner: CliRunner +) -> None: """Test that vacuum records list calls the expected methods.""" rec = dev.modules.get(Module.CleanRecords) assert rec @@ -46,7 +51,7 @@ async def test_vacuum_records_list(dev, mocker: MockerFixture, runner): @vacuum_devices -async def test_vacuum_consumables(dev, runner): +async def test_vacuum_consumables(dev: Device, runner: CliRunner) -> None: """Test that vacuum consumables calls the expected methods.""" cons = dev.modules.get(Module.Consumables) assert cons @@ -62,7 +67,9 @@ async def test_vacuum_consumables(dev, runner): @vacuum_devices -async def test_vacuum_consumables_reset(dev, mocker: MockerFixture, runner): +async def test_vacuum_consumables_reset( + dev: Device, mocker: MockerFixture, runner: CliRunner +) -> None: """Test that vacuum consumables reset calls the expected methods.""" cons = dev.modules.get(Module.Consumables) assert cons @@ -89,7 +96,9 @@ async def test_vacuum_consumables_reset(dev, mocker: MockerFixture, runner): @plug_iot -async def test_non_vacuum(dev, mocker: MockerFixture, runner): +async def test_non_vacuum( + dev: Device, mocker: MockerFixture, runner: CliRunner +) -> None: """Test that vacuum commands return an error if executed on a non-vacuum.""" assert dev.device_type is not DeviceType.Vacuum diff --git a/tests/protocols/test_iotprotocol.py b/tests/protocols/test_iotprotocol.py index 0db91fcab..8842b5510 100644 --- a/tests/protocols/test_iotprotocol.py +++ b/tests/protocols/test_iotprotocol.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import asyncio import errno import importlib @@ -12,6 +14,7 @@ from unittest.mock import AsyncMock import pytest +from pytest_mock import MockerFixture from kasa.credentials import Credentials from kasa.device import Device @@ -42,8 +45,13 @@ ids=("_deprecated_TPLinkSmartHomeProtocol", "IotProtocol-XorTransport"), ) @pytest.mark.parametrize("retry_count", [1, 3, 5]) -async def test_protocol_retries(mocker, retry_count, protocol_class, transport_class): - def aio_mock_writer(_, __): +async def test_protocol_retries( + mocker: MockerFixture, + retry_count: int, + protocol_class: type[BaseProtocol], + transport_class: type[XorTransport], +) -> None: + def aio_mock_writer(_: object, __: object): reader = mocker.patch("asyncio.StreamReader") writer = mocker.patch("asyncio.StreamWriter") @@ -72,8 +80,10 @@ def aio_mock_writer(_, __): ids=("_deprecated_TPLinkSmartHomeProtocol", "IotProtocol-XorTransport"), ) async def test_protocol_no_retry_on_unreachable( - mocker, protocol_class, transport_class -): + mocker: MockerFixture, + protocol_class: type[BaseProtocol], + transport_class: type[XorTransport], +) -> None: conn = mocker.patch( "asyncio.open_connection", side_effect=OSError(errno.EHOSTUNREACH, "No route to host"), @@ -96,8 +106,10 @@ async def test_protocol_no_retry_on_unreachable( ids=("_deprecated_TPLinkSmartHomeProtocol", "IotProtocol-XorTransport"), ) async def test_protocol_no_retry_connection_refused( - mocker, protocol_class, transport_class -): + mocker: MockerFixture, + protocol_class: type[BaseProtocol], + transport_class: type[XorTransport], +) -> None: conn = mocker.patch( "asyncio.open_connection", side_effect=ConnectionRefusedError, @@ -120,8 +132,10 @@ async def test_protocol_no_retry_connection_refused( ids=("_deprecated_TPLinkSmartHomeProtocol", "IotProtocol-XorTransport"), ) async def test_protocol_retry_recoverable_error( - mocker, protocol_class, transport_class -): + mocker: MockerFixture, + protocol_class: type[BaseProtocol], + transport_class: type[XorTransport], +) -> None: conn = mocker.patch( "asyncio.open_connection", side_effect=OSError(errno.ECONNRESET, "Connection reset by peer"), @@ -149,20 +163,24 @@ async def test_protocol_retry_recoverable_error( ) @pytest.mark.parametrize("retry_count", [1, 3, 5]) async def test_protocol_reconnect( - mocker, retry_count, protocol_class, transport_class, encryption_class -): + mocker: MockerFixture, + retry_count: int, + protocol_class: type[BaseProtocol], + transport_class: type[XorTransport], + encryption_class, +) -> None: remaining = retry_count encrypted = encryption_class.encrypt('{"great":"success"}')[ transport_class.BLOCK_SIZE : ] - def _fail_one_less_than_retry_count(*_): + def _fail_one_less_than_retry_count(*_) -> None: nonlocal remaining remaining -= 1 if remaining: raise Exception("Simulated write failure") - async def _mock_read(byte_count): + async def _mock_read(byte_count: int): nonlocal encrypted if byte_count == transport_class.BLOCK_SIZE: return struct.pack(">I", len(encrypted)) @@ -171,7 +189,7 @@ async def _mock_read(byte_count): raise ValueError(f"No mock for {byte_count}") - def aio_mock_writer(_, __): + def aio_mock_writer(_: object, __: object): reader = mocker.patch("asyncio.StreamReader") writer = mocker.patch("asyncio.StreamWriter") mocker.patch.object(writer, "write", _fail_one_less_than_retry_count) @@ -199,20 +217,23 @@ def aio_mock_writer(_, __): ids=("_deprecated_TPLinkSmartHomeProtocol", "IotProtocol-XorTransport"), ) async def test_protocol_handles_cancellation_during_write( - mocker, protocol_class, transport_class, encryption_class -): + mocker: MockerFixture, + protocol_class, + transport_class: type[XorTransport], + encryption_class, +) -> None: attempts = 0 encrypted = encryption_class.encrypt('{"great":"success"}')[ transport_class.BLOCK_SIZE : ] - def _cancel_first_attempt(*_): + def _cancel_first_attempt(*_) -> None: nonlocal attempts attempts += 1 if attempts == 1: raise asyncio.CancelledError("Simulated task cancel") - async def _mock_read(byte_count): + async def _mock_read(byte_count: int): nonlocal encrypted if byte_count == transport_class.BLOCK_SIZE: return struct.pack(">I", len(encrypted)) @@ -221,7 +242,7 @@ async def _mock_read(byte_count): raise ValueError(f"No mock for {byte_count}") - def aio_mock_writer(_, __): + def aio_mock_writer(_: object, __: object): reader = mocker.patch("asyncio.StreamReader") writer = mocker.patch("asyncio.StreamWriter") mocker.patch.object(writer, "write", _cancel_first_attempt) @@ -254,14 +275,17 @@ def aio_mock_writer(_, __): ids=("_deprecated_TPLinkSmartHomeProtocol", "IotProtocol-XorTransport"), ) async def test_protocol_handles_cancellation_during_connection( - mocker, protocol_class, transport_class, encryption_class -): + mocker: MockerFixture, + protocol_class, + transport_class: type[XorTransport], + encryption_class, +) -> None: attempts = 0 encrypted = encryption_class.encrypt('{"great":"success"}')[ transport_class.BLOCK_SIZE : ] - async def _mock_read(byte_count): + async def _mock_read(byte_count: int): nonlocal encrypted if byte_count == transport_class.BLOCK_SIZE: return struct.pack(">I", len(encrypted)) @@ -270,7 +294,7 @@ async def _mock_read(byte_count): raise ValueError(f"No mock for {byte_count}") - def aio_mock_writer(_, __): + def aio_mock_writer(_: object, __: object): nonlocal attempts attempts += 1 if attempts == 1: @@ -307,20 +331,23 @@ def aio_mock_writer(_, __): ids=("_deprecated_TPLinkSmartHomeProtocol", "IotProtocol-XorTransport"), ) async def test_protocol_handles_timeout_during_write( - mocker, protocol_class, transport_class, encryption_class -): + mocker: MockerFixture, + protocol_class, + transport_class: type[XorTransport], + encryption_class, +) -> None: attempts = 0 encrypted = encryption_class.encrypt('{"great":"success"}')[ transport_class.BLOCK_SIZE : ] - def _timeout_first_attempt(*_): + def _timeout_first_attempt(*_) -> None: nonlocal attempts attempts += 1 if attempts == 1: raise TimeoutError("Simulated timeout") - async def _mock_read(byte_count): + async def _mock_read(byte_count: int): nonlocal encrypted if byte_count == transport_class.BLOCK_SIZE: return struct.pack(">I", len(encrypted)) @@ -329,7 +356,7 @@ async def _mock_read(byte_count): raise ValueError(f"No mock for {byte_count}") - def aio_mock_writer(_, __): + def aio_mock_writer(_: object, __: object): reader = mocker.patch("asyncio.StreamReader") writer = mocker.patch("asyncio.StreamWriter") mocker.patch.object(writer, "write", _timeout_first_attempt) @@ -360,14 +387,17 @@ def aio_mock_writer(_, __): ids=("_deprecated_TPLinkSmartHomeProtocol", "IotProtocol-XorTransport"), ) async def test_protocol_handles_timeout_during_connection( - mocker, protocol_class, transport_class, encryption_class -): + mocker: MockerFixture, + protocol_class, + transport_class: type[XorTransport], + encryption_class, +) -> None: attempts = 0 encrypted = encryption_class.encrypt('{"great":"success"}')[ transport_class.BLOCK_SIZE : ] - async def _mock_read(byte_count): + async def _mock_read(byte_count: int): nonlocal encrypted if byte_count == transport_class.BLOCK_SIZE: return struct.pack(">I", len(encrypted)) @@ -376,7 +406,7 @@ async def _mock_read(byte_count): raise ValueError(f"No mock for {byte_count}") - def aio_mock_writer(_, __): + def aio_mock_writer(_: object, __: object): nonlocal attempts attempts += 1 if attempts == 1: @@ -414,16 +444,19 @@ def aio_mock_writer(_, __): ids=("_deprecated_TPLinkSmartHomeProtocol", "IotProtocol-XorTransport"), ) async def test_protocol_handles_timeout_failure_during_write( - mocker, protocol_class, transport_class, encryption_class -): + mocker: MockerFixture, + protocol_class, + transport_class: type[XorTransport], + encryption_class, +) -> None: encrypted = encryption_class.encrypt('{"great":"success"}')[ transport_class.BLOCK_SIZE : ] - def _timeout_all_attempts(*_): + def _timeout_all_attempts(*_) -> None: raise TimeoutError("Simulated timeout") - async def _mock_read(byte_count): + async def _mock_read(byte_count: int): nonlocal encrypted if byte_count == transport_class.BLOCK_SIZE: return struct.pack(">I", len(encrypted)) @@ -432,7 +465,7 @@ async def _mock_read(byte_count): raise ValueError(f"No mock for {byte_count}") - def aio_mock_writer(_, __): + def aio_mock_writer(_: object, __: object): reader = mocker.patch("asyncio.StreamReader") writer = mocker.patch("asyncio.StreamWriter") mocker.patch.object(writer, "write", _timeout_all_attempts) @@ -465,13 +498,16 @@ def aio_mock_writer(_, __): ids=("_deprecated_TPLinkSmartHomeProtocol", "IotProtocol-XorTransport"), ) async def test_protocol_handles_timeout_failure_during_connection( - mocker, protocol_class, transport_class, encryption_class -): + mocker: MockerFixture, + protocol_class, + transport_class: type[XorTransport], + encryption_class, +) -> None: encrypted = encryption_class.encrypt('{"great":"success"}')[ transport_class.BLOCK_SIZE : ] - async def _mock_read(byte_count): + async def _mock_read(byte_count: int): nonlocal encrypted if byte_count == transport_class.BLOCK_SIZE: return struct.pack(">I", len(encrypted)) @@ -480,7 +516,7 @@ async def _mock_read(byte_count): raise ValueError(f"No mock for {byte_count}") - def aio_mock_writer(_, __): + def aio_mock_writer(_: object, __: object): raise TimeoutError("Simulated timeout") config = DeviceConfig("127.0.0.1") @@ -513,15 +549,20 @@ def aio_mock_writer(_, __): @pytest.mark.parametrize("log_level", [logging.WARNING, logging.DEBUG]) @pytest.mark.xdist_group(name="caplog") async def test_protocol_logging( - mocker, caplog, log_level, protocol_class, transport_class, encryption_class -): + mocker: MockerFixture, + caplog: pytest.LogCaptureFixture, + log_level: int, + protocol_class: type[BaseProtocol], + transport_class: type[XorTransport], + encryption_class, +) -> None: caplog.set_level(log_level) logging.getLogger("kasa").setLevel(log_level) encrypted = encryption_class.encrypt('{"great":"success"}')[ transport_class.BLOCK_SIZE : ] - async def _mock_read(byte_count): + async def _mock_read(byte_count: int): nonlocal encrypted if byte_count == transport_class.BLOCK_SIZE: return struct.pack(">I", len(encrypted)) @@ -529,7 +570,7 @@ async def _mock_read(byte_count): return encrypted raise ValueError(f"No mock for {byte_count}") - def aio_mock_writer(_, __): + def aio_mock_writer(_: object, __: object): reader = mocker.patch("asyncio.StreamReader") writer = mocker.patch("asyncio.StreamWriter") mocker.patch.object(reader, "readexactly", _mock_read) @@ -561,13 +602,17 @@ def aio_mock_writer(_, __): ) @pytest.mark.parametrize("custom_port", [123, None]) async def test_protocol_custom_port( - mocker, custom_port, protocol_class, transport_class, encryption_class -): + mocker: MockerFixture, + custom_port: int | None, + protocol_class: type[BaseProtocol], + transport_class: type[XorTransport], + encryption_class, +) -> None: encrypted = encryption_class.encrypt('{"great":"success"}')[ transport_class.BLOCK_SIZE : ] - async def _mock_read(byte_count): + async def _mock_read(byte_count: int): nonlocal encrypted if byte_count == transport_class.BLOCK_SIZE: return struct.pack(">I", len(encrypted)) @@ -575,7 +620,7 @@ async def _mock_read(byte_count): return encrypted raise ValueError(f"No mock for {byte_count}") - def aio_mock_writer(_, port): + def aio_mock_writer(_: object, port: int): reader = mocker.patch("asyncio.StreamReader") writer = mocker.patch("asyncio.StreamWriter") if custom_port is None: @@ -601,7 +646,7 @@ def aio_mock_writer(_, port): "decrypt_class", [_deprecated_TPLinkSmartHomeProtocol, XorEncryption], ) -def test_encrypt(encrypt_class, decrypt_class): +def test_encrypt(encrypt_class, decrypt_class) -> None: d = json.dumps({"foo": 1, "bar": 2}) encrypted = encrypt_class.encrypt(d) # encrypt adds a 4 byte header @@ -613,7 +658,7 @@ def test_encrypt(encrypt_class, decrypt_class): "encrypt_class", [_deprecated_TPLinkSmartHomeProtocol, XorEncryption], ) -def test_encrypt_unicode(encrypt_class): +def test_encrypt_unicode(encrypt_class) -> None: d = "{'snowman': '\u2603'}" e = bytes( @@ -650,7 +695,7 @@ def test_encrypt_unicode(encrypt_class): "decrypt_class", [_deprecated_TPLinkSmartHomeProtocol, XorEncryption], ) -def test_decrypt_unicode(decrypt_class): +def test_decrypt_unicode(decrypt_class) -> None: e = bytes( [ 208, @@ -679,7 +724,7 @@ def test_decrypt_unicode(decrypt_class): assert d == decrypt_class.decrypt(e) -def _get_subclasses(of_class): +def _get_subclasses(of_class: type): package = sys.modules["kasa"] subclasses = set() for _, modname, _ in pkgutil.iter_modules(package.__path__): @@ -698,11 +743,13 @@ def _get_subclasses(of_class): @pytest.mark.parametrize( "class_name_obj", _get_subclasses(BaseProtocol), ids=lambda t: t[0] ) -def test_protocol_init_signature(class_name_obj): +def test_protocol_init_signature(class_name_obj: tuple[str, type]) -> None: if class_name_obj[0].startswith("_"): pytest.skip("Skipping internal protocols") return - params = list(inspect.signature(class_name_obj[1].__init__).parameters.values()) + params = list( + inspect.signature(class_name_obj[1].__init__).parameters.values() # type: ignore[misc] + ) assert len(params) == 2 assert params[0].name == "self" @@ -714,8 +761,10 @@ def test_protocol_init_signature(class_name_obj): @pytest.mark.parametrize( "class_name_obj", _get_subclasses(BaseTransport), ids=lambda t: t[0] ) -def test_transport_init_signature(class_name_obj): - params = list(inspect.signature(class_name_obj[1].__init__).parameters.values()) +def test_transport_init_signature(class_name_obj: tuple[str, type]) -> None: + params = list( + inspect.signature(class_name_obj[1].__init__).parameters.values() # type: ignore[misc] + ) assert len(params) == 2 assert params[0].name == "self" @@ -765,8 +814,13 @@ def test_transport_init_signature(class_name_obj): ], ) async def test_transport_credentials_hash( - mocker, transport_class, login_version, expected_hash, credentials, expected_blank -): + mocker: MockerFixture, + transport_class: type[BaseTransport], + login_version: int | None, + expected_hash: str | None, + credentials: Credentials | None, + expected_blank: bool, +) -> None: """Test that the actual hashing doesn't break and empty credential returns an empty hash.""" host = "127.0.0.1" @@ -788,7 +842,9 @@ async def test_transport_credentials_hash( "transport_class", [AesTransport, KlapTransport, KlapTransportV2, XorTransport], ) -async def test_transport_credentials_hash_from_config(mocker, transport_class): +async def test_transport_credentials_hash_from_config( + mocker: MockerFixture, transport_class: type[BaseTransport] +) -> None: """Test that credentials_hash provided via config sets correctly.""" host = "127.0.0.1" @@ -828,8 +884,12 @@ async def test_transport_credentials_hash_from_config(mocker, transport_class): ids=("_deprecated_TPLinkSmartHomeProtocol", "IotProtocol-XorTransport"), ) async def test_protocol_will_retry_on_connect( - mocker, protocol_class, transport_class, error, retry_expectation -): + mocker: MockerFixture, + protocol_class: type[BaseProtocol], + transport_class: type[XorTransport], + error: Exception, + retry_expectation: bool, +) -> None: retry_count = 2 conn = mocker.patch("asyncio.open_connection", side_effect=error) config = DeviceConfig("127.0.0.1") @@ -860,13 +920,17 @@ async def test_protocol_will_retry_on_connect( ids=("_deprecated_TPLinkSmartHomeProtocol", "IotProtocol-XorTransport"), ) async def test_protocol_will_retry_on_write( - mocker, protocol_class, transport_class, error, retry_expectation -): + mocker: MockerFixture, + protocol_class: type[BaseProtocol], + transport_class: type[XorTransport], + error: Exception, + retry_expectation: bool, +) -> None: retry_count = 2 writer = mocker.patch("asyncio.StreamWriter") write_mock = mocker.patch.object(writer, "write", side_effect=error) - def aio_mock_writer(_, __): + def aio_mock_writer(_: object, __: object): nonlocal writer reader = mocker.patch("asyncio.StreamReader") @@ -885,9 +949,9 @@ def aio_mock_writer(_, __): assert write_mock.call_count == expected_call_count -def test_deprecated_protocol(): +def test_deprecated_protocol() -> None: with pytest.deprecated_call(): - from kasa import TPLinkSmartHomeProtocol + from kasa import TPLinkSmartHomeProtocol # type: ignore[attr-defined] with pytest.raises(KasaException, match="host or transport must be supplied"): proto = TPLinkSmartHomeProtocol() @@ -898,7 +962,9 @@ def test_deprecated_protocol(): @device_iot @pytest.mark.xdist_group(name="caplog") -async def test_iot_queries_redaction(dev: IotDevice, caplog: pytest.LogCaptureFixture): +async def test_iot_queries_redaction( + dev: IotDevice, caplog: pytest.LogCaptureFixture +) -> None: """Test query sensitive info redaction.""" if isinstance(dev.protocol._transport, FakeIotTransport): device_id = "123456789ABCDEF" @@ -932,7 +998,7 @@ async def test_iot_queries_redaction(dev: IotDevice, caplog: pytest.LogCaptureFi assert "REDACTED_" + device_id[9::] in caplog.text -async def test_redact_data(): +async def test_redact_data() -> None: """Test redact data function.""" data = { "device_id": "123456789ABCDEF", diff --git a/tests/protocols/test_smartprotocol.py b/tests/protocols/test_smartprotocol.py index 514926353..9ccbc3cb2 100644 --- a/tests/protocols/test_smartprotocol.py +++ b/tests/protocols/test_smartprotocol.py @@ -26,7 +26,9 @@ ERRORS = [e for e in SmartErrorCode if e != 0] -async def test_smart_queries(dummy_protocol, mocker: pytest_mock.MockerFixture): +async def test_smart_queries( + dummy_protocol: SmartProtocol, mocker: pytest_mock.MockerFixture +) -> None: mock_response = {"result": {"great": "success"}, "error_code": 0} mocker.patch.object(dummy_protocol._transport, "send", return_value=mock_response) @@ -42,7 +44,9 @@ async def test_smart_queries(dummy_protocol, mocker: pytest_mock.MockerFixture): @pytest.mark.parametrize("error_code", ERRORS, ids=lambda e: e.name) -async def test_smart_device_errors(dummy_protocol, mocker, error_code): +async def test_smart_device_errors( + dummy_protocol: SmartProtocol, mocker: MockerFixture, error_code: SmartErrorCode +) -> None: mock_response = {"result": {"great": "success"}, "error_code": error_code.value} send_mock = mocker.patch.object( @@ -59,8 +63,11 @@ async def test_smart_device_errors(dummy_protocol, mocker, error_code): @pytest.mark.parametrize("error_code", [-13333, 13333]) @pytest.mark.xdist_group(name="caplog") async def test_smart_device_unknown_errors( - dummy_protocol, mocker, error_code, caplog: pytest.LogCaptureFixture -): + dummy_protocol: SmartProtocol, + mocker: MockerFixture, + error_code: int, + caplog: pytest.LogCaptureFixture, +) -> None: """Test handling of unknown error codes.""" mock_response = {"result": {"great": "success"}, "error_code": error_code} @@ -78,8 +85,8 @@ async def test_smart_device_unknown_errors( @pytest.mark.parametrize("error_code", ERRORS, ids=lambda e: e.name) async def test_smart_device_errors_in_multiple_request( - dummy_protocol, mocker, error_code -): + dummy_protocol: SmartProtocol, mocker: MockerFixture, error_code: SmartErrorCode +) -> None: mock_request = { "foobar1": {"foo": "bar", "bar": "foo"}, "foobar2": {"foo": "bar", "bar": "foo"}, @@ -113,10 +120,13 @@ async def test_smart_device_errors_in_multiple_request( @pytest.mark.parametrize("request_size", [1, 3, 5, 10]) @pytest.mark.parametrize("batch_size", [1, 2, 3, 4, 5]) async def test_smart_device_multiple_request( - dummy_protocol, mocker, request_size, batch_size -): + dummy_protocol: SmartProtocol, + mocker: MockerFixture, + request_size: int, + batch_size: int, +) -> None: requests = {} - mock_response = { + mock_response: dict = { "result": {"responses": []}, "error_code": 0, } @@ -138,8 +148,8 @@ async def test_smart_device_multiple_request( async def test_smart_device_multiple_request_json_decode_failure( - dummy_protocol, mocker -): + dummy_protocol: SmartProtocol, mocker: MockerFixture +) -> None: """Test the logic to disable multiple requests on JSON_DECODE_FAIL_ERROR.""" requests = {} mock_responses = [] @@ -169,8 +179,8 @@ async def test_smart_device_multiple_request_json_decode_failure( async def test_smart_device_multiple_request_json_decode_failure_twice( - dummy_protocol, mocker -): + dummy_protocol: SmartProtocol, mocker: MockerFixture +) -> None: """Test the logic to disable multiple requests on JSON_DECODE_FAIL_ERROR.""" requests = {} @@ -196,8 +206,8 @@ async def test_smart_device_multiple_request_json_decode_failure_twice( async def test_smart_device_multiple_request_non_json_decode_failure( - dummy_protocol, mocker -): + dummy_protocol: SmartProtocol, mocker: MockerFixture +) -> None: """Test the logic to disable multiple requests on JSON_DECODE_FAIL_ERROR. Ensure other exception types behave as expected. @@ -225,7 +235,9 @@ async def test_smart_device_multiple_request_non_json_decode_failure( assert send_mock.call_count == 1 -async def test_childdevicewrapper_unwrapping(dummy_protocol, mocker): +async def test_childdevicewrapper_unwrapping( + dummy_protocol: SmartProtocol, mocker: MockerFixture +) -> None: """Test that responseData gets unwrapped correctly.""" wrapped_protocol = _ChildProtocolWrapper("dummyid", dummy_protocol) mock_response = {"error_code": 0, "result": {"responseData": {"error_code": 0}}} @@ -235,7 +247,9 @@ async def test_childdevicewrapper_unwrapping(dummy_protocol, mocker): assert res == {"foobar": None} -async def test_childdevicewrapper_unwrapping_with_payload(dummy_protocol, mocker): +async def test_childdevicewrapper_unwrapping_with_payload( + dummy_protocol: SmartProtocol, mocker: MockerFixture +) -> None: wrapped_protocol = _ChildProtocolWrapper("dummyid", dummy_protocol) mock_response = { "error_code": 0, @@ -246,7 +260,9 @@ async def test_childdevicewrapper_unwrapping_with_payload(dummy_protocol, mocker assert res == {"foobar": {"bar": "bar"}} -async def test_childdevicewrapper_error(dummy_protocol, mocker): +async def test_childdevicewrapper_error( + dummy_protocol: SmartProtocol, mocker: MockerFixture +) -> None: """Test that errors inside the responseData payload cause an exception.""" wrapped_protocol = _ChildProtocolWrapper("dummyid", dummy_protocol) mock_response = {"error_code": 0, "result": {"responseData": {"error_code": -1001}}} @@ -256,7 +272,9 @@ async def test_childdevicewrapper_error(dummy_protocol, mocker): await wrapped_protocol.query(DUMMY_QUERY) -async def test_childdevicewrapper_unwrapping_multiplerequest(dummy_protocol, mocker): +async def test_childdevicewrapper_unwrapping_multiplerequest( + dummy_protocol: SmartProtocol, mocker: MockerFixture +) -> None: """Test that unwrapping multiplerequest works correctly.""" mock_response = { "error_code": 0, @@ -285,7 +303,9 @@ async def test_childdevicewrapper_unwrapping_multiplerequest(dummy_protocol, moc assert resp == {"get_device_info": {"foo": "bar"}, "second_command": {"bar": "foo"}} -async def test_childdevicewrapper_multiplerequest_error(dummy_protocol, mocker): +async def test_childdevicewrapper_multiplerequest_error( + dummy_protocol: SmartProtocol, mocker: MockerFixture +) -> None: """Test that errors inside multipleRequest response of responseData raise an exception.""" mock_response = { "error_code": 0, @@ -313,7 +333,9 @@ async def test_childdevicewrapper_multiplerequest_error(dummy_protocol, mocker): @pytest.mark.parametrize("list_sum", [5, 10, 30]) @pytest.mark.parametrize("batch_size", [1, 2, 3, 50]) -async def test_smart_protocol_lists_single_request(mocker, list_sum, batch_size): +async def test_smart_protocol_lists_single_request( + mocker: MockerFixture, list_sum: int, batch_size: int +) -> None: child_device_list = [{"foo": i} for i in range(list_sum)] response = { "get_child_device_list": { @@ -341,7 +363,9 @@ async def test_smart_protocol_lists_single_request(mocker, list_sum, batch_size) @pytest.mark.parametrize("list_sum", [5, 10, 30]) @pytest.mark.parametrize("batch_size", [1, 2, 3, 50]) -async def test_smart_protocol_lists_multiple_request(mocker, list_sum, batch_size): +async def test_smart_protocol_lists_multiple_request( + mocker: MockerFixture, list_sum: int, batch_size: int +) -> None: child_list = [{"foo": i} for i in range(list_sum)] response = { "get_child_device_list": { @@ -376,7 +400,9 @@ async def test_smart_protocol_lists_multiple_request(mocker, list_sum, batch_siz @pytest.mark.parametrize("list_sum", [5, 10, 30]) @pytest.mark.parametrize("batch_size", [1, 2, 3, 50]) -async def test_smartcam_protocol_list_request(mocker, list_sum, batch_size): +async def test_smartcam_protocol_list_request( + mocker: MockerFixture, list_sum: int, batch_size: int +) -> None: """Test smartcam protocol list handling for lists.""" child_list = [{"foo": i} for i in range(list_sum)] @@ -414,7 +440,9 @@ async def test_smartcam_protocol_list_request(mocker, list_sum, batch_size): assert resp == response -async def test_incomplete_list(mocker, caplog): +async def test_incomplete_list( + mocker: MockerFixture, caplog: pytest.LogCaptureFixture +) -> None: """Test for handling incomplete lists returned from queries.""" info = { "get_preset_rules": { @@ -464,7 +492,7 @@ async def test_incomplete_list(mocker, caplog): @pytest.mark.xdist_group(name="caplog") async def test_smart_queries_redaction( dev: SmartDevice, caplog: pytest.LogCaptureFixture -): +) -> None: """Test query sensitive info redaction.""" if isinstance(dev.protocol._transport, FakeSmartTransport): device_id = "123456789ABCDEF" @@ -495,7 +523,7 @@ async def test_smart_queries_redaction( async def test_no_method_returned_multiple( mocker: MockerFixture, caplog: pytest.LogCaptureFixture -): +) -> None: """Test protocol handles multiple requests that don't return the method.""" req = { "getDeviceInfo": {"device_info": {"name": ["basic_info", "info"]}}, @@ -540,7 +568,7 @@ async def test_no_method_returned_multiple( async def test_no_multiple_methods( mocker: MockerFixture, caplog: pytest.LogCaptureFixture -): +) -> None: """Test protocol sends NO_MULTI methods as single call.""" req = { "getDeviceInfo": {"device_info": {"name": ["basic_info", "info"]}}, diff --git a/tests/smartcam/modules/test_alarm.py b/tests/smartcam/modules/test_alarm.py index 0a176650f..48b249c25 100644 --- a/tests/smartcam/modules/test_alarm.py +++ b/tests/smartcam/modules/test_alarm.py @@ -16,7 +16,7 @@ @hub_smartcam -async def test_alarm(dev: Device): +async def test_alarm(dev: Device) -> None: """Test device alarm.""" alarm = dev.modules.get(Module.Alarm) assert alarm @@ -83,7 +83,7 @@ async def test_alarm(dev: Device): @hub_smartcam -async def test_alarm_invalid_setters(dev: Device): +async def test_alarm_invalid_setters(dev: Device) -> None: """Test device alarm invalid setter values.""" alarm = dev.modules.get(Module.Alarm) assert alarm @@ -105,7 +105,7 @@ async def test_alarm_invalid_setters(dev: Device): @hub_smartcam -async def test_alarm_features(dev: Device): +async def test_alarm_features(dev: Device) -> None: """Test device alarm features.""" alarm = dev.modules.get(Module.Alarm) assert alarm diff --git a/tests/smartcam/modules/test_battery.py b/tests/smartcam/modules/test_battery.py index 723cb22b3..9891298ad 100644 --- a/tests/smartcam/modules/test_battery.py +++ b/tests/smartcam/modules/test_battery.py @@ -17,7 +17,7 @@ @battery_smartcam -async def test_battery(dev: Device): +async def test_battery(dev: Device) -> None: """Test device battery.""" battery = dev.modules.get(SmartCamModule.SmartCamBattery) assert battery @@ -46,7 +46,9 @@ async def test_battery(dev: Device): ("12.3", 12.3), # sanity: happy path ], ) -async def test_battery_temperature_edge_cases(dev: Device, raw, expected): +async def test_battery_temperature_edge_cases( + dev: Device, raw: str | None, expected: float | None +) -> None: battery = dev.modules.get(SmartCamModule.SmartCamBattery) assert battery @@ -63,7 +65,9 @@ async def test_battery_temperature_edge_cases(dev: Device, raw, expected): ("12000", 12.0), # sanity: parses string -> float(...) / 1000 ], ) -async def test_battery_voltage_edge_cases(dev: Device, voltage_raw, expected_v): +async def test_battery_voltage_edge_cases( + dev: Device, voltage_raw: str | None, expected_v: float | None +) -> None: battery = dev.modules.get(SmartCamModule.SmartCamBattery) assert battery @@ -82,7 +86,9 @@ async def test_battery_voltage_edge_cases(dev: Device, voltage_raw, expected_v): ("NO", False), # sanity: string normalization path ], ) -async def test_battery_charging_edge_cases(dev: Device, charging_raw, expected): +async def test_battery_charging_edge_cases( + dev: Device, charging_raw: bool | str | None, expected: bool +) -> None: battery = dev.modules.get(SmartCamModule.SmartCamBattery) assert battery diff --git a/tests/smartcam/modules/test_camera.py b/tests/smartcam/modules/test_camera.py index d668f9f46..d8762ef7e 100644 --- a/tests/smartcam/modules/test_camera.py +++ b/tests/smartcam/modules/test_camera.py @@ -20,7 +20,7 @@ @device_smartcam -async def test_state(dev: Device): +async def test_state(dev: Device) -> None: if dev.device_type is DeviceType.Hub: pytest.skip("Hubs cannot be switched on and off") @@ -31,7 +31,7 @@ async def test_state(dev: Device): @not_child_camera_smartcam -async def test_stream_rtsp_url(dev: Device): +async def test_stream_rtsp_url(dev: Device) -> None: camera_module = dev.modules.get(Module.Camera) assert camera_module @@ -91,7 +91,7 @@ async def test_stream_rtsp_url(dev: Device): @not_child_camera_smartcam -async def test_onvif_url(dev: Device): +async def test_onvif_url(dev: Device) -> None: """Test the onvif url.""" camera_module = dev.modules.get(Module.Camera) assert camera_module diff --git a/tests/smartcam/modules/test_childsetup.py b/tests/smartcam/modules/test_childsetup.py index 090ea0338..dfef66ff7 100644 --- a/tests/smartcam/modules/test_childsetup.py +++ b/tests/smartcam/modules/test_childsetup.py @@ -15,7 +15,7 @@ @childsetup -async def test_childsetup_features(dev: Device): +async def test_childsetup_features(dev: Device) -> None: """Test the exposed features.""" cs = dev.modules[Module.ChildSetup] @@ -27,7 +27,7 @@ async def test_childsetup_features(dev: Device): @childsetup async def test_childsetup_pair( dev: Device, mocker: MockerFixture, caplog: pytest.LogCaptureFixture -): +) -> None: """Test device pairing.""" caplog.set_level(logging.INFO) mock_query_helper = mocker.spy(dev, "_query_helper") @@ -70,7 +70,7 @@ async def test_childsetup_pair( @childsetup async def test_childsetup_unpair( dev: Device, mocker: MockerFixture, caplog: pytest.LogCaptureFixture -): +) -> None: """Test unpair.""" mock_query_helper = mocker.spy(dev, "_query_helper") DUMMY_ID = "dummy_id" diff --git a/tests/smartcam/modules/test_detections.py b/tests/smartcam/modules/test_detections.py index c4659f7b1..b4a664310 100644 --- a/tests/smartcam/modules/test_detections.py +++ b/tests/smartcam/modules/test_detections.py @@ -25,8 +25,8 @@ class Detection(NamedTuple): def parametrize_detection( *, model_filter=None, - protocol_filter=None, - fixture_name="dev", + protocol_filter: set[str] | None = None, + fixture_name: str = "dev", extra_params_names: list[str], extra_params_values: list[Detection], ): @@ -139,7 +139,7 @@ def parametrize_detection( @params_detections async def test_detections( dev: Device, module: ModuleName[DetectionModule], feature_name: str -): +) -> None: detection = dev.modules.get(module) assert detection diff --git a/tests/smartcam/modules/test_pantilt.py b/tests/smartcam/modules/test_pantilt.py index fb58dc66a..626ac3337 100644 --- a/tests/smartcam/modules/test_pantilt.py +++ b/tests/smartcam/modules/test_pantilt.py @@ -15,7 +15,7 @@ @pantilt -async def test_pantilt_presets(dev: Device, mocker: MockerFixture): +async def test_pantilt_presets(dev: Device, mocker: MockerFixture) -> None: """Test PanTilt module preset functionality.""" pantilt_mod = dev.modules.get(Module.PanTilt) assert pantilt_mod is not None @@ -42,7 +42,7 @@ async def test_pantilt_presets(dev: Device, mocker: MockerFixture): @pantilt -async def test_pantilt_save_preset(dev: Device, mocker: MockerFixture): +async def test_pantilt_save_preset(dev: Device, mocker: MockerFixture) -> None: """Test PanTilt save_preset functionality.""" pantilt_mod = dev.modules.get(Module.PanTilt) assert pantilt_mod is not None @@ -62,7 +62,7 @@ async def test_pantilt_save_preset(dev: Device, mocker: MockerFixture): @pantilt -async def test_pantilt_invalid_preset(dev: Device, mocker: MockerFixture): +async def test_pantilt_invalid_preset(dev: Device, mocker: MockerFixture) -> None: """Test set_preset with invalid preset name raises ValueError.""" pantilt_mod = dev.modules.get(Module.PanTilt) assert pantilt_mod is not None @@ -81,7 +81,7 @@ async def test_pantilt_invalid_preset(dev: Device, mocker: MockerFixture): @pantilt -async def test_pantilt_move(dev: Device, mocker: MockerFixture): +async def test_pantilt_move(dev: Device, mocker: MockerFixture) -> None: """Test PanTilt move commands.""" pantilt_mod = dev.modules.get(Module.PanTilt) assert pantilt_mod is not None @@ -103,7 +103,7 @@ async def test_pantilt_move(dev: Device, mocker: MockerFixture): @pantilt -async def test_pantilt_goto_preset(dev: Device, mocker: MockerFixture): +async def test_pantilt_goto_preset(dev: Device, mocker: MockerFixture) -> None: """Test PanTilt goto_preset command.""" pantilt_mod = dev.modules.get(Module.PanTilt) assert pantilt_mod is not None @@ -119,7 +119,7 @@ async def test_pantilt_goto_preset(dev: Device, mocker: MockerFixture): @pantilt -async def test_pantilt_get_presets(dev: Device, mocker: MockerFixture): +async def test_pantilt_get_presets(dev: Device, mocker: MockerFixture) -> None: """Test PanTilt get_presets command.""" pantilt_mod = dev.modules.get(Module.PanTilt) assert pantilt_mod is not None @@ -135,7 +135,7 @@ async def test_pantilt_get_presets(dev: Device, mocker: MockerFixture): @pantilt -async def test_pantilt_set_preset_by_id(dev: Device, mocker: MockerFixture): +async def test_pantilt_set_preset_by_id(dev: Device, mocker: MockerFixture) -> None: """Test set_preset with preset ID instead of name.""" pantilt_mod = dev.modules.get(Module.PanTilt) assert pantilt_mod is not None @@ -160,7 +160,7 @@ async def test_pantilt_set_preset_by_id(dev: Device, mocker: MockerFixture): @pantilt -async def test_pantilt_set_preset_not_found(dev: Device, mocker: MockerFixture): +async def test_pantilt_set_preset_not_found(dev: Device, mocker: MockerFixture) -> None: """Test set_preset with non-existent preset returns empty dict.""" pantilt_mod = dev.modules.get(Module.PanTilt) assert pantilt_mod is not None @@ -177,7 +177,7 @@ async def test_pantilt_set_preset_not_found(dev: Device, mocker: MockerFixture): @pantilt -async def test_pantilt_step_features(dev: Device, mocker: MockerFixture): +async def test_pantilt_step_features(dev: Device, mocker: MockerFixture) -> None: """Test pan/tilt step features.""" pantilt_mod = dev.modules.get(Module.PanTilt) assert pantilt_mod is not None @@ -200,7 +200,7 @@ async def test_pantilt_step_features(dev: Device, mocker: MockerFixture): @pantilt -async def test_pantilt_no_presets_in_data(dev: Device, mocker: MockerFixture): +async def test_pantilt_no_presets_in_data(dev: Device, mocker: MockerFixture) -> None: """Test _presets returns empty dict when no preset data.""" pantilt_mod = dev.modules.get(Module.PanTilt) assert pantilt_mod is not None diff --git a/tests/smartcam/test_smartcamdevice.py b/tests/smartcam/test_smartcamdevice.py index 58ab2fd98..e2c46161f 100644 --- a/tests/smartcam/test_smartcamdevice.py +++ b/tests/smartcam/test_smartcamdevice.py @@ -17,7 +17,7 @@ @device_smartcam -async def test_state(dev: Device): +async def test_state(dev: Device) -> None: if dev.device_type is DeviceType.Hub: pytest.skip("Hubs cannot be switched on and off") @@ -38,7 +38,7 @@ async def test_state(dev: Device): @device_smartcam -async def test_alias(dev: Device): +async def test_alias(dev: Device) -> None: test_alias = "TEST1234" original = dev.alias @@ -53,7 +53,7 @@ async def test_alias(dev: Device): @hub_smartcam -async def test_hub(dev: Device): +async def test_hub(dev: Device) -> None: assert dev.children for child in dev.children: assert child.modules @@ -65,7 +65,7 @@ async def test_hub(dev: Device): @device_smartcam -async def test_wifi_scan(dev: SmartCamDevice): +async def test_wifi_scan(dev: SmartCamDevice) -> None: fake_scan_data = { "scanApList": { "onboarding": { @@ -97,7 +97,7 @@ async def test_wifi_scan(dev: SmartCamDevice): @device_smartcam -async def test_wifi_join_success_and_errors(dev: SmartCamDevice): +async def test_wifi_join_success_and_errors(dev: SmartCamDevice) -> None: dev._networks = [ type( "WifiNetwork", @@ -154,7 +154,7 @@ async def test_wifi_join_success_and_errors(dev: SmartCamDevice): @device_smartcam -async def test_device_time(dev: Device, freezer: FrozenDateTimeFactory): +async def test_device_time(dev: Device, freezer: FrozenDateTimeFactory) -> None: """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 @@ -165,7 +165,7 @@ async def test_device_time(dev: Device, freezer: FrozenDateTimeFactory): @device_smartcam -async def test_wifi_join_typeerror_on_non_rsa_key(dev: SmartCamDevice): +async def test_wifi_join_typeerror_on_non_rsa_key(dev: SmartCamDevice) -> None: dev._networks = [ type( "WifiNetwork",