Add parallel updates in lock and lock unit tests for switchbot integration (#143391)

Co-authored-by: J. Nick Koston <nick@koston.org>

homeassistant/components/switchbot/lock.py

@@ -13,6 +13,8 @@ from .const import CONF_LOCK_NIGHTLATCH, DEFAULT_LOCK_NIGHTLATCH
 from .coordinator import SwitchbotConfigEntry, SwitchbotDataUpdateCoordinator
 from .entity import SwitchbotEntity
 
+PARALLEL_UPDATES = 0
+
 
 async def async_setup_entry(
     hass: HomeAssistant,

tests/components/switchbot/__init__.py

@@ -488,3 +488,45 @@ WOSTRIP_SERVICE_INFO = BluetoothServiceInfoBleak(
     connectable=True,
     tx_power=-127,
 )
+
+
+WOLOCKPRO_SERVICE_INFO = BluetoothServiceInfoBleak(
+    name="WoLockPro",
+    manufacturer_data={2409: b"\xf7a\x07H\xe6\xe8-\x80\x00d\x00\x08"},
+    service_data={"0000fd3d-0000-1000-8000-00805f9b34fb": b"$\x80d"},
+    service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
+    address="AA:BB:CC:DD:EE:FF",
+    rssi=-60,
+    source="local",
+    advertisement=generate_advertisement_data(
+        local_name="WoLockPro",
+        manufacturer_data={2409: b"\xf7a\x07H\xe6\xe8-\x80\x00d\x00\x08"},
+        service_data={"0000fd3d-0000-1000-8000-00805f9b34fb": b"$\x80d"},
+        service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
+    ),
+    device=generate_ble_device("AA:BB:CC:DD:EE:FF", "WoLockPro"),
+    time=0,
+    connectable=True,
+    tx_power=-127,
+)
+
+
+LOCK_SERVICE_INFO = BluetoothServiceInfoBleak(
+    name="WoLock",
+    manufacturer_data={2409: b"\xca\xbaP\xddv;\x03\x03\x00 "},
+    service_data={"0000fd3d-0000-1000-8000-00805f9b34fb": b"o\x80d"},
+    service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
+    address="AA:BB:CC:DD:EE:FF",
+    rssi=-60,
+    source="local",
+    advertisement=generate_advertisement_data(
+        local_name="WoLock",
+        manufacturer_data={2409: b"\xca\xbaP\xddv;\x03\x03\x00 "},
+        service_data={"0000fd3d-0000-1000-8000-00805f9b34fb": b"o\x80d"},
+        service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
+    ),
+    device=generate_ble_device("AA:BB:CC:DD:EE:FF", "WoLock"),
+    time=0,
+    connectable=True,
+    tx_power=-127,
+)

tests/components/switchbot/conftest.py

@@ -2,7 +2,11 @@
 
 import pytest
 
-from homeassistant.components.switchbot.const import DOMAIN
+from homeassistant.components.switchbot.const import (
+    CONF_ENCRYPTION_KEY,
+    CONF_KEY_ID,
+    DOMAIN,
+)
 from homeassistant.const import CONF_ADDRESS, CONF_NAME, CONF_SENSOR_TYPE
 
 from tests.common import MockConfigEntry
@@ -25,3 +29,19 @@ def mock_entry_factory():
         },
         unique_id="aabbccddeeff",
     )
+
+
+@pytest.fixture
+def mock_entry_encrypted_factory():
+    """Fixture to create a MockConfigEntry with an encryption key and a customizable sensor type."""
+    return lambda sensor_type="lock": MockConfigEntry(
+        domain=DOMAIN,
+        data={
+            CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
+            CONF_NAME: "test-name",
+            CONF_SENSOR_TYPE: sensor_type,
+            CONF_KEY_ID: "ff",
+            CONF_ENCRYPTION_KEY: "ffffffffffffffffffffffffffffffff",
+        },
+        unique_id="aabbccddeeff",
+    )

tests/components/switchbot/test_lock.py

@@ -0,0 +1,105 @@
+"""Test the switchbot locks."""
+
+from collections.abc import Callable
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from homeassistant.components.bluetooth import BluetoothServiceInfoBleak
+from homeassistant.components.lock import DOMAIN as LOCK_DOMAIN
+from homeassistant.const import (
+    ATTR_ENTITY_ID,
+    SERVICE_LOCK,
+    SERVICE_OPEN,
+    SERVICE_UNLOCK,
+)
+from homeassistant.core import HomeAssistant
+
+from . import LOCK_SERVICE_INFO, WOLOCKPRO_SERVICE_INFO
+
+from tests.common import MockConfigEntry
+from tests.components.bluetooth import inject_bluetooth_service_info
+
+
+@pytest.mark.parametrize(
+    ("sensor_type", "service_info"),
+    [("lock_pro", WOLOCKPRO_SERVICE_INFO), ("lock", LOCK_SERVICE_INFO)],
+)
+@pytest.mark.parametrize(
+    ("service", "mock_method"),
+    [(SERVICE_UNLOCK, "unlock"), (SERVICE_LOCK, "lock")],
+)
+async def test_lock_services(
+    hass: HomeAssistant,
+    mock_entry_encrypted_factory: Callable[[str], MockConfigEntry],
+    sensor_type: str,
+    service: str,
+    mock_method: str,
+    service_info: BluetoothServiceInfoBleak,
+) -> None:
+    """Test lock and unlock services on lock and lockpro devices."""
+    inject_bluetooth_service_info(hass, service_info)
+
+    entry = mock_entry_encrypted_factory(sensor_type=sensor_type)
+    entry.add_to_hass(hass)
+
+    with patch(
+        f"homeassistant.components.switchbot.lock.switchbot.SwitchbotLock.{mock_method}",
+    ) as mocked_instance:
+        assert await hass.config_entries.async_setup(entry.entry_id)
+        await hass.async_block_till_done()
+
+        entity_id = "lock.test_name"
+
+        await hass.services.async_call(
+            LOCK_DOMAIN,
+            service,
+            {ATTR_ENTITY_ID: entity_id},
+            blocking=True,
+        )
+
+        mocked_instance.assert_awaited_once()
+
+
+@pytest.mark.parametrize(
+    ("sensor_type", "service_info"),
+    [("lock_pro", WOLOCKPRO_SERVICE_INFO), ("lock", LOCK_SERVICE_INFO)],
+)
+@pytest.mark.parametrize(
+    ("service", "mock_method"),
+    [(SERVICE_UNLOCK, "unlock_without_unlatch"), (SERVICE_OPEN, "unlock")],
+)
+async def test_lock_services_with_night_latch_enabled(
+    hass: HomeAssistant,
+    mock_entry_encrypted_factory: Callable[[str], MockConfigEntry],
+    sensor_type: str,
+    service: str,
+    mock_method: str,
+    service_info: BluetoothServiceInfoBleak,
+) -> None:
+    """Test lock service when night latch enabled."""
+    inject_bluetooth_service_info(hass, service_info)
+
+    entry = mock_entry_encrypted_factory(sensor_type=sensor_type)
+    entry.add_to_hass(hass)
+
+    mocked_instance = AsyncMock(return_value=True)
+
+    with patch.multiple(
+        "homeassistant.components.switchbot.lock.switchbot.SwitchbotLock",
+        is_night_latch_enabled=MagicMock(return_value=True),
+        **{mock_method: mocked_instance},
+    ):
+        assert await hass.config_entries.async_setup(entry.entry_id)
+        await hass.async_block_till_done()
+
+        entity_id = "lock.test_name"
+
+        await hass.services.async_call(
+            LOCK_DOMAIN,
+            service,
+            {ATTR_ENTITY_ID: entity_id},
+            blocking=True,
+        )
+
+        mocked_instance.assert_awaited_once()