drop compatibility with python3.9; upgrade alpine base image from v3.18.4 to v3.19.8

.github/workflows/python.yml

@@ -17,7 +17,7 @@ jobs:
     runs-on: ubuntu-24.04
     strategy:
       matrix:
-        python-version: ['3.9']
+        python-version: ['3.10']
     steps:
     - uses: actions/checkout@v5
     - uses: actions/setup-python@v6
@@ -34,7 +34,6 @@ jobs:
     strategy:
       matrix:
         python-version:
-        - '3.9'
         - '3.10'
         - '3.11'
       fail-fast: false

.pylintrc

@@ -4,7 +4,6 @@ load-plugins=pylint.extensions.check_elif,
              pylint.extensions.comparison_placement,
              pylint.extensions.confusing_elif,
              pylint.extensions.consider_ternary_expression,
-             pylint.extensions.emptystring,
              pylint.extensions.eq_without_hash,
              pylint.extensions.for_any_all,
              pylint.extensions.mccabe,
@@ -18,9 +17,7 @@ load-plugins=pylint.extensions.check_elif,
 
 [MESSAGES CONTROL]
 
-disable=consider-alternative-union-syntax, # requires python>=3.10
-        deprecated-typing-alias, # requires python>=3.9, e.g. for dict[...]
-        missing-function-docstring,
+disable=missing-function-docstring,
         missing-module-docstring
 
 [DESIGN]

CHANGELOG.md

@@ -20,14 +20,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   (bluepy-helper replaced with bleak)
 - replaced [paho-mqtt](https://github.com/eclipse/paho.mqtt.python)
   with its async wrapper [aiomqtt](https://github.com/sbtinstruments/aiomqtt)
-- container image: upgraded alpine base image from v3.13.1 to v3.18.4
+- container image: upgraded alpine base image from v3.13.1 to v3.19.8
 
 ### Removed
 - command-line option `--mqtt-enable-tls` (TLS now enabled by default)
 - compatibility with `python3.7`
-- compatibility with `python3.8`
-  (pySwitchbot v0.17.2 added constraint `bleak-retry-connector>=1.1.1`
-  requiring `python>=3.9`)
+- compatibility with `python3.8` (security support ended on 2024-10-07)
+- compatibility with `python3.9` (security support will end on 2025-10-31)
 
 ## [3.3.1] - 2022-08-31
 ### Fixed

Dockerfile

@@ -2,7 +2,7 @@
 
 # not using python:3.*-alpine cause glib-dev package depends on python3
 # https://pkgs.alpinelinux.org/package/v3.18/main/aarch64/glib-dev
-ARG BASE_IMAGE=docker.io/alpine:3.18.4
+ARG BASE_IMAGE=docker.io/alpine:3.19.8
 ARG SOURCE_DIR_PATH=/switchbot-mqtt
 
 
@@ -26,7 +26,7 @@ RUN apk add --no-cache \
     && adduser -S build
 
 USER build
-RUN pip3 install --user --no-cache-dir pipenv==2023.6.18
+RUN pip3 install --user --break-system-packages --no-cache-dir pipenv==2025.0.4
 
 ARG SOURCE_DIR_PATH
 COPY --chown=build:nobody Pipfile Pipfile.lock $SOURCE_DIR_PATH/

Pipfile

@@ -6,11 +6,6 @@ name = "pypi"
 [packages]
 switchbot-mqtt = {editable = true, path = "."}
 
-# remove marker breaking pipeline on python>3.9
-typing-extensions = {markers = ""}
-# remove marker breaking pipeline on python>3.10
-async-timeout = {markers = ""}
-
 [dev-packages]
 black = "*"
 mypy = "*"
@@ -20,11 +15,10 @@ pytest-asyncio = "*"
 pytest-cov = "*"
 
 # python3.10 compatibility
-# >   File "[...]/lib/python3.10/site-packages/mypy/main.py", line 11, in <module>
-# >     from typing_extensions import Final, NoReturn
-# > ModuleNotFoundError: No module named 'typing_extensions'
-typing-extensions = {markers = ""}
-# python<3.11 compatibility
+# >   File "…/python3.10/site-packages/pytest_asyncio/plugin.py", line 60, in …
+# >     from backports.asyncio.runner import Runner
+# > ModuleNotFoundError: No module named 'backports'
+"backports.asyncio.runner" = {markers = "python_version < '3.11'"}
 # >    File "[...]/lib/python3.10/site-packages/_pytest/_code/code.py", line 60, in <module>
 # >     from exceptiongroup import BaseExceptionGroup
 # > ModuleNotFoundError: No module named 'exceptiongroup'
@@ -33,15 +27,6 @@ exceptiongroup = {markers = "python_version < '3.11'"}
 # >     import tomli as tomllib
 # > ModuleNotFoundError: No module named 'tomli'
 tomli = {markers = "python_version < '3.11'"}
-# >   File "[...]/lib/python3.10/site-packages/astroid/decorators.py", line 16, in <module>
-# >     import wrapt
-# > ModuleNotFoundError: No module named 'wrapt'
-wrapt = "*"
-# remove `"markers": "python_version >= '3.11'"` to workaround:
-# >   File "[...]/lib/python3.7/site-packages/pylint/lint/parallel.py", line 13, in <module>
-# >     import dill
-# > ModuleNotFoundError: No module named 'dill'
-dill = {markers = "python_version >= '0'"}
 
 [requires]
 python_version = "3"

setup.py

@@ -65,7 +65,6 @@ setuptools.setup(
         "License :: OSI Approved :: GNU General Public License v3 or later (GPLv3+)",
         "Operating System :: POSIX :: Linux",
         # .github/workflows/python.yml
-        "Programming Language :: Python :: 3.9",
         "Programming Language :: Python :: 3.10",
         "Programming Language :: Python :: 3.11",
         "Topic :: Home Automation",
@@ -74,11 +73,9 @@ setuptools.setup(
     # >=3.6 variable type hints, f-strings, typing.Collection & * to force keyword-only arguments
     # >=3.7 postponed evaluation of type annotations (PEP563) & asyncio.run
     # >=3.8 unittest.mock.AsyncMock
-    # <=3.8 untested cause pySwitchbot v0.17.2 added constraint bleak-retry-connector>=1.1.1
-    #       requiring python>=3.9
-    # https://web.archive.org/web/20231104212919/https://github.com/Danielhiversen/pySwitchbot/compare/0.17.1..0.17.2
-    # https://web.archive.org/web/20231104212930/https://pypi.org/project/bleak-retry-connector/1.1.1/
-    python_requires=">=3.9",
+    # >=3.9 type hints dict[…], list[…] & tuple[…] (PEP585)
+    # >=3.10 union types as X | Y (PEP604)
+    python_requires=">=3.10",
     install_requires=[
         "bleak<0.22",
         # v0.14.0 replaced bluepy with bleak

switchbot_mqtt/__init__.py

@@ -16,10 +16,10 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
+import collections.abc
 import logging
 import socket
 import ssl
-import typing
 
 import aiomqtt
 
@@ -37,10 +37,10 @@ _MQTT_LAST_WILL_PAYLOAD = "offline"
 async def _listen(
     *,
     mqtt_client: aiomqtt.Client,
-    topic_callbacks: typing.Iterable[typing.Tuple[str, typing.Callable]],
+    topic_callbacks: collections.abc.Iterable[tuple[str, collections.abc.Callable]],
     mqtt_topic_prefix: str,
     retry_count: int,
-    device_passwords: typing.Dict[str, str],
+    device_passwords: dict[str, str],
     fetch_device_info: bool,
 ) -> None:
     async with mqtt_client.messages() as messages:
@@ -88,11 +88,11 @@ async def _run(  # pylint: disable=too-many-arguments
     mqtt_host: str,
     mqtt_port: int,
     mqtt_disable_tls: bool,
-    mqtt_username: typing.Optional[str],
-    mqtt_password: typing.Optional[str],
+    mqtt_username: str | None,
+    mqtt_password: str | None,
     mqtt_topic_prefix: str,
     retry_count: int,
-    device_passwords: typing.Dict[str, str],
+    device_passwords: dict[str, str],
     fetch_device_info: bool,
 ) -> None:
     _LOGGER.info(
@@ -119,7 +119,7 @@ async def _run(  # pylint: disable=too-many-arguments
         ),
     ) as mqtt_client:
         _log_mqtt_connected(mqtt_client=mqtt_client)
-        topic_callbacks: typing.List[typing.Tuple[str, typing.Callable]] = []
+        topic_callbacks: list[tuple[str, collections.abc.Callable]] = []
         for actor_class in (_ButtonAutomator, _CurtainMotor):
             async for topic, callback in actor_class.mqtt_subscribe(
                 mqtt_client=mqtt_client,

switchbot_mqtt/_actors/__init__.py

@@ -16,8 +16,8 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
+import collections.abc
 import logging
-import typing
 
 import aiomqtt
 import bleak
@@ -62,7 +62,7 @@ class _ButtonAutomator(_MQTTControlledActor):
         *,
         device: bleak.backends.device.BLEDevice,
         retry_count: int,
-        password: typing.Optional[str],
+        password: str | None,
     ) -> None:
         self.__device = switchbot.Switchbot(
             device=device, password=password, retry_count=retry_count
@@ -120,12 +120,8 @@ class _ButtonAutomator(_MQTTControlledActor):
 class _CurtainMotor(_MQTTControlledActor):
     # https://www.home-assistant.io/integrations/cover.mqtt/
     MQTT_COMMAND_TOPIC_LEVELS = _CURTAIN_TOPIC_LEVELS_PREFIX + ("set",)
-    _MQTT_SET_POSITION_TOPIC_LEVELS: typing.Tuple[_MQTTTopicLevel, ...] = (
-        _CURTAIN_TOPIC_LEVELS_PREFIX
-        + (
-            "position",
-            "set-percent",
-        )
+    _MQTT_SET_POSITION_TOPIC_LEVELS: tuple[_MQTTTopicLevel, ...] = (
+        _CURTAIN_TOPIC_LEVELS_PREFIX + ("position", "set-percent")
     )
     _MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS = _CURTAIN_TOPIC_LEVELS_PREFIX + (
         "request-device-info",
@@ -149,7 +145,7 @@ class _CurtainMotor(_MQTTControlledActor):
         *,
         device: bleak.backends.device.BLEDevice,
         retry_count: int,
-        password: typing.Optional[str],
+        password: str | None,
     ) -> None:
         # > The position of the curtain is saved in self._pos with 0 = open and 100 = closed.
         # https://github.com/Danielhiversen/pySwitchbot/blob/0.10.0/switchbot/__init__.py#L150
@@ -267,7 +263,7 @@ class _CurtainMotor(_MQTTControlledActor):
         message: aiomqtt.Message,
         mqtt_topic_prefix: str,
         retry_count: int,
-        device_passwords: typing.Dict[str, str],
+        device_passwords: dict[str, str],
         fetch_device_info: bool,
     ) -> None:
         # pylint: disable=unused-argument; callback
@@ -308,7 +304,7 @@ class _CurtainMotor(_MQTTControlledActor):
         cls,
         *,
         enable_device_info_update_topic: bool,
-    ) -> typing.Dict[typing.Tuple[_MQTTTopicLevel, ...], typing.Callable]:
+    ) -> dict[tuple[_MQTTTopicLevel, ...], collections.abc.Callable]:
         callbacks = super()._get_mqtt_message_callbacks(
             enable_device_info_update_topic=enable_device_info_update_topic
         )

switchbot_mqtt/_actors/base.py

@@ -26,6 +26,7 @@
 from __future__ import annotations  # PEP563 (default in python>=3.10)
 
 import abc
+import collections.abc
 import logging
 import typing
 
@@ -45,14 +46,10 @@ _LOGGER = logging.getLogger(__name__)
 
 
 class _MQTTControlledActor(abc.ABC):
-    MQTT_COMMAND_TOPIC_LEVELS: typing.Tuple[_MQTTTopicLevel, ...] = NotImplemented
-    _MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS: typing.Tuple[_MQTTTopicLevel, ...] = (
-        NotImplemented
-    )
-    MQTT_STATE_TOPIC_LEVELS: typing.Tuple[_MQTTTopicLevel, ...] = NotImplemented
-    _MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS: typing.Tuple[_MQTTTopicLevel, ...] = (
-        NotImplemented
-    )
+    MQTT_COMMAND_TOPIC_LEVELS: tuple[_MQTTTopicLevel, ...] = NotImplemented
+    _MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS: tuple[_MQTTTopicLevel, ...] = NotImplemented
+    MQTT_STATE_TOPIC_LEVELS: tuple[_MQTTTopicLevel, ...] = NotImplemented
+    _MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS: tuple[_MQTTTopicLevel, ...] = NotImplemented
 
     @classmethod
     def get_mqtt_update_device_info_topic(cls, *, prefix: str, mac_address: str) -> str:
@@ -76,11 +73,11 @@ class _MQTTControlledActor(abc.ABC):
         *,
         device: bleak.backends.device.BLEDevice,
         retry_count: int,
-        password: typing.Optional[str],
+        password: str | None,
     ) -> None:
         # alternative: pySwitchbot >=0.10.0 provides SwitchbotDevice.get_mac()
         self._mac_address = device.address
-        self._basic_device_info: typing.Optional[typing.Dict[str, typing.Any]] = None
+        self._basic_device_info: dict[str, typing.Any] | None = None
 
     @abc.abstractmethod
     def _get_device(self) -> switchbot.SwitchbotDevice:
@@ -118,10 +115,10 @@ class _MQTTControlledActor(abc.ABC):
         *,
         topic: aiomqtt.Topic,
         mqtt_topic_prefix: str,
-        expected_topic_levels: typing.Collection[_MQTTTopicLevel],
+        expected_topic_levels: collections.abc.Collection[_MQTTTopicLevel],
         retry_count: int,
-        device_passwords: typing.Dict[str, str],
-    ) -> typing.Optional[_MQTTControlledActor]:
+        device_passwords: dict[str, str],
+    ) -> _MQTTControlledActor | None:
         try:
             mac_address = _parse_mqtt_topic(
                 topic=topic.value,
@@ -138,7 +135,7 @@ class _MQTTControlledActor(abc.ABC):
         # disabling mypy to workaround false-positive:
         # > error: Missing named argument "service_uuids" for
         # . "find_device_by_address" of "BleakScanner"  [call-arg]
-        device = await bleak.BleakScanner.find_device_by_address(mac_address)  # type: ignore
+        device = await bleak.BleakScanner.find_device_by_address(mac_address)
         if device is None:
             _LOGGER.error(
                 "failed to find bluetooth low energy device with mac address %s",
@@ -160,7 +157,7 @@ class _MQTTControlledActor(abc.ABC):
         message: aiomqtt.Message,
         mqtt_topic_prefix: str,
         retry_count: int,
-        device_passwords: typing.Dict[str, str],
+        device_passwords: dict[str, str],
         fetch_device_info: bool,
     ) -> None:
         # pylint: disable=unused-argument; callback
@@ -202,7 +199,7 @@ class _MQTTControlledActor(abc.ABC):
         message: aiomqtt.Message,
         mqtt_topic_prefix: str,
         retry_count: int,
-        device_passwords: typing.Dict[str, str],
+        device_passwords: dict[str, str],
         fetch_device_info: bool,
     ) -> None:
         # pylint: disable=unused-argument; callback
@@ -232,7 +229,7 @@ class _MQTTControlledActor(abc.ABC):
         cls,
         *,
         enable_device_info_update_topic: bool,
-    ) -> typing.Dict[typing.Tuple[_MQTTTopicLevel, ...], typing.Callable]:
+    ) -> dict[tuple[_MQTTTopicLevel, ...], collections.abc.Callable]:
         # returning dict because `paho.mqtt.client.Client.message_callback_add` overwrites
         # callbacks with same topic pattern
         # https://github.com/eclipse/paho.mqtt.python/blob/v1.6.1/src/paho/mqtt/client.py#L2304
@@ -251,7 +248,7 @@ class _MQTTControlledActor(abc.ABC):
         mqtt_client: aiomqtt.Client,
         mqtt_topic_prefix: str,
         fetch_device_info: bool,
-    ) -> typing.AsyncIterator[typing.Tuple[str, typing.Callable]]:
+    ) -> collections.abc.AsyncIterator[tuple[str, collections.abc.Callable]]:
         for topic_levels, callback in cls._get_mqtt_message_callbacks(
             enable_device_info_update_topic=fetch_device_info
         ).items():
@@ -268,7 +265,7 @@ class _MQTTControlledActor(abc.ABC):
         self,
         *,
         topic_prefix: str,
-        topic_levels: typing.Iterable[_MQTTTopicLevel],
+        topic_levels: collections.abc.Iterable[_MQTTTopicLevel],
         payload: bytes,
         mqtt_client: aiomqtt.Client,
     ) -> None:

switchbot_mqtt/_utils.py

@@ -16,10 +16,10 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
+import collections.abc
 import enum
 import queue  # pylint: disable=unused-import; in type hint
 import re
-import typing
 
 _MAC_ADDRESS_REGEX = re.compile(r"^[0-9a-f]{2}(:[0-9a-f]{2}){5}$")
 
@@ -32,13 +32,13 @@ class _MQTTTopicPlaceholder(enum.Enum):
     MAC_ADDRESS = "MAC_ADDRESS"
 
 
-_MQTTTopicLevel = typing.Union[str, _MQTTTopicPlaceholder]
+_MQTTTopicLevel = str | _MQTTTopicPlaceholder
 
 
 def _join_mqtt_topic_levels(
     *,
     topic_prefix: str,
-    topic_levels: typing.Iterable[_MQTTTopicLevel],
+    topic_levels: collections.abc.Iterable[_MQTTTopicLevel],
     mac_address: str,
 ) -> str:
     return topic_prefix + "/".join(
@@ -51,11 +51,11 @@ def _parse_mqtt_topic(
     *,
     topic: str,
     expected_prefix: str,
-    expected_levels: typing.Collection[_MQTTTopicLevel],
-) -> typing.Dict[_MQTTTopicPlaceholder, str]:
+    expected_levels: collections.abc.Collection[_MQTTTopicLevel],
+) -> dict[_MQTTTopicPlaceholder, str]:
     if not topic.startswith(expected_prefix):
         raise ValueError(f"expected topic prefix {expected_prefix}, got topic {topic}")
-    attrs: typing.Dict[_MQTTTopicPlaceholder, str] = {}
+    attrs: dict[_MQTTTopicPlaceholder, str] = {}
     topic_split = topic[len(expected_prefix) :].split("/")
     if len(topic_split) != len(expected_levels):
         raise ValueError(f"unexpected topic {topic}")

tests/test_actor_base.py

@@ -16,7 +16,6 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
-import typing
 import unittest.mock
 
 import bleak.backends.device
@@ -45,7 +44,7 @@ async def test_execute_command_abstract() -> None:
             self,
             device: bleak.backends.device.BLEDevice,
             retry_count: int,
-            password: typing.Optional[str],
+            password: str | None,
         ) -> None:
             super().__init__(device=device, retry_count=retry_count, password=password)
 

tests/test_cli.py

@@ -94,7 +94,7 @@ def test_console_entry_point() -> None:
     ],
 )
 def test__main(
-    argv: typing.List[str],
+    argv: list[str],
     expected_mqtt_host: str,
     expected_mqtt_port: int,
     expected_username: str,
@@ -198,7 +198,7 @@ def test__main_mqtt_password_file_collision(
     ],
 )
 def test__main_device_password_file(
-    tmp_path: pathlib.Path, device_passwords: typing.Dict[str, str]
+    tmp_path: pathlib.Path, device_passwords: dict[str, str]
 ) -> None:
     device_passwords_path = tmp_path.joinpath("passwords.json")
     device_passwords_path.write_text(json.dumps(device_passwords), encoding="utf8")
@@ -226,7 +226,7 @@ def test__main_device_password_file(
     )
 
 
-_RUN_DEFAULT_KWARGS: typing.Dict[str, typing.Any] = {
+_RUN_DEFAULT_KWARGS: dict[str, typing.Any] = {
     "mqtt_host": "localhost",
     "mqtt_port": 8883,
     "mqtt_disable_tls": False,
@@ -275,7 +275,7 @@ def test__main_mqtt_disable_tls_overwrite_port() -> None:
     [([], "homeassistant/"), (["--mqtt-topic-prefix", ""], "")],
 )
 def test__main_mqtt_topic_prefix(
-    additional_argv: typing.List[str], expected_topic_prefix: str
+    additional_argv: list[str], expected_topic_prefix: str
 ) -> None:
     with unittest.mock.patch("switchbot_mqtt._run") as run_mock, unittest.mock.patch(
         "sys.argv", ["", "--mqtt-host", "localhost"] + additional_argv
@@ -345,7 +345,7 @@ def test__main_fetch_device_info() -> None:
     ],
 )
 def test__main_log_config(
-    additional_argv: typing.List[str], root_log_level: int, log_format: str
+    additional_argv: list[str], root_log_level: int, log_format: str
 ) -> None:
     with unittest.mock.patch(
         "sys.argv", ["", "--mqtt-host", "localhost"] + additional_argv

tests/test_mqtt.py

@@ -16,10 +16,10 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
+import collections.abc
 import logging
 import socket
 import ssl
-import typing
 import unittest.mock
 
 import _pytest.logging  # pylint: disable=import-private-name; typing
@@ -45,7 +45,7 @@ async def test__listen(caplog: _pytest.logging.LogCaptureFixture) -> None:
     mqtt_client = unittest.mock.AsyncMock()
     messages_mock = unittest.mock.AsyncMock()
 
-    async def _msg_iter() -> typing.AsyncIterator[aiomqtt.Message]:
+    async def _msg_iter() -> collections.abc.AsyncIterator[aiomqtt.Message]:
         for topic, payload in [
             ("/foo", b"foo1"),
             ("/baz/21/bar", b"42/2"),
@@ -110,7 +110,7 @@ async def test__listen(caplog: _pytest.logging.LogCaptureFixture) -> None:
 def test__log_mqtt_connected(
     caplog: _pytest.logging.LogCaptureFixture,
     socket_family: int,  # socket.AddressFamily,
-    peername: typing.Tuple[typing.Union[str, int]],
+    peername: tuple[str | int],
     peername_log: str,
 ) -> None:
     mqtt_client = unittest.mock.MagicMock()
@@ -142,7 +142,7 @@ async def test__run(
     mqtt_host: str,
     mqtt_port: int,
     retry_count: int,
-    device_passwords: typing.Dict[str, str],
+    device_passwords: dict[str, str],
     fetch_device_info: bool,
 ) -> None:
     with unittest.mock.patch("aiomqtt.Client") as mqtt_client_mock, unittest.mock.patch(
@@ -299,7 +299,7 @@ async def test__run_authentication(
     mqtt_host: str,
     mqtt_port: int,
     mqtt_username: str,
-    mqtt_password: typing.Optional[str],
+    mqtt_password: str | None,
 ) -> None:
     with unittest.mock.patch("aiomqtt.Client") as mqtt_client_mock, unittest.mock.patch(
         "switchbot_mqtt._listen"
@@ -343,36 +343,38 @@ async def test__run_authentication_missing_username(
         )
 
 
+class _ActorMockBase(_MQTTControlledActor):
+    def __init__(
+        self,
+        device: bleak.backends.device.BLEDevice,
+        retry_count: int,
+        password: str | None,
+    ) -> None:
+        super().__init__(device=device, retry_count=retry_count, password=password)
+
+    async def execute_command(
+        self,
+        *,
+        mqtt_message_payload: bytes,
+        mqtt_client: aiomqtt.Client,
+        update_device_info: bool,
+        mqtt_topic_prefix: str,
+    ) -> None:
+        pass
+
+    def _get_device(self) -> None:
+        return None
+
+
 def _mock_actor_class(
     *,
-    command_topic_levels: typing.Tuple[_MQTTTopicLevel, ...] = NotImplemented,
-    request_info_levels: typing.Tuple[_MQTTTopicLevel, ...] = NotImplemented,
-) -> typing.Type:
-    class _ActorMock(_MQTTControlledActor):
+    command_topic_levels: tuple[_MQTTTopicLevel, ...] = NotImplemented,
+    request_info_levels: tuple[_MQTTTopicLevel, ...] = NotImplemented,
+) -> type[_ActorMockBase]:
+    class _ActorMock(_ActorMockBase):
         MQTT_COMMAND_TOPIC_LEVELS = command_topic_levels
         _MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS = request_info_levels
 
-        def __init__(
-            self,
-            device: bleak.backends.device.BLEDevice,
-            retry_count: int,
-            password: typing.Optional[str],
-        ) -> None:
-            super().__init__(device=device, retry_count=retry_count, password=password)
-
-        async def execute_command(
-            self,
-            *,
-            mqtt_message_payload: bytes,
-            mqtt_client: aiomqtt.Client,
-            update_device_info: bool,
-            mqtt_topic_prefix: str,
-        ) -> None:
-            pass
-
-        def _get_device(self) -> None:
-            return None
-
     return _ActorMock
 
 
@@ -390,12 +392,13 @@ def _mock_actor_class(
 @pytest.mark.parametrize("payload", [b"", b"whatever"])
 async def test__mqtt_update_device_info_callback(
     caplog: _pytest.logging.LogCaptureFixture,
-    topic_levels: typing.Tuple[_MQTTTopicLevel, ...],
+    topic_levels: tuple[_MQTTTopicLevel, ...],
     topic: str,
     expected_mac_address: str,
     payload: bytes,
 ) -> None:
     ActorMock = _mock_actor_class(request_info_levels=topic_levels)
+    mqtt_client = unittest.mock.Mock()
     message = aiomqtt.Message(
         topic=topic, payload=payload, qos=0, retain=False, mid=0, properties=None
     )
@@ -410,7 +413,7 @@ async def test__mqtt_update_device_info_callback(
         logging.DEBUG
     ):
         await ActorMock._mqtt_update_device_info_callback(
-            mqtt_client="client_dummy",
+            mqtt_client=mqtt_client,
             message=message,
             mqtt_topic_prefix="prfx/",
             retry_count=21,  # tested in test__mqtt_command_callback
@@ -420,7 +423,7 @@ async def test__mqtt_update_device_info_callback(
     find_device_mock.assert_awaited_once_with(expected_mac_address)
     init_mock.assert_called_once_with(device=device, retry_count=21, password=None)
     update_mock.assert_called_once_with(
-        mqtt_client="client_dummy", mqtt_topic_prefix="prfx/"
+        mqtt_client=mqtt_client, mqtt_topic_prefix="prfx/"
     )
     assert caplog.record_tuples == [
         (
@@ -454,7 +457,7 @@ async def test__mqtt_update_device_info_callback_ignore_retained(
         logging.DEBUG
     ):
         await ActorMock._mqtt_update_device_info_callback(
-            mqtt_client="client_dummy",
+            mqtt_client=unittest.mock.Mock(),
             message=message,
             mqtt_topic_prefix="ignored",
             retry_count=21,
@@ -540,7 +543,7 @@ async def test__mqtt_update_device_info_callback_ignore_retained(
 async def test__mqtt_command_callback(
     caplog: _pytest.logging.LogCaptureFixture,
     topic_prefix: str,
-    command_topic_levels: typing.Tuple[_MQTTTopicLevel, ...],
+    command_topic_levels: tuple[_MQTTTopicLevel, ...],
     topic: str,
     payload: bytes,
     expected_mac_address: str,
@@ -548,6 +551,7 @@ async def test__mqtt_command_callback(
     fetch_device_info: bool,
 ) -> None:
     ActorMock = _mock_actor_class(command_topic_levels=command_topic_levels)
+    mqtt_client = unittest.mock.Mock()
     message = aiomqtt.Message(
         topic=topic, payload=payload, qos=0, retain=False, mid=0, properties=None
     )
@@ -563,7 +567,7 @@ async def test__mqtt_command_callback(
         logging.DEBUG
     ):
         await ActorMock._mqtt_command_callback(
-            mqtt_client="client_dummy",
+            mqtt_client=mqtt_client,
             message=message,
             retry_count=retry_count,
             device_passwords={},
@@ -575,7 +579,7 @@ async def test__mqtt_command_callback(
         device=device, retry_count=retry_count, password=None
     )
     execute_command_mock.assert_awaited_once_with(
-        mqtt_client="client_dummy",
+        mqtt_client=mqtt_client,
         mqtt_message_payload=payload,
         update_device_info=fetch_device_info,
         mqtt_topic_prefix=topic_prefix,
@@ -599,11 +603,12 @@ async def test__mqtt_command_callback(
     ],
 )
 async def test__mqtt_command_callback_password(
-    mac_address: str, expected_password: typing.Optional[str]
+    mac_address: str, expected_password: str | None
 ) -> None:
     ActorMock = _mock_actor_class(
         command_topic_levels=("switchbot", _MQTTTopicPlaceholder.MAC_ADDRESS)
     )
+    mqtt_client = unittest.mock.Mock()
     message = aiomqtt.Message(
         topic="prefix-switchbot/" + mac_address,
         payload=b"whatever",
@@ -622,7 +627,7 @@ async def test__mqtt_command_callback_password(
         ActorMock, "execute_command"
     ) as execute_command_mock:
         await ActorMock._mqtt_command_callback(
-            mqtt_client="client_dummy",
+            mqtt_client=mqtt_client,
             message=message,
             retry_count=3,
             device_passwords={
@@ -638,7 +643,7 @@ async def test__mqtt_command_callback_password(
         device=device, retry_count=3, password=expected_password
     )
     execute_command_mock.assert_awaited_once_with(
-        mqtt_client="client_dummy",
+        mqtt_client=mqtt_client,
         mqtt_message_payload=b"whatever",
         update_device_info=True,
         mqtt_topic_prefix="prefix-",
@@ -671,7 +676,7 @@ async def test__mqtt_command_callback_unexpected_topic(
         logging.DEBUG
     ):
         await ActorMock._mqtt_command_callback(
-            mqtt_client="client_dummy",
+            mqtt_client=unittest.mock.Mock(),
             message=message,
             retry_count=3,
             device_passwords={},
@@ -715,7 +720,7 @@ async def test__mqtt_command_callback_invalid_mac_address(
         logging.DEBUG
     ):
         await ActorMock._mqtt_command_callback(
-            mqtt_client="client_dummy",
+            mqtt_client=unittest.mock.Mock(),
             message=message,
             retry_count=3,
             device_passwords={},
@@ -761,7 +766,7 @@ async def test__mqtt_command_callback_device_not_found(
         logging.DEBUG
     ):
         await ActorMock._mqtt_command_callback(
-            mqtt_client="client_dummy",
+            mqtt_client=unittest.mock.Mock(),
             message=message,
             retry_count=3,
             device_passwords={},
@@ -806,7 +811,7 @@ async def test__mqtt_command_callback_ignore_retained(
         logging.DEBUG
     ):
         await ActorMock._mqtt_command_callback(
-            mqtt_client="client_dummy",
+            mqtt_client=unittest.mock.Mock(),
             message=message,
             retry_count=4,
             device_passwords={},
@@ -850,7 +855,7 @@ async def test__mqtt_command_callback_ignore_retained(
 async def test__report_state(
     caplog: _pytest.logging.LogCaptureFixture,
     topic_prefix: str,
-    state_topic_levels: typing.Tuple[_MQTTTopicLevel, ...],
+    state_topic_levels: tuple[_MQTTTopicLevel, ...],
     mac_address: str,
     expected_topic: str,
     state: bytes,
@@ -864,7 +869,7 @@ async def test__report_state(
             self,
             device: bleak.backends.device.BLEDevice,
             retry_count: int,
-            password: typing.Optional[str],
+            password: str | None,
         ) -> None:
             super().__init__(device=device, retry_count=retry_count, password=password)
 

tests/test_switchbot_button_automator.py

@@ -21,7 +21,6 @@
 # pylint: disable=duplicate-code; similarities with tests for curtain motor
 
 import logging
-import typing
 import unittest.mock
 
 import _pytest.logging  # pylint: disable=import-private-name; typing
@@ -90,7 +89,7 @@ async def test_execute_command(
     caplog: _pytest.logging.LogCaptureFixture,
     topic_prefix: str,
     mac_address: str,
-    password: typing.Optional[str],
+    password: str | None,
     retry_count: int,
     message_payload: bytes,
     action_name: str,

tests/test_switchbot_curtain_motor.py

@@ -17,7 +17,6 @@
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
 import logging
-import typing
 import unittest.mock
 
 import _pytest.logging  # pylint: disable=import-private-name; typing
@@ -218,7 +217,7 @@ async def test_execute_command(
     caplog: _pytest.logging.LogCaptureFixture,
     topic_prefix: str,
     mac_address: str,
-    password: typing.Optional[str],
+    password: str | None,
     retry_count: int,
     message_payload: bytes,
     action_name: str,

tests/test_switchbot_curtain_motor_position.py

@@ -30,6 +30,14 @@ from switchbot_mqtt._actors import _CurtainMotor
 # pylint: disable=protected-access,too-many-positional-arguments
 
 
+def _create_mqtt_message(
+    *, topic: str, payload: bytes, retain: bool = False
+) -> aiomqtt.Message:
+    return aiomqtt.Message(
+        topic=topic, payload=payload, qos=0, retain=retain, mid=0, properties=None
+    )
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
     ("topic", "payload", "expected_mac_address", "expected_position_percent"),
@@ -63,9 +71,7 @@ async def test__mqtt_set_position_callback(
     retry_count: int,
     expected_position_percent: int,
 ) -> None:
-    message = aiomqtt.Message(
-        topic=topic, payload=payload, qos=0, retain=False, mid=0, properties=None
-    )
+    message = _create_mqtt_message(topic=topic, payload=payload)
     device = unittest.mock.Mock()
     device.address = expected_mac_address
     with unittest.mock.patch.object(
@@ -109,13 +115,10 @@ async def test__mqtt_set_position_callback(
 async def test__mqtt_set_position_callback_ignore_retained(
     caplog: _pytest.logging.LogCaptureFixture,
 ) -> None:
-    message = aiomqtt.Message(
+    message = _create_mqtt_message(
         topic="homeassistant/cover/switchbot-curtain/aa:bb:cc:dd:ee:ff/position/set-percent",
         payload=b"42",
-        qos=0,
         retain=True,
-        mid=0,
-        properties=None,
     )
     with unittest.mock.patch(
         "switchbot.SwitchbotCurtain"
@@ -143,13 +146,8 @@ async def test__mqtt_set_position_callback_ignore_retained(
 async def test__mqtt_set_position_callback_unexpected_topic(
     caplog: _pytest.logging.LogCaptureFixture,
 ) -> None:
-    message = aiomqtt.Message(
-        topic="switchbot-curtain/aa:bb:cc:dd:ee:ff/position/set",
-        payload=b"42",
-        qos=0,
-        retain=False,
-        mid=0,
-        properties=None,
+    message = _create_mqtt_message(
+        topic="switchbot-curtain/aa:bb:cc:dd:ee:ff/position/set", payload=b"42"
     )
     with unittest.mock.patch(
         "switchbot.SwitchbotCurtain"
@@ -176,13 +174,9 @@ async def test__mqtt_set_position_callback_unexpected_topic(
 async def test__mqtt_set_position_callback_invalid_mac_address(
     caplog: _pytest.logging.LogCaptureFixture,
 ) -> None:
-    message = aiomqtt.Message(
+    message = _create_mqtt_message(
         topic="tnatsissaemoh/cover/switchbot-curtain/aa:bb:cc:dd:ee/position/set-percent",
         payload=b"42",
-        qos=0,
-        retain=False,
-        mid=0,
-        properties=None,
     )
     with unittest.mock.patch(
         "switchbot.SwitchbotCurtain"
@@ -211,13 +205,9 @@ async def test__mqtt_set_position_callback_invalid_position(
     caplog: _pytest.logging.LogCaptureFixture,
     payload: bytes,
 ) -> None:
-    message = aiomqtt.Message(
+    message = _create_mqtt_message(
         topic="homeassistant/cover/switchbot-curtain/aa:bb:cc:dd:ee:ff/position/set-percent",
         payload=payload,
-        qos=0,
-        retain=False,
-        mid=0,
-        properties=None,
     )
     with unittest.mock.patch.object(
         bleak.BleakScanner, "find_device_by_address"
@@ -249,13 +239,9 @@ async def test__mqtt_set_position_callback_invalid_position(
 async def test__mqtt_set_position_callback_command_failed(
     caplog: _pytest.logging.LogCaptureFixture,
 ) -> None:
-    message = aiomqtt.Message(
+    message = _create_mqtt_message(
         topic="cover/switchbot-curtain/aa:bb:cc:dd:ee:ff/position/set-percent",
         payload=b"21",
-        qos=0,
-        retain=False,
-        mid=0,
-        properties=None,
     )
     device = unittest.mock.Mock()
     device.address = "aa:bb:cc:dd:ee:ff"

tests/test_utils.py

@@ -16,8 +16,6 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
-import typing
-
 import pytest
 
 # pylint: disable=import-private-name; internal
@@ -87,9 +85,9 @@ def test__mac_address_valid(mac_address: str, valid: bool) -> None:
 )
 def test__parse_mqtt_topic(
     expected_prefix: str,
-    expected_levels: typing.List[_MQTTTopicLevel],
+    expected_levels: list[_MQTTTopicLevel],
     topic: str,
-    expected_attrs: typing.Dict[_MQTTTopicPlaceholder, str],
+    expected_attrs: dict[_MQTTTopicPlaceholder, str],
 ) -> None:
     assert (
         _parse_mqtt_topic(
@@ -134,7 +132,7 @@ def test__parse_mqtt_topic_unexpected_prefix() -> None:
     ],
 )
 def test__parse_mqtt_topic_fail(
-    expected_prefix: str, expected_levels: typing.List[_MQTTTopicLevel], topic: str
+    expected_prefix: str, expected_levels: list[_MQTTTopicLevel], topic: str
 ) -> None:
     with pytest.raises(ValueError):
         _parse_mqtt_topic(