.pre-commit-config.yaml

@@ -9,7 +9,7 @@ ci:
 
 repos:
   - repo: https://github.com/commitizen-tools/commitizen
-    rev: v4.8.2
+    rev: v4.8.3
     hooks:
       - id: commitizen
         stages: [commit-msg]
@@ -38,7 +38,7 @@ repos:
       - id: pyupgrade
         args: [--py311-plus]
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.11.12
+    rev: v0.11.13
     hooks:
       - id: ruff
         args: [--fix]

setup.py

@@ -20,7 +20,7 @@ setup(
         "cryptography>=39.0.0",
         "pyOpenSSL>=23.0.0",
     ],
-    version="0.65.0",
+    version="0.66.0",
     description="A library to communicate with Switchbot",
     long_description=long_description,
     long_description_content_type="text/markdown",

switchbot/__init__.py

@@ -11,11 +11,15 @@ from bleak_retry_connector import (
 from .adv_parser import SwitchbotSupportedType, parse_advertisement_data
 from .const import (
     AirPurifierMode,
+    BulbColorMode,
+    CeilingLightColorMode,
+    ColorMode,
     FanMode,
     HumidifierAction,
     HumidifierMode,
     HumidifierWaterLevel,
     LockStatus,
+    StripLightColorMode,
     SwitchbotAccountConnectionError,
     SwitchbotApiError,
     SwitchbotAuthenticationError,
@@ -28,11 +32,11 @@ from .devices.bot import Switchbot
 from .devices.bulb import SwitchbotBulb
 from .devices.ceiling_light import SwitchbotCeilingLight
 from .devices.curtain import SwitchbotCurtain
-from .devices.device import ColorMode, SwitchbotDevice, SwitchbotEncryptedDevice
+from .devices.device import SwitchbotDevice, SwitchbotEncryptedDevice
 from .devices.evaporative_humidifier import SwitchbotEvaporativeHumidifier
 from .devices.fan import SwitchbotFan
 from .devices.humidifier import SwitchbotHumidifier
-from .devices.light_strip import SwitchbotLightStrip
+from .devices.light_strip import SwitchbotLightStrip, SwitchbotStripLight3
 from .devices.lock import SwitchbotLock
 from .devices.plug import SwitchbotPlugMini
 from .devices.relay_switch import SwitchbotRelaySwitch, SwitchbotRelaySwitch2PM
@@ -43,6 +47,8 @@ from .models import SwitchBotAdvertisement
 
 __all__ = [
     "AirPurifierMode",
+    "BulbColorMode",
+    "CeilingLightColorMode",
     "ColorMode",
     "FanMode",
     "GetSwitchbotDevices",
@@ -50,6 +56,7 @@ __all__ = [
     "HumidifierMode",
     "HumidifierWaterLevel",
     "LockStatus",
+    "StripLightColorMode",
     "SwitchBotAdvertisement",
     "Switchbot",
     "Switchbot",
@@ -76,6 +83,7 @@ __all__ = [
     "SwitchbotRelaySwitch",
     "SwitchbotRelaySwitch2PM",
     "SwitchbotRollerShade",
+    "SwitchbotStripLight3",
     "SwitchbotSupportedType",
     "SwitchbotSupportedType",
     "SwitchbotVacuum",

switchbot/adv_parser.py

@@ -24,7 +24,7 @@ from .adv_parsers.hubmini_matter import process_hubmini_matter
 from .adv_parsers.humidifier import process_evaporative_humidifier, process_wohumidifier
 from .adv_parsers.keypad import process_wokeypad
 from .adv_parsers.leak import process_leak
-from .adv_parsers.light_strip import process_wostrip
+from .adv_parsers.light_strip import process_light, process_wostrip
 from .adv_parsers.lock import process_lock2, process_wolock, process_wolock_pro
 from .adv_parsers.meter import process_wosensorth, process_wosensorth_c
 from .adv_parsers.motion import process_wopresence
@@ -325,6 +325,18 @@ SUPPORTED_TYPES: dict[str | bytes, SwitchbotSupportedType] = {
         "func": process_relay_switch_2pm,
         "manufacturer_id": 2409,
     },
+    b"\x00\x10\xd0\xb0": {
+        "modelName": SwitchbotModel.FLOOR_LAMP,
+        "modelFriendlyName": "Floor Lamp",
+        "func": process_light,
+        "manufacturer_id": 2409,
+    },
+    b"\x00\x10\xd0\xb1": {
+        "modelName": SwitchbotModel.STRIP_LIGHT_3,
+        "modelFriendlyName": "Strip Light 3",
+        "func": process_light,
+        "manufacturer_id": 2409,
+    },
 }
 
 _SWITCHBOT_MODEL_TO_CHAR = {

switchbot/adv_parsers/light_strip.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import struct
+
 
 def process_wostrip(
     data: bytes | None, mfr_data: bytes | None
@@ -14,8 +16,17 @@ def process_wostrip(
         "isOn": bool(mfr_data[7] & 0b10000000),
         "brightness": mfr_data[7] & 0b01111111,
         "delay": bool(mfr_data[8] & 0b10000000),
-        "preset": bool(mfr_data[8] & 0b00001000),
-        "color_mode": mfr_data[8] & 0b00000111,
-        "speed": mfr_data[9] & 0b01111111,
-        "loop_index": mfr_data[10] & 0b11111110,
+        "network_state": (mfr_data[8] & 0b01110000) >> 4,
+        "color_mode": mfr_data[8] & 0b00001111,
     }
+
+
+def process_light(data: bytes | None, mfr_data: bytes | None) -> dict[str, bool | int]:
+    """Support for strip light 3 and floor lamp."""
+    common_data = process_wostrip(data, mfr_data)
+    if not common_data:
+        return {}
+
+    light_data = {"cw": struct.unpack(">H", mfr_data[16:18])[0]}
+
+    return common_data | light_data

switchbot/const/__init__.py

@@ -10,6 +10,12 @@ from .evaporative_humidifier import (
     HumidifierWaterLevel,
 )
 from .fan import FanMode
+from .light import (
+    BulbColorMode,
+    CeilingLightColorMode,
+    ColorMode,
+    StripLightColorMode,
+)
 
 # Preserve old LockStatus export for backwards compatibility
 from .lock import LockStatus
@@ -85,6 +91,8 @@ class SwitchbotModel(StrEnum):
     LOCK_LITE = "Lock Lite"
     GARAGE_DOOR_OPENER = "Garage Door Opener"
     RELAY_SWITCH_2PM = "Relay Switch 2PM"
+    STRIP_LIGHT_3 = "Strip Light 3"
+    FLOOR_LAMP = "Floor Lamp"
 
 
 __all__ = [
@@ -92,11 +100,15 @@ __all__ = [
     "DEFAULT_RETRY_TIMEOUT",
     "DEFAULT_SCAN_TIMEOUT",
     "AirPurifierMode",
+    "BulbColorMode",
+    "CeilingLightColorMode",
+    "ColorMode",
     "FanMode",
     "HumidifierAction",
     "HumidifierMode",
     "HumidifierWaterLevel",
     "LockStatus",
+    "StripLightColorMode",
     "SwitchbotAccountConnectionError",
     "SwitchbotApiError",
     "SwitchbotAuthenticationError",

switchbot/const/light.py

@@ -0,0 +1,34 @@
+from enum import Enum
+
+
+class ColorMode(Enum):
+    OFF = 0
+    COLOR_TEMP = 1
+    RGB = 2
+    EFFECT = 3
+
+
+class StripLightColorMode(Enum):
+    RGB = 2
+    SCENE = 3
+    MUSIC = 4
+    CONTROLLER = 5
+    COLOR_TEMP = 6
+    UNKNOWN = 10
+
+
+class BulbColorMode(Enum):
+    COLOR_TEMP = 1
+    RGB = 2
+    DYNAMIC = 3
+    UNKNOWN = 10
+
+
+class CeilingLightColorMode(Enum):
+    COLOR_TEMP = 0
+    NIGHT = 1
+    MUSIC = 4
+    UNKNOWN = 10
+
+
+DEFAULT_COLOR_TEMP = 4001

switchbot/devices/base_light.py

@@ -1,13 +1,12 @@
 from __future__ import annotations
 
 import logging
-import time
 from abc import abstractmethod
 from typing import Any
 
 from ..helpers import create_background_task
 from ..models import SwitchBotAdvertisement
-from .device import ColorMode, SwitchbotDevice
+from .device import SwitchbotDevice
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -43,9 +42,10 @@ class SwitchbotBaseLight(SwitchbotDevice):
         return self._get_adv_value("brightness") or 0
 
     @property
-    def color_mode(self) -> ColorMode:
+    @abstractmethod
+    def color_mode(self) -> Any:
         """Return the current color mode."""
-        return ColorMode(self._get_adv_value("color_mode") or 0)
+        raise NotImplementedError("Subclasses must implement color mode")
 
     @property
     def min_temp(self) -> int:
@@ -57,10 +57,19 @@ class SwitchbotBaseLight(SwitchbotDevice):
         """Return maximum color temp."""
         return 6500
 
+    @property
+    def get_effect_list(self) -> list[str] | None:
+        """Return the list of supported effects."""
+        return None
+
     def is_on(self) -> bool | None:
         """Return bulb state from cache."""
         return self._get_adv_value("isOn")
 
+    def get_effect(self):
+        """Return the current effect."""
+        return self._get_adv_value("effect")
+
     @abstractmethod
     async def turn_on(self) -> bool:
         """Turn device on."""
@@ -81,13 +90,18 @@ class SwitchbotBaseLight(SwitchbotDevice):
     async def set_rgb(self, brightness: int, r: int, g: int, b: int) -> bool:
         """Set rgb."""
 
-    def poll_needed(self, last_poll_time: float | None) -> bool:
-        """Return if poll is needed."""
-        return False
-
-    async def update(self) -> None:
-        """Update device data."""
-        self._last_full_update = time.monotonic()
+    async def _send_multiple_commands(self, keys: list[str]) -> bool:
+        """
+        Send multiple commands to device.
+
+        Since we current have no way to tell which command the device
+        needs we send both.
+        """
+        final_result = False
+        for key in keys:
+            result = await self._send_command(key)
+            final_result |= self._check_command_result(result, 0, {1})
+        return final_result
 
 
 class SwitchbotSequenceBaseLight(SwitchbotBaseLight):

switchbot/devices/bulb.py

@@ -1,9 +1,11 @@
 from __future__ import annotations
 
 import logging
+from typing import Any
 
+from ..const.light import BulbColorMode, ColorMode
 from .base_light import SwitchbotSequenceBaseLight
-from .device import REQ_HEADER, ColorMode
+from .device import REQ_HEADER, SwitchbotOperationError, update_after_operation
 
 BULB_COMMAND_HEADER = "4701"
 BULB_REQUEST = f"{REQ_HEADER}4801"
@@ -18,9 +20,27 @@ BRIGHTNESS_KEY = f"{BULB_COMMAND}14"
 RGB_KEY = f"{BULB_COMMAND}16"
 CW_KEY = f"{BULB_COMMAND}17"
 
+DEVICE_GET_VERSION_KEY = "570003"
+DEVICE_GET_BASIC_SETTINGS_KEY = "570f4801"
+
 _LOGGER = logging.getLogger(__name__)
 
 
+EFFECT_DICT = {
+    "Colorful": "570F4701010300",
+    "Flickering": "570F4701010301",
+    "Breathing": "570F4701010302",
+}
+
+# Private mapping from device-specific color modes to original ColorMode enum
+_BULB_COLOR_MODE_MAP = {
+    BulbColorMode.COLOR_TEMP: ColorMode.COLOR_TEMP,
+    BulbColorMode.RGB: ColorMode.RGB,
+    BulbColorMode.DYNAMIC: ColorMode.EFFECT,
+    BulbColorMode.UNKNOWN: ColorMode.OFF,
+}
+
+
 class SwitchbotBulb(SwitchbotSequenceBaseLight):
     """Representation of a Switchbot bulb."""
 
@@ -29,41 +49,47 @@ class SwitchbotBulb(SwitchbotSequenceBaseLight):
         """Return the supported color modes."""
         return {ColorMode.RGB, ColorMode.COLOR_TEMP}
 
-    async def update(self) -> None:
-        """Update state of device."""
-        result = await self._send_command(BULB_REQUEST)
-        self._update_state(result)
-        await super().update()
+    @property
+    def color_mode(self) -> ColorMode:
+        """Return the current color mode."""
+        device_mode = BulbColorMode(self._get_adv_value("color_mode") or 10)
+        return _BULB_COLOR_MODE_MAP.get(device_mode, ColorMode.OFF)
+
+    @property
+    def get_effect_list(self) -> list[str]:
+        """Return the list of supported effects."""
+        return list(EFFECT_DICT.keys())
 
+    @update_after_operation
     async def turn_on(self) -> bool:
         """Turn device on."""
         result = await self._send_command(BULB_ON_KEY)
-        self._update_state(result)
-        return self._check_command_result(result, 1, {0x80})
+        return self._check_command_result(result, 0, {1})
 
+    @update_after_operation
     async def turn_off(self) -> bool:
         """Turn device off."""
         result = await self._send_command(BULB_OFF_KEY)
-        self._update_state(result)
-        return self._check_command_result(result, 1, {0x00})
+        return self._check_command_result(result, 0, {1})
 
+    @update_after_operation
     async def set_brightness(self, brightness: int) -> bool:
         """Set brightness."""
         assert 0 <= brightness <= 100, "Brightness must be between 0 and 100"
         result = await self._send_command(f"{BRIGHTNESS_KEY}{brightness:02X}")
-        self._update_state(result)
-        return self._check_command_result(result, 1, {0x80})
+        return self._check_command_result(result, 0, {1})
 
+    @update_after_operation
     async def set_color_temp(self, brightness: int, color_temp: int) -> bool:
         """Set color temp."""
         assert 0 <= brightness <= 100, "Brightness must be between 0 and 100"
-        assert 2700 <= color_temp <= 6500, "Color Temp must be between 0 and 100"
+        assert 2700 <= color_temp <= 6500, "Color Temp must be between 2700 and 6500"
         result = await self._send_command(
             f"{CW_BRIGHTNESS_KEY}{brightness:02X}{color_temp:04X}"
         )
-        self._update_state(result)
-        return self._check_command_result(result, 1, {0x80})
+        return self._check_command_result(result, 0, {1})
 
+    @update_after_operation
     async def set_rgb(self, brightness: int, r: int, g: int, b: int) -> bool:
         """Set rgb."""
         assert 0 <= brightness <= 100, "Brightness must be between 0 and 100"
@@ -73,22 +99,45 @@ class SwitchbotBulb(SwitchbotSequenceBaseLight):
         result = await self._send_command(
             f"{RGB_BRIGHTNESS_KEY}{brightness:02X}{r:02X}{g:02X}{b:02X}"
         )
-        self._update_state(result)
-        return self._check_command_result(result, 1, {0x80})
-
-    def _update_state(self, result: bytes | None) -> None:
-        """Update device state."""
-        if not result or len(result) < 10:
-            return
-        self._state["r"] = result[3]
-        self._state["g"] = result[4]
-        self._state["b"] = result[5]
-        self._state["cw"] = int(result[6:8].hex(), 16)
-        self._override_state(
-            {
-                "isOn": result[1] == 0x80,
-                "color_mode": result[10],
-            }
+        return self._check_command_result(result, 0, {1})
+
+    @update_after_operation
+    async def set_effect(self, effect: str) -> bool:
+        """Set effect."""
+        effect_template = EFFECT_DICT.get(effect)
+        if not effect_template:
+            raise SwitchbotOperationError(f"Effect {effect} not supported")
+        result = await self._send_command(effect_template)
+        if result:
+            self._override_state({"effect": effect})
+        return result
+
+    async def get_basic_info(self) -> dict[str, Any] | None:
+        """Get device basic settings."""
+        if not (_data := await self._get_basic_info(DEVICE_GET_BASIC_SETTINGS_KEY)):
+            return None
+        if not (_version_info := await self._get_basic_info(DEVICE_GET_VERSION_KEY)):
+            return None
+
+        _LOGGER.debug(
+            "data: %s, version info: %s, address: %s",
+            _data,
+            _version_info,
+            self._device.address,
         )
-        _LOGGER.debug("%s: update state: %s = %s", self.name, result.hex(), self._state)
-        self._fire_callbacks()
+
+        self._state["r"] = _data[3]
+        self._state["g"] = _data[4]
+        self._state["b"] = _data[5]
+        self._state["cw"] = int.from_bytes(_data[6:8], "big")
+
+        return {
+            "isOn": bool(_data[1] & 0b10000000),
+            "brightness": _data[2] & 0b01111111,
+            "r": self._state["r"],
+            "g": self._state["g"],
+            "b": self._state["b"],
+            "cw": self._state["cw"],
+            "color_mode": _data[10] & 0b00001111,
+            "firmware": _version_info[2] / 10.0,
+        }

switchbot/devices/ceiling_light.py

@@ -1,9 +1,15 @@
 from __future__ import annotations
 
 import logging
+from typing import Any
 
-from .base_light import SwitchbotBaseLight
-from .device import REQ_HEADER, ColorMode
+from ..const.light import (
+    DEFAULT_COLOR_TEMP,
+    CeilingLightColorMode,
+    ColorMode,
+)
+from .base_light import SwitchbotSequenceBaseLight
+from .device import REQ_HEADER, update_after_operation
 
 CEILING_LIGHT_COMMAND_HEADER = "5401"
 CEILING_LIGHT_REQUEST = f"{REQ_HEADER}5501"
@@ -14,56 +20,86 @@ CEILING_LIGHT_OFF_KEY = f"{CEILING_LIGHT_COMMAND}02FF01FFFF"
 CW_BRIGHTNESS_KEY = f"{CEILING_LIGHT_COMMAND}010001"
 BRIGHTNESS_KEY = f"{CEILING_LIGHT_COMMAND}01FF01"
 
+DEVICE_GET_VERSION_KEY = "5702"
+DEVICE_GET_BASIC_SETTINGS_KEY = "570f5581"
 
 _LOGGER = logging.getLogger(__name__)
 
+# Private mapping from device-specific color modes to original ColorMode enum
+_CEILING_LIGHT_COLOR_MODE_MAP = {
+    CeilingLightColorMode.COLOR_TEMP: ColorMode.COLOR_TEMP,
+    CeilingLightColorMode.NIGHT: ColorMode.COLOR_TEMP,
+    CeilingLightColorMode.MUSIC: ColorMode.EFFECT,
+    CeilingLightColorMode.UNKNOWN: ColorMode.OFF,
+}
 
-class SwitchbotCeilingLight(SwitchbotBaseLight):
-    """Representation of a Switchbot bulb."""
+
+class SwitchbotCeilingLight(SwitchbotSequenceBaseLight):
+    """Representation of a Switchbot ceiling light."""
 
     @property
     def color_modes(self) -> set[ColorMode]:
         """Return the supported color modes."""
         return {ColorMode.COLOR_TEMP}
 
+    @property
+    def color_mode(self) -> ColorMode:
+        """Return the current color mode."""
+        device_mode = CeilingLightColorMode(self._get_adv_value("color_mode") or 10)
+        return _CEILING_LIGHT_COLOR_MODE_MAP.get(device_mode, ColorMode.OFF)
+
+    @update_after_operation
     async def turn_on(self) -> bool:
         """Turn device on."""
         result = await self._send_command(CEILING_LIGHT_ON_KEY)
-        ret = self._check_command_result(result, 0, {0x01})
-        self._override_state({"isOn": True})
-        self._fire_callbacks()
-        return ret
+        return self._check_command_result(result, 0, {1})
 
+    @update_after_operation
     async def turn_off(self) -> bool:
         """Turn device off."""
         result = await self._send_command(CEILING_LIGHT_OFF_KEY)
-        ret = self._check_command_result(result, 0, {0x01})
-        self._override_state({"isOn": False})
-        self._fire_callbacks()
-        return ret
+        return self._check_command_result(result, 0, {1})
 
+    @update_after_operation
     async def set_brightness(self, brightness: int) -> bool:
         """Set brightness."""
         assert 0 <= brightness <= 100, "Brightness must be between 0 and 100"
-        result = await self._send_command(f"{BRIGHTNESS_KEY}{brightness:02X}0FA1")
-        ret = self._check_command_result(result, 0, {0x01})
-        self._override_state({"brightness": brightness, "isOn": True})
-        self._fire_callbacks()
-        return ret
+        color_temp = self._state.get("cw", DEFAULT_COLOR_TEMP)
+        result = await self._send_command(
+            f"{BRIGHTNESS_KEY}{brightness:02X}{color_temp:04X}"
+        )
+        return self._check_command_result(result, 0, {1})
 
+    @update_after_operation
     async def set_color_temp(self, brightness: int, color_temp: int) -> bool:
         """Set color temp."""
         assert 0 <= brightness <= 100, "Brightness must be between 0 and 100"
-        assert 2700 <= color_temp <= 6500, "Color Temp must be between 0 and 100"
+        assert 2700 <= color_temp <= 6500, "Color Temp must be between 2700 and 6500"
         result = await self._send_command(
             f"{CW_BRIGHTNESS_KEY}{brightness:02X}{color_temp:04X}"
         )
-        ret = self._check_command_result(result, 0, {0x01})
-        self._state["cw"] = color_temp
-        self._override_state({"brightness": brightness, "isOn": True})
-        self._fire_callbacks()
-        return ret
-
-    async def set_rgb(self, brightness: int, r: int, g: int, b: int) -> bool:
-        """Set rgb."""
-        # Not supported on this device
+        return self._check_command_result(result, 0, {1})
+
+    async def get_basic_info(self) -> dict[str, Any] | None:
+        """Get device basic settings."""
+        if not (_data := await self._get_basic_info(DEVICE_GET_BASIC_SETTINGS_KEY)):
+            return None
+        if not (_version_info := await self._get_basic_info(DEVICE_GET_VERSION_KEY)):
+            return None
+
+        _LOGGER.debug(
+            "data: %s, version info: %s, address: %s",
+            _data,
+            _version_info,
+            self._device.address,
+        )
+
+        self._state["cw"] = int.from_bytes(_data[3:5], "big")
+
+        return {
+            "isOn": bool(_data[1] & 0b10000000),
+            "color_mode": _data[1] & 0b01000000,
+            "brightness": _data[2] & 0b01111111,
+            "cw": self._state["cw"],
+            "firmware": _version_info[2] / 10.0,
+        }

switchbot/devices/device.py

@@ -8,7 +8,6 @@ import logging
 import time
 from collections.abc import Callable
 from dataclasses import replace
-from enum import Enum
 from typing import Any, TypeVar, cast
 from uuid import UUID
 
@@ -29,6 +28,7 @@ from ..api_config import SWITCHBOT_APP_API_BASE_URL, SWITCHBOT_APP_CLIENT_ID
 from ..const import (
     DEFAULT_RETRY_COUNT,
     DEFAULT_SCAN_TIMEOUT,
+    ColorMode,  # noqa: F401
     SwitchbotAccountConnectionError,
     SwitchbotApiError,
     SwitchbotAuthenticationError,
@@ -60,13 +60,6 @@ DBUS_ERROR_BACKOFF_TIME = 0.25
 DISCONNECT_DELAY = 8.5
 
 
-class ColorMode(Enum):
-    OFF = 0
-    COLOR_TEMP = 1
-    RGB = 2
-    EFFECT = 3
-
-
 # If the scanner is in passive mode, we
 # need to poll the device to get the
 # battery and a few rarely updating
@@ -210,6 +203,54 @@ class SwitchbotBaseDevice:
         key_suffix = key[4:]
         return KEY_PASSWORD_PREFIX + key_action + self._password_encoded + key_suffix
 
+    async def _send_command_locked_with_retry(
+        self, key: str, command: bytes, retry: int, max_attempts: int
+    ) -> bytes | None:
+        for attempt in range(max_attempts):
+            try:
+                return await self._send_command_locked(key, command)
+            except BleakNotFoundError:
+                _LOGGER.error(
+                    "%s: device not found, no longer in range, or poor RSSI: %s",
+                    self.name,
+                    self.rssi,
+                    exc_info=True,
+                )
+                raise
+            except CharacteristicMissingError as ex:
+                if attempt == retry:
+                    _LOGGER.error(
+                        "%s: characteristic missing: %s; Stopping trying; RSSI: %s",
+                        self.name,
+                        ex,
+                        self.rssi,
+                        exc_info=True,
+                    )
+                    raise
+
+                _LOGGER.debug(
+                    "%s: characteristic missing: %s; RSSI: %s",
+                    self.name,
+                    ex,
+                    self.rssi,
+                    exc_info=True,
+                )
+            except BLEAK_RETRY_EXCEPTIONS:
+                if attempt == retry:
+                    _LOGGER.error(
+                        "%s: communication failed; Stopping trying; RSSI: %s",
+                        self.name,
+                        self.rssi,
+                        exc_info=True,
+                    )
+                    raise
+
+                _LOGGER.debug(
+                    "%s: communication failed with:", self.name, exc_info=True
+                )
+
+        raise RuntimeError("Unreachable")
+
     async def _send_command(self, key: str, retry: int | None = None) -> bytes | None:
         """Send command to device and read response."""
         if retry is None:
@@ -224,50 +265,9 @@ class SwitchbotBaseDevice:
                 self.rssi,
             )
         async with self._operation_lock:
-            for attempt in range(max_attempts):
-                try:
-                    return await self._send_command_locked(key, command)
-                except BleakNotFoundError:
-                    _LOGGER.error(
-                        "%s: device not found, no longer in range, or poor RSSI: %s",
-                        self.name,
-                        self.rssi,
-                        exc_info=True,
-                    )
-                    raise
-                except CharacteristicMissingError as ex:
-                    if attempt == retry:
-                        _LOGGER.error(
-                            "%s: characteristic missing: %s; Stopping trying; RSSI: %s",
-                            self.name,
-                            ex,
-                            self.rssi,
-                            exc_info=True,
-                        )
-                        raise
-
-                    _LOGGER.debug(
-                        "%s: characteristic missing: %s; RSSI: %s",
-                        self.name,
-                        ex,
-                        self.rssi,
-                        exc_info=True,
-                    )
-                except BLEAK_RETRY_EXCEPTIONS:
-                    if attempt == retry:
-                        _LOGGER.error(
-                            "%s: communication failed; Stopping trying; RSSI: %s",
-                            self.name,
-                            self.rssi,
-                            exc_info=True,
-                        )
-                        raise
-
-                    _LOGGER.debug(
-                        "%s: communication failed with:", self.name, exc_info=True
-                    )
-
-        raise RuntimeError("Unreachable")
+            return await self._send_command_locked_with_retry(
+                key, command, retry, max_attempts
+            )
 
     @property
     def name(self) -> str:
@@ -839,37 +839,73 @@ class SwitchbotEncryptedDevice(SwitchbotDevice):
         if not encrypt:
             return await super()._send_command(key[:2] + "000000" + key[2:], retry)
 
-        result = await self._ensure_encryption_initialized()
-        if not result:
-            _LOGGER.error("Failed to initialize encryption")
-            return None
+        if retry is None:
+            retry = self._retry_count
 
-        encrypted = (
-            key[:2] + self._key_id + self._iv[0:2].hex() + self._encrypt(key[2:])
-        )
-        result = await super()._send_command(encrypted, retry)
-        return result[:1] + self._decrypt(result[4:])
+        if self._operation_lock.locked():
+            _LOGGER.debug(
+                "%s: Operation already in progress, waiting for it to complete; RSSI: %s",
+                self.name,
+                self.rssi,
+            )
+
+        async with self._operation_lock:
+            if not (result := await self._ensure_encryption_initialized()):
+                _LOGGER.error("Failed to initialize encryption")
+                return None
+
+            encrypted = (
+                key[:2] + self._key_id + self._iv[0:2].hex() + self._encrypt(key[2:])
+            )
+            command = bytearray.fromhex(self._commandkey(encrypted))
+            _LOGGER.debug("%s: Scheduling command %s", self.name, command.hex())
+            max_attempts = retry + 1
+
+            result = await self._send_command_locked_with_retry(
+                encrypted, command, retry, max_attempts
+            )
+            if result is None:
+                return None
+            return result[:1] + self._decrypt(result[4:])
 
     async def _ensure_encryption_initialized(self) -> bool:
+        """Ensure encryption is initialized, must be called with operation lock held."""
+        assert self._operation_lock.locked(), "Operation lock must be held"
+
         if self._iv is not None:
             return True
 
-        result = await self._send_command(
-            COMMAND_GET_CK_IV + self._key_id, encrypt=False
+        _LOGGER.debug("%s: Initializing encryption", self.name)
+        # Call parent's _send_command_locked_with_retry directly since we already hold the lock
+        key = COMMAND_GET_CK_IV + self._key_id
+        command = bytearray.fromhex(self._commandkey(key[:2] + "000000" + key[2:]))
+
+        result = await self._send_command_locked_with_retry(
+            key[:2] + "000000" + key[2:],
+            command,
+            self._retry_count,
+            self._retry_count + 1,
         )
-        ok = self._check_command_result(result, 0, {1})
-        if ok:
+        if result is None:
+            return False
+
+        if ok := self._check_command_result(result, 0, {1}):
             self._iv = result[4:]
+            self._cipher = None  # Reset cipher when IV changes
+            _LOGGER.debug("%s: Encryption initialized successfully", self.name)
 
         return ok
 
     async def _execute_disconnect(self) -> None:
-        await super()._execute_disconnect()
-        self._iv = None
-        self._cipher = None
+        async with self._connect_lock:
+            self._iv = None
+            self._cipher = None
+            await self._execute_disconnect_with_lock()
 
     def _get_cipher(self) -> Cipher:
         if self._cipher is None:
+            if self._iv is None:
+                raise RuntimeError("Cannot create cipher: IV is None")
             self._cipher = Cipher(
                 algorithms.AES128(self._encryption_key), modes.CTR(self._iv)
             )
@@ -878,12 +914,16 @@ class SwitchbotEncryptedDevice(SwitchbotDevice):
     def _encrypt(self, data: str) -> str:
         if len(data) == 0:
             return ""
+        if self._iv is None:
+            raise RuntimeError("Cannot encrypt: IV is None")
         encryptor = self._get_cipher().encryptor()
         return (encryptor.update(bytearray.fromhex(data)) + encryptor.finalize()).hex()
 
     def _decrypt(self, data: bytearray) -> bytes:
         if len(data) == 0:
             return b""
+        if self._iv is None:
+            raise RuntimeError("Cannot decrypt: IV is None")
         decryptor = self._get_cipher().decryptor()
         return decryptor.update(data) + decryptor.finalize()
 

switchbot/devices/light_strip.py

@@ -1,9 +1,19 @@
 from __future__ import annotations
 
 import logging
+from typing import Any
 
+from bleak.backends.device import BLEDevice
+
+from ..const import SwitchbotModel
+from ..const.light import ColorMode, StripLightColorMode
 from .base_light import SwitchbotSequenceBaseLight
-from .device import REQ_HEADER, ColorMode
+from .device import (
+    REQ_HEADER,
+    SwitchbotEncryptedDevice,
+    SwitchbotOperationError,
+    update_after_operation,
+)
 
 STRIP_COMMMAND_HEADER = "4901"
 STRIP_REQUEST = f"{REQ_HEADER}4A01"
@@ -12,11 +22,103 @@ STRIP_COMMAND = f"{REQ_HEADER}{STRIP_COMMMAND_HEADER}"
 # Strip keys
 STRIP_ON_KEY = f"{STRIP_COMMAND}01"
 STRIP_OFF_KEY = f"{STRIP_COMMAND}02"
+COLOR_TEMP_KEY = f"{STRIP_COMMAND}11"
 RGB_BRIGHTNESS_KEY = f"{STRIP_COMMAND}12"
 BRIGHTNESS_KEY = f"{STRIP_COMMAND}14"
 
+DEVICE_GET_VERSION_KEY = "570003"
+
 _LOGGER = logging.getLogger(__name__)
 
+EFFECT_DICT = {
+    "Christmas": [
+        "570F49070200033C01",
+        "570F490701000600009902006D0EFF0021",
+        "570F490701000603009902006D0EFF0021",
+    ],
+    "Halloween": ["570F49070200053C04", "570F490701000300FF6A009E00ED00EA0F"],
+    "Sunset": [
+        "570F49070200033C3C",
+        "570F490701000900FF9000ED8C04DD5800",
+        "570F490701000903FF2E008E0B004F0500",
+        "570F4907010009063F0010270056140033",
+    ],
+    "Vitality": [
+        "570F49070200053C02",
+        "570F490701000600C5003FD9530AEC9800",
+        "570F490701000603FFDF0000895500468B",
+    ],
+    "Flashing": [
+        "570F49070200053C02",
+        "570F4907010006000000FF00FF00FF0000",
+        "570F490701000603FFFF0000FFFFA020F0",
+    ],
+    "Strobe": ["570F49070200043C02", "570F490701000300FF00E19D70FFFF0515"],
+    "Fade": [
+        "570F49070200043C04",
+        "570F490701000500FF5481FF00E19D70FF",
+        "570F490701000503FF0515FF7FEB",
+    ],
+    "Smooth": [
+        "570F49070200033C02",
+        "570F4907010007000036FC00F6FF00ED13",
+        "570F490701000703F6FF00FF8300FF0800",
+        "570F490701000706FF00E1",
+    ],
+    "Forest": [
+        "570F49070200033C06",
+        "570F490701000400006400228B223CB371",
+        "570F49070100040390EE90",
+    ],
+    "Ocean": [
+        "570F49070200033C06",
+        "570F4907010007004400FF0061FF007BFF",
+        "570F490701000703009DFF00B2FF00CBFF",
+        "570F49070100070600E9FF",
+    ],
+    "Autumn": [
+        "570F49070200043C05",
+        "570F490701000700D10035922D13A16501",
+        "570F490701000703AB9100DD8C00F4AA29",
+        "570F490701000706E8D000",
+    ],
+    "Cool": [
+        "570F49070200043C04",
+        "570F490701000600001A63006C9A00468B",
+        "570F490701000603009DA50089BE4378B6",
+    ],
+    "Flow": [
+        "570F49070200033C02",
+        "570F490701000600FF00D8E100FFAA00FF",
+        "570F4907010006037F00FF5000FF1900FF",
+    ],
+    "Relax": [
+        "570F49070200033C03",
+        "570F490701000400FF8C00FF7200FF1D00",
+        "570F490701000403FF5500",
+    ],
+    "Modern": [
+        "570F49070200043C03",
+        "570F49070100060089231A5F8969829E5A",
+        "570F490701000603BCB05EEDBE5AFF9D60",
+    ],
+    "Rose": [
+        "570F49070200043C04",
+        "570F490701000500FF1969BC215F7C0225",
+        "570F490701000503600C2B35040C",
+    ],
+}
+
+# Private mapping from device-specific color modes to original ColorMode enum
+_STRIP_LIGHT_COLOR_MODE_MAP = {
+    StripLightColorMode.RGB: ColorMode.RGB,
+    StripLightColorMode.SCENE: ColorMode.EFFECT,
+    StripLightColorMode.MUSIC: ColorMode.EFFECT,
+    StripLightColorMode.CONTROLLER: ColorMode.EFFECT,
+    StripLightColorMode.COLOR_TEMP: ColorMode.COLOR_TEMP,
+    StripLightColorMode.UNKNOWN: ColorMode.OFF,
+}
+
 
 class SwitchbotLightStrip(SwitchbotSequenceBaseLight):
     """Representation of a Switchbot light strip."""
@@ -26,35 +128,37 @@ class SwitchbotLightStrip(SwitchbotSequenceBaseLight):
         """Return the supported color modes."""
         return {ColorMode.RGB}
 
-    async def update(self) -> None:
-        """Update state of device."""
-        result = await self._send_command(STRIP_REQUEST)
-        self._update_state(result)
-        await super().update()
+    @property
+    def color_mode(self) -> ColorMode:
+        """Return the current color mode."""
+        device_mode = StripLightColorMode(self._get_adv_value("color_mode") or 10)
+        return _STRIP_LIGHT_COLOR_MODE_MAP.get(device_mode, ColorMode.OFF)
+
+    @property
+    def get_effect_list(self) -> list[str]:
+        """Return the list of supported effects."""
+        return list(EFFECT_DICT.keys())
 
+    @update_after_operation
     async def turn_on(self) -> bool:
         """Turn device on."""
         result = await self._send_command(STRIP_ON_KEY)
-        self._update_state(result)
-        return self._check_command_result(result, 1, {0x80})
+        return self._check_command_result(result, 0, {1})
 
+    @update_after_operation
     async def turn_off(self) -> bool:
         """Turn device off."""
         result = await self._send_command(STRIP_OFF_KEY)
-        self._update_state(result)
-        return self._check_command_result(result, 1, {0x00})
+        return self._check_command_result(result, 0, {1})
 
+    @update_after_operation
     async def set_brightness(self, brightness: int) -> bool:
         """Set brightness."""
         assert 0 <= brightness <= 100, "Brightness must be between 0 and 100"
         result = await self._send_command(f"{BRIGHTNESS_KEY}{brightness:02X}")
-        self._update_state(result)
-        return self._check_command_result(result, 1, {0x80})
-
-    async def set_color_temp(self, brightness: int, color_temp: int) -> bool:
-        """Set color temp."""
-        # not supported on this device
+        return self._check_command_result(result, 0, {1})
 
+    @update_after_operation
     async def set_rgb(self, brightness: int, r: int, g: int, b: int) -> bool:
         """Set rgb."""
         assert 0 <= brightness <= 100, "Brightness must be between 0 and 100"
@@ -64,21 +168,92 @@ class SwitchbotLightStrip(SwitchbotSequenceBaseLight):
         result = await self._send_command(
             f"{RGB_BRIGHTNESS_KEY}{brightness:02X}{r:02X}{g:02X}{b:02X}"
         )
-        self._update_state(result)
-        return self._check_command_result(result, 1, {0x80})
-
-    def _update_state(self, result: bytes | None) -> None:
-        """Update device state."""
-        if not result or len(result) < 10:
-            return
-        self._state["r"] = result[3]
-        self._state["g"] = result[4]
-        self._state["b"] = result[5]
-        self._override_state(
-            {
-                "isOn": result[1] == 0x80,
-                "color_mode": result[10],
-            }
+        return self._check_command_result(result, 0, {1})
+
+    @update_after_operation
+    async def set_effect(self, effect: str) -> bool:
+        """Set effect."""
+        effect_template = EFFECT_DICT.get(effect)
+        if not effect_template:
+            raise SwitchbotOperationError(f"Effect {effect} not supported")
+        result = await self._send_multiple_commands(effect_template)
+        if result:
+            self._override_state({"effect": effect})
+        return result
+
+    async def get_basic_info(
+        self,
+        device_get_basic_info: str = STRIP_REQUEST,
+        device_get_version_info: str = DEVICE_GET_VERSION_KEY,
+    ) -> dict[str, Any] | None:
+        """Get device basic settings."""
+        if not (_data := await self._get_basic_info(device_get_basic_info)):
+            return None
+        if not (_version_info := await self._get_basic_info(device_get_version_info)):
+            return None
+
+        _LOGGER.debug(
+            "data: %s, version info: %s, address: %s",
+            _data,
+            _version_info,
+            self._device.address,
+        )
+
+        self._state["r"] = _data[3]
+        self._state["g"] = _data[4]
+        self._state["b"] = _data[5]
+        self._state["cw"] = int.from_bytes(_data[7:9], "big")
+
+        return {
+            "isOn": bool(_data[1] & 0b10000000),
+            "brightness": _data[2] & 0b01111111,
+            "r": self._state["r"],
+            "g": self._state["g"],
+            "b": self._state["b"],
+            "cw": self._state["cw"],
+            "color_mode": _data[10] & 0b00001111,
+            "firmware": _version_info[2] / 10.0,
+        }
+
+
+class SwitchbotStripLight3(SwitchbotEncryptedDevice, SwitchbotLightStrip):
+    """Support for switchbot strip light3 and floor lamp."""
+
+    def __init__(
+        self,
+        device: BLEDevice,
+        key_id: str,
+        encryption_key: str,
+        interface: int = 0,
+        model: SwitchbotModel = SwitchbotModel.STRIP_LIGHT_3,
+        **kwargs: Any,
+    ) -> None:
+        super().__init__(device, key_id, encryption_key, model, interface, **kwargs)
+
+    @classmethod
+    async def verify_encryption_key(
+        cls,
+        device: BLEDevice,
+        key_id: str,
+        encryption_key: str,
+        model: SwitchbotModel = SwitchbotModel.STRIP_LIGHT_3,
+        **kwargs: Any,
+    ) -> bool:
+        return await super().verify_encryption_key(
+            device, key_id, encryption_key, model, **kwargs
+        )
+
+    @property
+    def color_modes(self) -> set[ColorMode]:
+        """Return the supported color modes."""
+        return {ColorMode.RGB, ColorMode.COLOR_TEMP}
+
+    @update_after_operation
+    async def set_color_temp(self, brightness: int, color_temp: int) -> bool:
+        """Set color temp."""
+        assert 0 <= brightness <= 100
+        assert self.min_temp <= color_temp <= self.max_temp
+        result = await self._send_command(
+            f"{COLOR_TEMP_KEY}{brightness:02X}{color_temp:04X}"
         )
-        _LOGGER.debug("%s: update state: %s = %s", self.name, result.hex(), self._state)
-        self._fire_callbacks()
+        return self._check_command_result(result, 0, {1})

tests/__init__.py

@@ -6,7 +6,7 @@ from switchbot import SwitchbotModel
 @dataclass
 class AdvTestCase:
     manufacturer_data: bytes | None
-    service_data: bytes
+    service_data: bytes | None
     data: dict
     model: str | bytes
     modelFriendlyName: str

tests/test_adv_parser.py

@@ -3290,6 +3290,84 @@ def test_humidifer_with_empty_data() -> None:
             "Garage Door Opener",
             SwitchbotModel.GARAGE_DOOR_OPENER,
         ),
+        AdvTestCase(
+            b'\xc0N0\xe0U\x9a\x85\x9e"\xd0\x00\x00\x00\x00\x00\x00\x12\x91\x00',
+            b"\x00\x00\x00\x00\x10\xd0\xb1",
+            {
+                "sequence_number": 133,
+                "isOn": True,
+                "brightness": 30,
+                "delay": False,
+                "network_state": 2,
+                "color_mode": 2,
+                "cw": 4753,
+            },
+            b"\x00\x10\xd0\xb1",
+            "Strip Light 3",
+            SwitchbotModel.STRIP_LIGHT_3,
+        ),
+        AdvTestCase(
+            b'\xa0\x85\xe3e,\x06P\xaa"\xd4\x00\x00\x00\x00\x00\x00\r\x93\x00',
+            b"\x00\x00\x00\x00\x10\xd0\xb0",
+            {
+                "sequence_number": 80,
+                "isOn": True,
+                "brightness": 42,
+                "delay": False,
+                "network_state": 2,
+                "color_mode": 2,
+                "cw": 3475,
+            },
+            b"\x00\x10\xd0\xb0",
+            "Floor Lamp",
+            SwitchbotModel.FLOOR_LAMP,
+        ),
+        AdvTestCase(
+            b"\xef\xfe\xfb\x9d\x10\xfe\n\x01\x18\xf3$",
+            b"q\x00",
+            {
+                "brightness": 1,
+                "color_mode": 1,
+                "cw": 6387,
+                "isOn": False,
+                "sequence_number": 10,
+            },
+            "q",
+            "Ceiling Light",
+            SwitchbotModel.CEILING_LIGHT,
+        ),
+        AdvTestCase(
+            b'|,g\xc8\x15&jR"l\x00\x00\x00\x00\x00\x00',
+            b"r\x00d",
+            {
+                "brightness": 82,
+                "color_mode": 2,
+                "delay": False,
+                "isOn": False,
+                "sequence_number": 106,
+                "network_state": 2,
+            },
+            "r",
+            "Light Strip",
+            SwitchbotModel.LIGHT_STRIP,
+        ),
+        AdvTestCase(
+            b"@L\xca\xa7_\x12\x02\x81\x12\x00\x00",
+            b"u\x00d",
+            {
+                "brightness": 1,
+                "color_mode": 2,
+                "delay": False,
+                "isOn": True,
+                "loop_index": 0,
+                "preset": False,
+                "sequence_number": 2,
+                "speed": 0,
+            },
+            "u",
+            "Color Bulb",
+            SwitchbotModel.COLOR_BULB,
+        ),
     ],
 )
 def test_adv_active(test_case: AdvTestCase) -> None:
@@ -3376,6 +3454,70 @@ def test_adv_active(test_case: AdvTestCase) -> None:
             "Garage Door Opener",
             SwitchbotModel.GARAGE_DOOR_OPENER,
         ),
+        AdvTestCase(
+            b'\xc0N0\xe0U\x9a\x85\x9e"\xd0\x00\x00\x00\x00\x00\x00\x12\x91\x00',
+            None,
+            {
+                "sequence_number": 133,
+                "isOn": True,
+                "brightness": 30,
+                "delay": False,
+                "network_state": 2,
+                "color_mode": 2,
+                "cw": 4753,
+            },
+            b"\x00\x10\xd0\xb1",
+            "Strip Light 3",
+            SwitchbotModel.STRIP_LIGHT_3,
+        ),
+        AdvTestCase(
+            b'\xa0\x85\xe3e,\x06P\xaa"\xd4\x00\x00\x00\x00\x00\x00\r\x93\x00',
+            None,
+            {
+                "sequence_number": 80,
+                "isOn": True,
+                "brightness": 42,
+                "delay": False,
+                "network_state": 2,
+                "color_mode": 2,
+                "cw": 3475,
+            },
+            b"\x00\x10\xd0\xb0",
+            "Floor Lamp",
+            SwitchbotModel.FLOOR_LAMP,
+        ),
+        AdvTestCase(
+            b'|,g\xc8\x15&jR"l\x00\x00\x00\x00\x00\x00',
+            None,
+            {
+                "brightness": 82,
+                "color_mode": 2,
+                "delay": False,
+                "isOn": False,
+                "sequence_number": 106,
+                "network_state": 2,
+            },
+            "r",
+            "Light Strip",
+            SwitchbotModel.LIGHT_STRIP,
+        ),
+        AdvTestCase(
+            b"@L\xca\xa7_\x12\x02\x81\x12\x00\x00",
+            None,
+            {
+                "brightness": 1,
+                "color_mode": 2,
+                "delay": False,
+                "isOn": True,
+                "loop_index": 0,
+                "preset": False,
+                "sequence_number": 2,
+                "speed": 0,
+            },
+            "u",
+            "Color Bulb",
+            SwitchbotModel.COLOR_BULB,
+        ),
     ],
 )
 def test_adv_passive(test_case: AdvTestCase) -> None:
@@ -3464,6 +3606,46 @@ def test_adv_passive(test_case: AdvTestCase) -> None:
             "Garage Door Opener",
             SwitchbotModel.GARAGE_DOOR_OPENER,
         ),
+        AdvTestCase(
+            None,
+            b"\x00\x00\x00\x00\x10\xd0\xb1",
+            {},
+            b"\x00\x10\xd0\xb1",
+            "Strip Light 3",
+            SwitchbotModel.STRIP_LIGHT_3,
+        ),
+        AdvTestCase(
+            None,
+            b"\x00\x00\x00\x00\x10\xd0\xb0",
+            {},
+            b"\x00\x10\xd0\xb0",
+            "Floor Lamp",
+            SwitchbotModel.FLOOR_LAMP,
+        ),
+        AdvTestCase(
+            None,
+            b"q\x00",
+            {},
+            "q",
+            "Ceiling Light",
+            SwitchbotModel.CEILING_LIGHT,
+        ),
+        AdvTestCase(
+            None,
+            b"r\x00d",
+            {},
+            "r",
+            "Light Strip",
+            SwitchbotModel.LIGHT_STRIP,
+        ),
+        AdvTestCase(
+            None,
+            b"u\x00d",
+            {},
+            "u",
+            "Color Bulb",
+            SwitchbotModel.COLOR_BULB,
+        ),
     ],
 )
 def test_adv_with_empty_data(test_case: AdvTestCase) -> None:

tests/test_bulb.py

@@ -0,0 +1,220 @@
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from bleak.backends.device import BLEDevice
+
+from switchbot import SwitchBotAdvertisement, SwitchbotModel
+from switchbot.const.light import ColorMode
+from switchbot.devices import bulb
+from switchbot.devices.device import SwitchbotOperationError
+
+from .test_adv_parser import generate_ble_device
+
+
+def create_device_for_command_testing(init_data: dict | None = None):
+    ble_device = generate_ble_device("aa:bb:cc:dd:ee:ff", "any")
+    device = bulb.SwitchbotBulb(ble_device)
+    device.update_from_advertisement(make_advertisement_data(ble_device, init_data))
+    device._send_command = AsyncMock()
+    device._check_command_result = MagicMock()
+    device.update = AsyncMock()
+    return device
+
+
+def make_advertisement_data(ble_device: BLEDevice, init_data: dict | None = None):
+    """Set advertisement data with defaults."""
+    if init_data is None:
+        init_data = {}
+
+    return SwitchBotAdvertisement(
+        address="aa:bb:cc:dd:ee:ff",
+        data={
+            "rawAdvData": b"u\x00d",
+            "data": {
+                "brightness": 1,
+                "color_mode": 2,
+                "delay": False,
+                "isOn": True,
+                "loop_index": 0,
+                "preset": False,
+                "sequence_number": 2,
+                "speed": 0,
+            }
+            | init_data,
+            "isEncrypted": False,
+            "model": "u",
+            "modelFriendlyName": "Color Bulb",
+            "modelName": SwitchbotModel.COLOR_BULB,
+        },
+        device=ble_device,
+        rssi=-80,
+        active=True,
+    )
+
+
+@pytest.mark.asyncio
+async def test_default_info():
+    """Test default initialization of the color bulb."""
+    device = create_device_for_command_testing()
+
+    assert device.rgb is None
+
+    device._state = {"r": 30, "g": 0, "b": 0, "cw": 3200}
+
+    assert device.is_on() is True
+    assert device.on is True
+    assert device.color_mode == ColorMode.RGB
+    assert device.color_modes == {ColorMode.RGB, ColorMode.COLOR_TEMP}
+    assert device.rgb == (30, 0, 0)
+    assert device.color_temp == 3200
+    assert device.brightness == 1
+    assert device.min_temp == 2700
+    assert device.max_temp == 6500
+    assert device.get_effect_list == list(bulb.EFFECT_DICT.keys())
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("basic_info", "version_info"), [(True, False), (False, True), (False, False)]
+)
+async def test_get_basic_info_returns_none(basic_info, version_info):
+    device = create_device_for_command_testing()
+
+    async def mock_get_basic_info(arg):
+        if arg == bulb.DEVICE_GET_BASIC_SETTINGS_KEY:
+            return basic_info
+        if arg == bulb.DEVICE_GET_VERSION_KEY:
+            return version_info
+        return None
+
+    device._get_basic_info = AsyncMock(side_effect=mock_get_basic_info)
+
+    assert await device.get_basic_info() is None
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("info_data", "result"),
+    [
+        (
+            {
+                "basic_info": b"\x01\x80\x01\xff\x91\x96\x00\x00\xff\xff\x02",
+                "version_info": b"\x01\x01\x11",
+            },
+            [True, 1, 255, 145, 150, 0, 2, 1.7],
+        ),
+        (
+            {
+                "basic_info": b"\x01\x80;\x00\x00\x00\x0c\x99\xff\xff\x01",
+                "version_info": b"\x01\x01\x11",
+            },
+            [True, 59, 0, 0, 0, 3225, 1, 1.7],
+        ),
+        (
+            {
+                "basic_info": b"\x01\x80\t!7\xff\x00\x00\xff\xff\x02",
+                "version_info": b"\x01\x01\x11",
+            },
+            [True, 9, 33, 55, 255, 0, 2, 1.7],
+        ),
+    ],
+)
+async def test_get_basic_info(info_data, result):
+    """Test getting basic info from the color bulb."""
+    device = create_device_for_command_testing()
+
+    async def mock_get_basic_info(args: str) -> list[int] | None:
+        if args == bulb.DEVICE_GET_BASIC_SETTINGS_KEY:
+            return info_data["basic_info"]
+        if args == bulb.DEVICE_GET_VERSION_KEY:
+            return info_data["version_info"]
+        return None
+
+    device._get_basic_info = AsyncMock(side_effect=mock_get_basic_info)
+    info = await device.get_basic_info()
+
+    assert info["isOn"] is result[0]
+    assert info["brightness"] == result[1]
+    assert info["r"] == result[2]
+    assert info["g"] == result[3]
+    assert info["b"] == result[4]
+    assert info["cw"] == result[5]
+    assert info["color_mode"] == result[6]
+    assert info["firmware"] == result[7]
+
+
+@pytest.mark.asyncio
+async def test_set_color_temp():
+    """Test setting color temperature."""
+    device = create_device_for_command_testing()
+
+    await device.set_color_temp(50, 3000)
+
+    device._send_command.assert_called_with(f"{bulb.CW_BRIGHTNESS_KEY}320BB8")
+
+
+@pytest.mark.asyncio
+async def test_turn_on():
+    """Test turning on the color bulb."""
+    device = create_device_for_command_testing({"isOn": True})
+
+    await device.turn_on()
+
+    device._send_command.assert_called_with(bulb.BULB_ON_KEY)
+
+    assert device.is_on() is True
+
+
+@pytest.mark.asyncio
+async def test_turn_off():
+    """Test turning off the color bulb."""
+    device = create_device_for_command_testing({"isOn": False})
+
+    await device.turn_off()
+
+    device._send_command.assert_called_with(bulb.BULB_OFF_KEY)
+
+    assert device.is_on() is False
+
+
+@pytest.mark.asyncio
+async def test_set_brightness():
+    """Test setting brightness."""
+    device = create_device_for_command_testing()
+
+    await device.set_brightness(75)
+
+    device._send_command.assert_called_with(f"{bulb.BRIGHTNESS_KEY}4B")
+
+
+@pytest.mark.asyncio
+async def test_set_rgb():
+    """Test setting RGB values."""
+    device = create_device_for_command_testing()
+
+    await device.set_rgb(100, 255, 128, 64)
+
+    device._send_command.assert_called_with(f"{bulb.RGB_BRIGHTNESS_KEY}64FF8040")
+
+
+@pytest.mark.asyncio
+async def test_set_effect_with_invalid_effect():
+    """Test setting an invalid effect."""
+    device = create_device_for_command_testing()
+
+    with pytest.raises(
+        SwitchbotOperationError, match="Effect invalid_effect not supported"
+    ):
+        await device.set_effect("invalid_effect")
+
+
+@pytest.mark.asyncio
+async def test_set_effect_with_valid_effect():
+    """Test setting a valid effect."""
+    device = create_device_for_command_testing()
+
+    await device.set_effect("Colorful")
+
+    device._send_command.assert_called_with(bulb.EFFECT_DICT["Colorful"])
+
+    assert device.get_effect() == "Colorful"

tests/test_ceiling_light.py

@@ -0,0 +1,179 @@
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from bleak.backends.device import BLEDevice
+
+from switchbot import SwitchBotAdvertisement, SwitchbotModel
+from switchbot.const.light import ColorMode
+from switchbot.devices import ceiling_light
+
+from .test_adv_parser import generate_ble_device
+
+
+def create_device_for_command_testing(init_data: dict | None = None):
+    ble_device = generate_ble_device("aa:bb:cc:dd:ee:ff", "any")
+    device = ceiling_light.SwitchbotCeilingLight(ble_device)
+    device.update_from_advertisement(make_advertisement_data(ble_device, init_data))
+    device._send_command = AsyncMock()
+    device._check_command_result = MagicMock()
+    device.update = AsyncMock()
+    return device
+
+
+def make_advertisement_data(ble_device: BLEDevice, init_data: dict | None = None):
+    """Set advertisement data with defaults."""
+    if init_data is None:
+        init_data = {}
+
+    return SwitchBotAdvertisement(
+        address="aa:bb:cc:dd:ee:ff",
+        data={
+            "rawAdvData": b"q\x00",
+            "data": {
+                "brightness": 1,
+                "color_mode": 1,
+                "cw": 6387,
+                "isOn": False,
+                "sequence_number": 10,
+            }
+            | init_data,
+            "isEncrypted": False,
+            "model": b"q\x00",
+            "modelFriendlyName": "Ceiling Light",
+            "modelName": SwitchbotModel.CEILING_LIGHT,
+        },
+        device=ble_device,
+        rssi=-80,
+        active=True,
+    )
+
+
+@pytest.mark.asyncio
+async def test_default_info():
+    """Test default initialization of the ceiling light."""
+    device = create_device_for_command_testing()
+
+    assert device.rgb is None
+
+    device._state = {"cw": 3200}
+
+    assert device.is_on() is False
+    assert device.on is False
+    assert device.color_mode == ColorMode.COLOR_TEMP
+    assert device.color_modes == {ColorMode.COLOR_TEMP}
+    assert device.color_temp == 3200
+    assert device.brightness == 1
+    assert device.min_temp == 2700
+    assert device.max_temp == 6500
+    assert device.get_effect_list is None
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("basic_info", "version_info"), [(True, False), (False, True), (False, False)]
+)
+async def test_get_basic_info_returns_none(basic_info, version_info):
+    device = create_device_for_command_testing()
+
+    async def mock_get_basic_info(arg):
+        if arg == ceiling_light.DEVICE_GET_BASIC_SETTINGS_KEY:
+            return basic_info
+        if arg == ceiling_light.DEVICE_GET_VERSION_KEY:
+            return version_info
+        return None
+
+    device._get_basic_info = AsyncMock(side_effect=mock_get_basic_info)
+
+    assert await device.get_basic_info() is None
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("info_data", "result"),
+    [
+        (
+            {
+                "basic_info": b"\x01\x80=\x0f\xa1\x00\x01",
+                "version_info": b"\x01d\x15\x0f\x00\x00\x00\x00\x00\x00\x00\n\x00",
+            },
+            [True, 61, 4001, 0, 2.1],
+        ),
+        (
+            {
+                "basic_info": b"\x01\x80\x0e\x12B\x00\x01",
+                "version_info": b"\x01d\x15\x0f\x00\x00\x00\x00\x00\x00\x00\n\x00",
+            },
+            [True, 14, 4674, 0, 2.1],
+        ),
+        (
+            {
+                "basic_info": b"\x01\x00\x0e\x10\x96\x00\x01",
+                "version_info": b"\x01d\x15\x0f\x00\x00\x00\x00\x00\x00\x00\n\x00",
+            },
+            [False, 14, 4246, 0, 2.1],
+        ),
+    ],
+)
+async def test_get_basic_info(info_data, result):
+    """Test getting basic info from the ceiling light."""
+    device = create_device_for_command_testing()
+
+    async def mock_get_basic_info(args: str) -> list[int] | None:
+        if args == ceiling_light.DEVICE_GET_BASIC_SETTINGS_KEY:
+            return info_data["basic_info"]
+        if args == ceiling_light.DEVICE_GET_VERSION_KEY:
+            return info_data["version_info"]
+        return None
+
+    device._get_basic_info = AsyncMock(side_effect=mock_get_basic_info)
+    info = await device.get_basic_info()
+
+    assert info["isOn"] is result[0]
+    assert info["brightness"] == result[1]
+    assert info["cw"] == result[2]
+    assert info["color_mode"] == result[3]
+    assert info["firmware"] == result[4]
+
+
+@pytest.mark.asyncio
+async def test_set_color_temp():
+    """Test setting color temperature."""
+    device = create_device_for_command_testing()
+
+    await device.set_color_temp(50, 3000)
+
+    device._send_command.assert_called_with(f"{ceiling_light.CW_BRIGHTNESS_KEY}320BB8")
+
+
+@pytest.mark.asyncio
+async def test_turn_on():
+    """Test turning on the ceiling light."""
+    device = create_device_for_command_testing({"isOn": True})
+
+    await device.turn_on()
+
+    device._send_command.assert_called_with(ceiling_light.CEILING_LIGHT_ON_KEY)
+
+    assert device.is_on() is True
+
+
+@pytest.mark.asyncio
+async def test_turn_off():
+    """Test turning off the ceiling light."""
+    device = create_device_for_command_testing({"isOn": False})
+
+    await device.turn_off()
+
+    device._send_command.assert_called_with(ceiling_light.CEILING_LIGHT_OFF_KEY)
+
+    assert device.is_on() is False
+
+
+@pytest.mark.asyncio
+async def test_set_brightness():
+    """Test setting brightness."""
+    device = create_device_for_command_testing()
+
+    await device.set_brightness(75)
+
+    device._send_command.assert_called_with(f"{ceiling_light.BRIGHTNESS_KEY}4B0FA1")

tests/test_colormode_imports.py

@@ -0,0 +1,88 @@
+"""Test ColorMode imports for backward compatibility."""
+
+
+def test_colormode_import_from_main_module():
+    """Test that ColorMode can be imported from the main switchbot module."""
+    from switchbot import ColorMode
+
+    # Verify it's the enum we expect
+    assert hasattr(ColorMode, "OFF")
+    assert hasattr(ColorMode, "COLOR_TEMP")
+    assert hasattr(ColorMode, "RGB")
+    assert hasattr(ColorMode, "EFFECT")
+
+    # Verify the values
+    assert ColorMode.OFF.value == 0
+    assert ColorMode.COLOR_TEMP.value == 1
+    assert ColorMode.RGB.value == 2
+    assert ColorMode.EFFECT.value == 3
+
+
+def test_colormode_import_from_device_module():
+    """Test that ColorMode can be imported from switchbot.devices.device for backward compatibility."""
+    from switchbot.devices.device import ColorMode
+
+    # Verify it's the enum we expect
+    assert hasattr(ColorMode, "OFF")
+    assert hasattr(ColorMode, "COLOR_TEMP")
+    assert hasattr(ColorMode, "RGB")
+    assert hasattr(ColorMode, "EFFECT")
+
+    # Verify the values
+    assert ColorMode.OFF.value == 0
+    assert ColorMode.COLOR_TEMP.value == 1
+    assert ColorMode.RGB.value == 2
+    assert ColorMode.EFFECT.value == 3
+
+
+def test_colormode_import_from_const():
+    """Test that ColorMode can be imported from switchbot.const."""
+    from switchbot.const import ColorMode
+
+    # Verify it's the enum we expect
+    assert hasattr(ColorMode, "OFF")
+    assert hasattr(ColorMode, "COLOR_TEMP")
+    assert hasattr(ColorMode, "RGB")
+    assert hasattr(ColorMode, "EFFECT")
+
+    # Verify the values
+    assert ColorMode.OFF.value == 0
+    assert ColorMode.COLOR_TEMP.value == 1
+    assert ColorMode.RGB.value == 2
+    assert ColorMode.EFFECT.value == 3
+
+
+def test_colormode_import_from_const_light():
+    """Test that ColorMode can be imported from switchbot.const.light."""
+    from switchbot.const.light import ColorMode
+
+    # Verify it's the enum we expect
+    assert hasattr(ColorMode, "OFF")
+    assert hasattr(ColorMode, "COLOR_TEMP")
+    assert hasattr(ColorMode, "RGB")
+    assert hasattr(ColorMode, "EFFECT")
+
+    # Verify the values
+    assert ColorMode.OFF.value == 0
+    assert ColorMode.COLOR_TEMP.value == 1
+    assert ColorMode.RGB.value == 2
+    assert ColorMode.EFFECT.value == 3
+
+
+def test_all_colormode_imports_are_same_object():
+    """Test that all ColorMode imports reference the same enum object."""
+    from switchbot import ColorMode as ColorMode1
+    from switchbot.const import ColorMode as ColorMode3
+    from switchbot.const.light import ColorMode as ColorMode4
+    from switchbot.devices.device import ColorMode as ColorMode2
+
+    # They should all be the exact same object
+    assert ColorMode1 is ColorMode2
+    assert ColorMode2 is ColorMode3
+    assert ColorMode3 is ColorMode4
+
+    # And their members should be the same
+    assert ColorMode1.OFF is ColorMode2.OFF
+    assert ColorMode1.COLOR_TEMP is ColorMode3.COLOR_TEMP
+    assert ColorMode1.RGB is ColorMode4.RGB
+    assert ColorMode1.EFFECT is ColorMode2.EFFECT

tests/test_encrypted_device.py

@@ -0,0 +1,367 @@
+"""Tests for SwitchbotEncryptedDevice base class."""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from bleak.exc import BleakDBusError
+
+from switchbot import SwitchbotModel
+from switchbot.devices.device import (
+    SwitchbotEncryptedDevice,
+)
+
+from .test_adv_parser import generate_ble_device
+
+
+class MockEncryptedDevice(SwitchbotEncryptedDevice):
+    """Mock encrypted device for testing."""
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(*args, **kwargs)
+        self.update_count: int = 0
+
+    async def update(self, interface: int | None = None) -> None:
+        self.update_count += 1
+
+
+def create_encrypted_device(
+    model: SwitchbotModel = SwitchbotModel.LOCK,
+) -> MockEncryptedDevice:
+    """Create an encrypted device for testing."""
+    ble_device = generate_ble_device("aa:bb:cc:dd:ee:ff", "Test Device")
+    return MockEncryptedDevice(
+        ble_device, "01", "0123456789abcdef0123456789abcdef", model=model
+    )
+
+
+@pytest.mark.asyncio
+async def test_encrypted_device_init() -> None:
+    """Test encrypted device initialization."""
+    device = create_encrypted_device()
+    assert device._key_id == "01"
+    assert device._encryption_key == bytearray.fromhex(
+        "0123456789abcdef0123456789abcdef"
+    )
+    assert device._iv is None
+    assert device._cipher is None
+
+
+@pytest.mark.asyncio
+async def test_encrypted_device_init_validation() -> None:
+    """Test encrypted device initialization with invalid parameters."""
+    ble_device = generate_ble_device("aa:bb:cc:dd:ee:ff", "Test Device")
+
+    # Test empty key_id
+    with pytest.raises(ValueError, match="key_id is missing"):
+        MockEncryptedDevice(
+            ble_device, "", "0123456789abcdef0123456789abcdef", SwitchbotModel.LOCK
+        )
+
+    # Test invalid key_id length
+    with pytest.raises(ValueError, match="key_id is invalid"):
+        MockEncryptedDevice(
+            ble_device, "1", "0123456789abcdef0123456789abcdef", SwitchbotModel.LOCK
+        )
+
+    # Test empty encryption_key
+    with pytest.raises(ValueError, match="encryption_key is missing"):
+        MockEncryptedDevice(ble_device, "01", "", SwitchbotModel.LOCK)
+
+    # Test invalid encryption_key length
+    with pytest.raises(ValueError, match="encryption_key is invalid"):
+        MockEncryptedDevice(ble_device, "01", "0123456789abcdef", SwitchbotModel.LOCK)
+
+
+@pytest.mark.asyncio
+async def test_send_command_unencrypted() -> None:
+    """Test sending unencrypted command."""
+    device = create_encrypted_device()
+
+    with patch.object(device, "_send_command_locked_with_retry") as mock_send:
+        mock_send.return_value = b"\x01\x00\x00\x00"
+
+        result = await device._send_command("570200", encrypt=False)
+
+        assert result == b"\x01\x00\x00\x00"
+        mock_send.assert_called_once()
+        # Verify the key was padded with zeros for unencrypted command
+        call_args = mock_send.call_args[0]
+        assert call_args[0] == "570000000200"  # Original key with zeros inserted
+
+
+@pytest.mark.asyncio
+async def test_send_command_encrypted_success() -> None:
+    """Test successful encrypted command."""
+    device = create_encrypted_device()
+
+    # Mock the connection and command execution
+    with (
+        patch.object(device, "_send_command_locked_with_retry") as mock_send,
+        patch.object(device, "_decrypt") as mock_decrypt,
+    ):
+        mock_decrypt.return_value = b"decrypted_response"
+
+        # First call is for IV initialization, second is for the actual command
+        mock_send.side_effect = [
+            b"\x01\x00\x00\x00\x12\x34\x56\x78\x9a\xbc\xde\xf0\x12\x34\x56\x78\x9a\xbc\xde\xf0",  # IV response (16 bytes)
+            b"\x01\x00\x00\x00encrypted_response",  # Command response
+        ]
+
+        result = await device._send_command("570200", encrypt=True)
+
+        assert result is not None
+        assert mock_send.call_count == 2
+        # Verify IV was initialized
+        assert device._iv is not None
+
+
+@pytest.mark.asyncio
+async def test_send_command_iv_already_initialized() -> None:
+    """Test sending encrypted command when IV is already initialized."""
+    device = create_encrypted_device()
+
+    # Pre-set the IV
+    device._iv = b"\x12\x34\x56\x78\x9a\xbc\xde\xf0\x12\x34\x56\x78\x9a\xbc\xde\xf0"
+
+    with (
+        patch.object(device, "_send_command_locked_with_retry") as mock_send,
+        patch.object(device, "_encrypt") as mock_encrypt,
+        patch.object(device, "_decrypt") as mock_decrypt,
+    ):
+        mock_encrypt.return_value = (
+            "656e637279707465645f64617461"  # "encrypted_data" in hex
+        )
+        mock_decrypt.return_value = b"decrypted_response"
+        mock_send.return_value = b"\x01\x00\x00\x00encrypted_response"
+
+        result = await device._send_command("570200", encrypt=True)
+
+        assert result == b"\x01decrypted_response"
+        # Should only call once since IV is already initialized
+        mock_send.assert_called_once()
+        mock_encrypt.assert_called_once()
+        mock_decrypt.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_iv_race_condition_during_disconnect() -> None:
+    """Test that commands during disconnect are handled properly."""
+    device = create_encrypted_device()
+
+    # Pre-set the IV
+    device._iv = b"\x12\x34\x56\x78\x9a\xbc\xde\xf0\x12\x34\x56\x78"
+
+    # Mock the connection
+    mock_client = AsyncMock()
+    mock_client.is_connected = True
+    device._client = mock_client
+
+    async def simulate_disconnect() -> None:
+        """Simulate disconnect happening during command execution."""
+        await asyncio.sleep(0.01)  # Small delay
+        await device._execute_disconnect()
+
+    with (
+        patch.object(device, "_send_command_locked_with_retry") as mock_send,
+        patch.object(device, "_ensure_connected"),
+        patch.object(device, "_encrypt") as mock_encrypt,
+        patch.object(device, "_decrypt") as mock_decrypt,
+    ):
+        mock_encrypt.return_value = "656e63727970746564"  # "encrypted" in hex
+        mock_decrypt.return_value = b"response"
+        mock_send.return_value = b"\x01\x00\x00\x00response"
+
+        # Start command and disconnect concurrently
+        command_task = asyncio.create_task(device._send_command("570200"))
+        disconnect_task = asyncio.create_task(simulate_disconnect())
+
+        # Both should complete without error
+        result, _ = await asyncio.gather(
+            command_task, disconnect_task, return_exceptions=True
+        )
+
+        # Command should have completed successfully
+        assert isinstance(result, bytes) or result is None
+        # IV should be cleared after disconnect
+        assert device._iv is None
+
+
+@pytest.mark.asyncio
+async def test_ensure_encryption_initialized_with_lock_held() -> None:
+    """Test that _ensure_encryption_initialized properly handles the operation lock."""
+    device = create_encrypted_device()
+
+    # Acquire the operation lock
+    async with device._operation_lock:
+        with patch.object(device, "_send_command_locked_with_retry") as mock_send:
+            mock_send.return_value = b"\x01\x00\x00\x00\x12\x34\x56\x78\x9a\xbc\xde\xf0\x12\x34\x56\x78\x9a\xbc\xde\xf0"
+
+            result = await device._ensure_encryption_initialized()
+
+            assert result is True
+            assert (
+                device._iv
+                == b"\x12\x34\x56\x78\x9a\xbc\xde\xf0\x12\x34\x56\x78\x9a\xbc\xde\xf0"
+            )
+            assert device._cipher is None  # Should be reset when IV changes
+
+
+@pytest.mark.asyncio
+async def test_ensure_encryption_initialized_failure() -> None:
+    """Test _ensure_encryption_initialized when IV initialization fails."""
+    device = create_encrypted_device()
+
+    async with device._operation_lock:
+        with patch.object(device, "_send_command_locked_with_retry") as mock_send:
+            # Return failure response
+            mock_send.return_value = b"\x00"
+
+            result = await device._ensure_encryption_initialized()
+
+            assert result is False
+            assert device._iv is None
+
+
+@pytest.mark.asyncio
+async def test_encrypt_decrypt_with_valid_iv() -> None:
+    """Test encryption and decryption with valid IV."""
+    device = create_encrypted_device()
+    device._iv = b"\x00" * 16  # Use zeros for predictable test
+
+    # Test encryption
+    encrypted = device._encrypt("48656c6c6f")  # "Hello" in hex
+    assert isinstance(encrypted, str)
+    assert len(encrypted) > 0
+
+    # Test decryption
+    decrypted = device._decrypt(bytearray.fromhex(encrypted))
+    assert decrypted.hex() == "48656c6c6f"
+
+
+@pytest.mark.asyncio
+async def test_encrypt_with_none_iv() -> None:
+    """Test that encryption raises error when IV is None."""
+    device = create_encrypted_device()
+    device._iv = None
+
+    with pytest.raises(RuntimeError, match="Cannot encrypt: IV is None"):
+        device._encrypt("48656c6c6f")
+
+
+@pytest.mark.asyncio
+async def test_decrypt_with_none_iv() -> None:
+    """Test that decryption raises error when IV is None."""
+    device = create_encrypted_device()
+    device._iv = None
+
+    with pytest.raises(RuntimeError, match="Cannot decrypt: IV is None"):
+        device._decrypt(bytearray.fromhex("48656c6c6f"))
+
+
+@pytest.mark.asyncio
+async def test_get_cipher_with_none_iv() -> None:
+    """Test that _get_cipher raises error when IV is None."""
+    device = create_encrypted_device()
+    device._iv = None
+
+    with pytest.raises(RuntimeError, match="Cannot create cipher: IV is None"):
+        device._get_cipher()
+
+
+@pytest.mark.asyncio
+async def test_execute_disconnect_clears_encryption_state() -> None:
+    """Test that disconnect properly clears encryption state."""
+    device = create_encrypted_device()
+    device._iv = b"\x12\x34\x56\x78\x9a\xbc\xde\xf0\x12\x34\x56\x78\x9a\xbc\xde\xf0"
+    device._cipher = None  # type: ignore[assignment]
+
+    # Mock client
+    mock_client = AsyncMock()
+    device._client = mock_client
+
+    with patch.object(device, "_execute_disconnect_with_lock") as mock_disconnect:
+        await device._execute_disconnect()
+
+    assert device._iv is None
+    assert device._cipher is None
+    mock_disconnect.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_concurrent_commands_with_same_device() -> None:
+    """Test multiple concurrent commands on the same device."""
+    device = create_encrypted_device()
+
+    # Pre-initialize IV (16 bytes for AES CTR mode)
+    device._iv = b"\x12\x34\x56\x78\x9a\xbc\xde\xf0\x12\x34\x56\x78\x9a\xbc\xde\xf0"
+
+    with (
+        patch.object(device, "_send_command_locked_with_retry") as mock_send,
+        patch.object(device, "_encrypt") as mock_encrypt,
+        patch.object(device, "_decrypt") as mock_decrypt,
+    ):
+        mock_encrypt.return_value = "656e63727970746564"  # "encrypted" in hex
+        mock_decrypt.return_value = b"response"
+        mock_send.return_value = b"\x01\x00\x00\x00data"
+
+        # Send multiple commands concurrently
+        tasks = [
+            device._send_command("570200"),
+            device._send_command("570201"),
+            device._send_command("570202"),
+        ]
+
+        results = await asyncio.gather(*tasks)
+
+        # All commands should succeed
+        assert all(result == b"\x01response" for result in results)
+        assert mock_send.call_count == 3
+
+
+@pytest.mark.asyncio
+async def test_command_retry_with_encryption() -> None:
+    """Test command retry logic with encrypted commands."""
+    device = create_encrypted_device()
+    device._retry_count = 2
+
+    # Pre-initialize IV (16 bytes for AES CTR mode)
+    device._iv = b"\x12\x34\x56\x78\x9a\xbc\xde\xf0\x12\x34\x56\x78\x9a\xbc\xde\xf0"
+
+    with (
+        patch.object(device, "_send_command_locked") as mock_send_locked,
+        patch.object(device, "_ensure_connected"),
+        patch.object(device, "_encrypt") as mock_encrypt,
+        patch.object(device, "_decrypt") as mock_decrypt,
+    ):
+        mock_encrypt.return_value = "656e63727970746564"  # "encrypted" in hex
+        mock_decrypt.return_value = b"response"
+
+        # First attempt fails, second succeeds
+        mock_send_locked.side_effect = [
+            BleakDBusError("org.bluez.Error", []),
+            b"\x01\x00\x00\x00data",
+        ]
+
+        result = await device._send_command("570200")
+
+        assert result == b"\x01response"
+        assert mock_send_locked.call_count == 2
+
+
+@pytest.mark.asyncio
+async def test_empty_data_encryption_decryption() -> None:
+    """Test encryption/decryption of empty data."""
+    device = create_encrypted_device()
+    device._iv = b"\x00" * 16
+
+    # Test empty encryption
+    encrypted = device._encrypt("")
+    assert encrypted == ""
+
+    # Test empty decryption
+    decrypted = device._decrypt(bytearray())
+    assert decrypted == b""

tests/test_strip_light.py

@@ -0,0 +1,302 @@
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from bleak.backends.device import BLEDevice
+
+from switchbot import SwitchBotAdvertisement, SwitchbotModel
+from switchbot.const.light import ColorMode
+from switchbot.devices import light_strip
+from switchbot.devices.base_light import SwitchbotBaseLight
+from switchbot.devices.device import SwitchbotEncryptedDevice, SwitchbotOperationError
+
+from .test_adv_parser import generate_ble_device
+
+
+def create_device_for_command_testing(init_data: dict | None = None):
+    ble_device = generate_ble_device("aa:bb:cc:dd:ee:ff", "any")
+    device = light_strip.SwitchbotStripLight3(
+        ble_device, "ff", "ffffffffffffffffffffffffffffffff"
+    )
+    device.update_from_advertisement(make_advertisement_data(ble_device, init_data))
+    device._send_command = AsyncMock()
+    device._check_command_result = MagicMock()
+    device.update = AsyncMock()
+    return device
+
+
+def make_advertisement_data(ble_device: BLEDevice, init_data: dict | None = None):
+    """Set advertisement data with defaults."""
+    if init_data is None:
+        init_data = {}
+
+    return SwitchBotAdvertisement(
+        address="aa:bb:cc:dd:ee:ff",
+        data={
+            "rawAdvData": b"\x00\x00\x00\x00\x10\xd0\xb1",
+            "data": {
+                "sequence_number": 133,
+                "isOn": True,
+                "brightness": 30,
+                "delay": False,
+                "network_state": 2,
+                "color_mode": 2,
+                "cw": 4753,
+            }
+            | init_data,
+            "isEncrypted": False,
+            "model": b"\x00\x10\xd0\xb1",
+            "modelFriendlyName": "Strip Light 3",
+            "modelName": SwitchbotModel.STRIP_LIGHT_3,
+        },
+        device=ble_device,
+        rssi=-80,
+        active=True,
+    )
+
+
+@pytest.mark.asyncio
+async def test_default_info():
+    """Test default initialization of the strip light."""
+    device = create_device_for_command_testing()
+
+    assert device.rgb is None
+
+    device._state = {"r": 30, "g": 0, "b": 0, "cw": 3200}
+
+    assert device.is_on() is True
+    assert device.on is True
+    assert device.color_mode == ColorMode.RGB
+    assert device.color_modes == {
+        ColorMode.RGB,
+        ColorMode.COLOR_TEMP,
+    }
+    assert device.rgb == (30, 0, 0)
+    assert device.color_temp == 3200
+    assert device.brightness == 30
+    assert device.min_temp == 2700
+    assert device.max_temp == 6500
+    assert device.get_effect_list == list(light_strip.EFFECT_DICT.keys())
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("basic_info", "version_info"), [(True, False), (False, True), (False, False)]
+)
+async def test_get_basic_info_returns_none(basic_info, version_info):
+    device = create_device_for_command_testing()
+
+    async def mock_get_basic_info(arg):
+        if arg == light_strip.STRIP_REQUEST:
+            return basic_info
+        if arg == light_strip.DEVICE_GET_VERSION_KEY:
+            return version_info
+        return None
+
+    device._get_basic_info = AsyncMock(side_effect=mock_get_basic_info)
+
+    assert await device.get_basic_info() is None
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("info_data", "result"),
+    [
+        (
+            {
+                "basic_info": b"\x01\x00<\xff\x00\xd8\x00\x19d\x00\x03",
+                "version_info": b"\x01\x01\n",
+            },
+            [False, 60, 255, 0, 216, 6500, 3, 1.0],
+        ),
+        (
+            {
+                "basic_info": b"\x01\x80NK\xff:\x00\x19d\xff\x02",
+                "version_info": b"\x01\x01\n",
+            },
+            [True, 78, 75, 255, 58, 6500, 2, 1.0],
+        ),
+        (
+            {
+                "basic_info": b"\x01\x80$K\xff:\x00\x13\xf9\xff\x06",
+                "version_info": b"\x01\x01\n",
+            },
+            [True, 36, 75, 255, 58, 5113, 6, 1.0],
+        ),
+    ],
+)
+async def test_strip_light_get_basic_info(info_data, result):
+    """Test getting basic info from the strip light."""
+    device = create_device_for_command_testing()
+
+    async def mock_get_basic_info(args: str) -> list[int] | None:
+        if args == light_strip.STRIP_REQUEST:
+            return info_data["basic_info"]
+        if args == light_strip.DEVICE_GET_VERSION_KEY:
+            return info_data["version_info"]
+        return None
+
+    device._get_basic_info = AsyncMock(side_effect=mock_get_basic_info)
+    info = await device.get_basic_info()
+
+    assert info["isOn"] is result[0]
+    assert info["brightness"] == result[1]
+    assert info["r"] == result[2]
+    assert info["g"] == result[3]
+    assert info["b"] == result[4]
+    assert info["cw"] == result[5]
+    assert info["color_mode"] == result[6]
+    assert info["firmware"] == result[7]
+
+
+@pytest.mark.asyncio
+async def test_set_color_temp():
+    """Test setting color temperature."""
+    device = create_device_for_command_testing()
+
+    await device.set_color_temp(50, 3000)
+
+    device._send_command.assert_called_with(f"{light_strip.COLOR_TEMP_KEY}320BB8")
+
+
+@pytest.mark.asyncio
+async def test_turn_on():
+    """Test turning on the strip light."""
+    device = create_device_for_command_testing({"isOn": True})
+
+    await device.turn_on()
+
+    device._send_command.assert_called_with(light_strip.STRIP_ON_KEY)
+
+    assert device.is_on() is True
+
+
+@pytest.mark.asyncio
+async def test_turn_off():
+    """Test turning off the strip light."""
+    device = create_device_for_command_testing({"isOn": False})
+
+    await device.turn_off()
+
+    device._send_command.assert_called_with(light_strip.STRIP_OFF_KEY)
+
+    assert device.is_on() is False
+
+
+@pytest.mark.asyncio
+async def test_set_brightness():
+    """Test setting brightness."""
+    device = create_device_for_command_testing()
+
+    await device.set_brightness(75)
+
+    device._send_command.assert_called_with(f"{light_strip.BRIGHTNESS_KEY}4B")
+
+
+@pytest.mark.asyncio
+async def test_set_rgb():
+    """Test setting RGB values."""
+    device = create_device_for_command_testing()
+
+    await device.set_rgb(100, 255, 128, 64)
+
+    device._send_command.assert_called_with(f"{light_strip.RGB_BRIGHTNESS_KEY}64FF8040")
+
+
+@pytest.mark.asyncio
+async def test_set_effect_with_invalid_effect():
+    """Test setting an invalid effect."""
+    device = create_device_for_command_testing()
+
+    with pytest.raises(
+        SwitchbotOperationError, match="Effect invalid_effect not supported"
+    ):
+        await device.set_effect("invalid_effect")
+
+
+@pytest.mark.asyncio
+async def test_set_effect_with_valid_effect():
+    """Test setting a valid effect."""
+    device = create_device_for_command_testing()
+    device._send_multiple_commands = AsyncMock()
+
+    await device.set_effect("Christmas")
+
+    device._send_multiple_commands.assert_called_with(
+        light_strip.EFFECT_DICT["Christmas"]
+    )
+
+    assert device.get_effect() == "Christmas"
+
+
+@pytest.mark.asyncio
+@patch.object(SwitchbotEncryptedDevice, "verify_encryption_key", new_callable=AsyncMock)
+async def test_verify_encryption_key(mock_parent_verify):
+    ble_device = generate_ble_device("aa:bb:cc:dd:ee:ff", "any")
+    key_id = "ff"
+    encryption_key = "ffffffffffffffffffffffffffffffff"
+
+    mock_parent_verify.return_value = True
+
+    result = await light_strip.SwitchbotStripLight3.verify_encryption_key(
+        device=ble_device,
+        key_id=key_id,
+        encryption_key=encryption_key,
+    )
+
+    mock_parent_verify.assert_awaited_once_with(
+        ble_device,
+        key_id,
+        encryption_key,
+        SwitchbotModel.STRIP_LIGHT_3,
+    )
+
+    assert result is True
+
+
+def create_strip_light_device(init_data: dict | None = None):
+    ble_device = generate_ble_device("aa:bb:cc:dd:ee:ff", "any")
+    return light_strip.SwitchbotLightStrip(ble_device)
+
+
+@pytest.mark.asyncio
+async def test_strip_light_supported_color_modes():
+    """Test that the strip light supports the expected color modes."""
+    device = create_strip_light_device()
+
+    assert device.color_modes == {
+        ColorMode.RGB,
+    }
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("commands", "results", "final_result"),
+    [
+        (("command1", "command2"), [(b"\x01", False), (None, False)], False),
+        (("command1", "command2"), [(None, False), (b"\x01", True)], True),
+        (("command1", "command2"), [(b"\x01", True), (b"\x01", False)], True),
+    ],
+)
+async def test_send_multiple_commands(commands, results, final_result):
+    """Test sending multiple commands."""
+    device = create_device_for_command_testing()
+
+    device._send_command = AsyncMock(side_effect=[r[0] for r in results])
+
+    device._check_command_result = MagicMock(side_effect=[r[1] for r in results])
+
+    result = await device._send_multiple_commands(list(commands))
+
+    assert result is final_result
+
+
+@pytest.mark.asyncio
+async def test_unimplemented_color_mode():
+    class TestDevice(SwitchbotBaseLight):
+        pass
+
+    ble_device = generate_ble_device("aa:bb:cc:dd:ee:ff", "any")
+    device = TestDevice(ble_device)
+
+    with pytest.raises(NotImplementedError):
+        _ = device.color_mode