diff --git a/switchbot/__init__.py b/switchbot/__init__.py index b404a7f..e418188 100644 --- a/switchbot/__init__.py +++ b/switchbot/__init__.py @@ -36,6 +36,7 @@ SwitchbotDevice, SwitchbotEncryptedDevice, SwitchbotOperationError, + fetch_cloud_devices, ) from .devices.evaporative_humidifier import SwitchbotEvaporativeHumidifier from .devices.fan import SwitchbotFan @@ -104,6 +105,7 @@ "SwitchbotVacuum", "close_stale_connections", "close_stale_connections_by_address", + "fetch_cloud_devices", "get_device", "parse_advertisement_data", ] diff --git a/switchbot/adv_parser.py b/switchbot/adv_parser.py index 3c88faf..66ba755 100644 --- a/switchbot/adv_parser.py +++ b/switchbot/adv_parser.py @@ -45,6 +45,7 @@ from .adv_parsers.vacuum import process_vacuum, process_vacuum_k from .const import SwitchbotModel from .models import SwitchBotAdvertisement +from .utils import format_mac_upper _LOGGER = logging.getLogger(__name__) @@ -54,6 +55,8 @@ ) MFR_DATA_ORDER = (2409, 741, 89) +_MODEL_TO_MAC_CACHE: dict[str, SwitchbotModel] = {} + class SwitchbotSupportedType(TypedDict): """Supported type of Switchbot.""" @@ -383,6 +386,10 @@ def parse_advertisement_data( model: SwitchbotModel | None = None, ) -> SwitchBotAdvertisement | None: """Parse advertisement data.""" + upper_mac = format_mac_upper(device.address) + if model is None and upper_mac in _MODEL_TO_MAC_CACHE: + model = _MODEL_TO_MAC_CACHE[upper_mac] + service_data = advertisement_data.service_data _service_data = None @@ -470,3 +477,8 @@ def _parse_data( ) return data + + +def populate_model_to_mac_cache(mac: str, model: SwitchbotModel) -> None: + """Populate the model to MAC address cache.""" + _MODEL_TO_MAC_CACHE[mac] = model diff --git a/switchbot/devices/device.py b/switchbot/devices/device.py index 9113a21..cda6c73 100644 --- a/switchbot/devices/device.py +++ b/switchbot/devices/device.py @@ -24,6 +24,7 @@ ) from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes +from ..adv_parser import populate_model_to_mac_cache from ..api_config import SWITCHBOT_APP_API_BASE_URL, SWITCHBOT_APP_CLIENT_ID from ..const import ( DEFAULT_RETRY_COUNT, @@ -37,9 +38,43 @@ from ..discovery import GetSwitchbotDevices from ..helpers import create_background_task from ..models import SwitchBotAdvertisement +from ..utils import format_mac_upper _LOGGER = logging.getLogger(__name__) + +def _extract_region(userinfo: dict[str, Any]) -> str: + """Extract region from user info, defaulting to 'us'.""" + if "botRegion" in userinfo and userinfo["botRegion"] != "": + return userinfo["botRegion"] + return "us" + + +# Mapping from API model names to SwitchbotModel enum values +API_MODEL_TO_ENUM: dict[str, SwitchbotModel] = { + "WoHand": SwitchbotModel.BOT, + "WoCurtain": SwitchbotModel.CURTAIN, + "WoHumi": SwitchbotModel.HUMIDIFIER, + "WoPlug": SwitchbotModel.PLUG_MINI, + "WoPlugUS": SwitchbotModel.PLUG_MINI, + "WoContact": SwitchbotModel.CONTACT_SENSOR, + "WoStrip": SwitchbotModel.LIGHT_STRIP, + "WoSensorTH": SwitchbotModel.METER, + "WoMeter": SwitchbotModel.METER, + "WoMeterPlus": SwitchbotModel.METER_PRO, + "WoPresence": SwitchbotModel.MOTION_SENSOR, + "WoBulb": SwitchbotModel.COLOR_BULB, + "WoCeiling": SwitchbotModel.CEILING_LIGHT, + "WoLock": SwitchbotModel.LOCK, + "WoBlindTilt": SwitchbotModel.BLIND_TILT, + "WoIOSensor": SwitchbotModel.IO_METER, # Outdoor Meter + "WoButton": SwitchbotModel.REMOTE, # Remote button + "WoLinkMini": SwitchbotModel.HUBMINI_MATTER, # Hub Mini + "W1083002": SwitchbotModel.RELAY_SWITCH_1, # Relay Switch 1 + "W1079000": SwitchbotModel.METER_PRO, # Meter Pro (another variant) + "W1102001": SwitchbotModel.STRIP_LIGHT_3, # RGBWW Strip Light 3 +} + REQ_HEADER = "570f" @@ -164,6 +199,113 @@ def __init__( self._last_full_update: float = -PASSIVE_POLL_INTERVAL self._timed_disconnect_task: asyncio.Task[None] | None = None + @classmethod + async def _async_get_user_info( + cls, + session: aiohttp.ClientSession, + auth_headers: dict[str, str], + ) -> dict[str, Any]: + try: + return await cls.api_request( + session, "account", "account/api/v1/user/userinfo", {}, auth_headers + ) + except Exception as err: + raise SwitchbotAccountConnectionError( + f"Failed to retrieve SwitchBot Account user details: {err}" + ) from err + + @classmethod + async def _get_auth_result( + cls, + session: aiohttp.ClientSession, + username: str, + password: str, + ) -> dict[str, Any]: + """Authenticate with SwitchBot API.""" + try: + return await cls.api_request( + session, + "account", + "account/api/v1/user/login", + { + "clientId": SWITCHBOT_APP_CLIENT_ID, + "username": username, + "password": password, + "grantType": "password", + "verifyCode": "", + }, + ) + except Exception as err: + raise SwitchbotAuthenticationError(f"Authentication failed: {err}") from err + + @classmethod + async def get_devices( + cls, + session: aiohttp.ClientSession, + username: str, + password: str, + ) -> dict[str, SwitchbotModel]: + """Get devices from SwitchBot API and return formatted MAC to model mapping.""" + try: + auth_result = await cls._get_auth_result(session, username, password) + auth_headers = {"authorization": auth_result["access_token"]} + except Exception as err: + raise SwitchbotAuthenticationError(f"Authentication failed: {err}") from err + + userinfo = await cls._async_get_user_info(session, auth_headers) + region = _extract_region(userinfo) + + try: + device_info = await cls.api_request( + session, + f"wonderlabs.{region}", + "wonder/device/v3/getdevice", + { + "required_type": "All", + }, + auth_headers, + ) + except Exception as err: + raise SwitchbotAccountConnectionError( + f"Failed to retrieve devices from SwitchBot Account: {err}" + ) from err + + items: list[dict[str, Any]] = device_info["Items"] + mac_to_model: dict[str, SwitchbotModel] = {} + + for item in items: + if "device_mac" not in item: + continue + + if ( + "device_detail" not in item + or "device_type" not in item["device_detail"] + ): + continue + + mac = item["device_mac"] + model_name = item["device_detail"]["device_type"] + + # Format MAC to uppercase with colons + formatted_mac = format_mac_upper(mac) + + # Map API model name to SwitchbotModel enum if possible + if model_name in API_MODEL_TO_ENUM: + model = API_MODEL_TO_ENUM[model_name] + mac_to_model[formatted_mac] = model + # Populate the cache + populate_model_to_mac_cache(formatted_mac, model) + else: + # Log the full item payload for unknown models + _LOGGER.debug( + "Unknown model %s for device %s, full item: %s", + model_name, + formatted_mac, + item, + ) + + return mac_to_model + @classmethod async def api_request( cls, @@ -809,34 +951,13 @@ async def async_retrieve_encryption_key( device_mac = device_mac.replace(":", "").replace("-", "").upper() try: - auth_result = await cls.api_request( - session, - "account", - "account/api/v1/user/login", - { - "clientId": SWITCHBOT_APP_CLIENT_ID, - "username": username, - "password": password, - "grantType": "password", - "verifyCode": "", - }, - ) + auth_result = await cls._get_auth_result(session, username, password) auth_headers = {"authorization": auth_result["access_token"]} except Exception as err: raise SwitchbotAuthenticationError(f"Authentication failed: {err}") from err - try: - userinfo = await cls.api_request( - session, "account", "account/api/v1/user/userinfo", {}, auth_headers - ) - if "botRegion" in userinfo and userinfo["botRegion"] != "": - region = userinfo["botRegion"] - else: - region = "us" - except Exception as err: - raise SwitchbotAccountConnectionError( - f"Failed to retrieve SwitchBot Account user details: {err}" - ) from err + userinfo = await cls._async_get_user_info(session, auth_headers) + region = _extract_region(userinfo) try: device_info = await cls.api_request( @@ -1023,3 +1144,13 @@ def update_from_advertisement(self, advertisement: SwitchBotAdvertisement) -> No ) if current_state != new_state: create_background_task(self.update()) + + +async def fetch_cloud_devices( + session: aiohttp.ClientSession, + username: str, + password: str, +) -> dict[str, SwitchbotModel]: + """Fetch devices from SwitchBot API and return MAC to model mapping.""" + # Get devices from the API (which also populates the cache) + return await SwitchbotBaseDevice.get_devices(session, username, password) diff --git a/switchbot/utils.py b/switchbot/utils.py new file mode 100644 index 0000000..235de77 --- /dev/null +++ b/switchbot/utils.py @@ -0,0 +1,24 @@ +"""Utility functions for switchbot.""" + +from functools import lru_cache + + +@lru_cache(maxsize=512) +def format_mac_upper(mac: str) -> str: + """Format the mac address string to uppercase with colons.""" + to_test = mac + + if len(to_test) == 17 and to_test.count(":") == 5: + return to_test.upper() + + if len(to_test) == 17 and to_test.count("-") == 5: + to_test = to_test.replace("-", "") + elif len(to_test) == 14 and to_test.count(".") == 2: + to_test = to_test.replace(".", "") + + if len(to_test) == 12: + # no : included + return ":".join(to_test.upper()[i : i + 2] for i in range(0, 12, 2)) + + # Not sure how formatted, return original + return mac.upper() diff --git a/tests/test_adv_parser.py b/tests/test_adv_parser.py index 43abdf2..319e144 100644 --- a/tests/test_adv_parser.py +++ b/tests/test_adv_parser.py @@ -8,7 +8,11 @@ from bleak.backends.scanner import AdvertisementData from switchbot import HumidifierMode, SwitchbotModel -from switchbot.adv_parser import parse_advertisement_data +from switchbot.adv_parser import ( + _MODEL_TO_MAC_CACHE, + parse_advertisement_data, + populate_model_to_mac_cache, +) from switchbot.const.lock import LockStatus from switchbot.models import SwitchBotAdvertisement @@ -3790,3 +3794,73 @@ def test_adv_with_empty_data(test_case: AdvTestCase) -> None: rssi=-97, active=True, ) + + +def test_parse_advertisement_with_mac_cache() -> None: + """Test that populating the MAC cache helps identify unknown passive devices.""" + # Clear the cache to ensure clean test + _MODEL_TO_MAC_CACHE.clear() + + # Create a passive lock device with manufacturer data only (no service data) + # This would normally not be identifiable without the cache + mac_address = "C1:64:B8:7D:06:05" + ble_device = generate_ble_device(mac_address, "WoLock") + + # Lock passive advertisement with manufacturer data + # This is real data from a WoLock device in passive mode + adv_data = generate_advertisement_data( + manufacturer_data={2409: b"\xc1d\xb8}\x06\x05\x00\x00\x00\x00\x00"}, + service_data={}, + rssi=-70, + ) + + # First attempt: Without cache, parser cannot identify the device model + result_without_cache = parse_advertisement_data(ble_device, adv_data) + assert result_without_cache is None, "Should not decode without model hint" + + # Now populate the cache with the device's MAC and model + populate_model_to_mac_cache(mac_address, SwitchbotModel.LOCK) + + # Second attempt: With cache, parser can now identify and decode the device + result_with_cache = parse_advertisement_data(ble_device, adv_data) + assert result_with_cache is not None, "Should decode with MAC cache" + assert result_with_cache.data["modelName"] == SwitchbotModel.LOCK + assert result_with_cache.data["modelFriendlyName"] == "Lock" + assert result_with_cache.active is False # Passive advertisement + + # Clean up + _MODEL_TO_MAC_CACHE.clear() + + +def test_parse_advertisement_with_mac_cache_curtain() -> None: + """Test MAC cache with a passive curtain device.""" + # Clear the cache + _MODEL_TO_MAC_CACHE.clear() + + # Create a passive curtain device + mac_address = "CC:F4:C4:F9:AC:6C" + ble_device = generate_ble_device(mac_address, None) + + # Curtain passive advertisement with only manufacturer data + adv_data = generate_advertisement_data( + manufacturer_data={2409: b"\xccOLG\x00c\x00\x00\x11\x00\x00"}, + service_data={}, + rssi=-85, + ) + + # Without cache, cannot identify + result_without_cache = parse_advertisement_data(ble_device, adv_data) + assert result_without_cache is None + + # Populate cache + populate_model_to_mac_cache(mac_address, SwitchbotModel.CURTAIN) + + # With cache, can identify and parse + result_with_cache = parse_advertisement_data(ble_device, adv_data) + assert result_with_cache is not None + assert result_with_cache.data["modelName"] == SwitchbotModel.CURTAIN + assert result_with_cache.data["modelFriendlyName"] == "Curtain" + assert result_with_cache.active is False + + # Clean up + _MODEL_TO_MAC_CACHE.clear() diff --git a/tests/test_device.py b/tests/test_device.py new file mode 100644 index 0000000..410d66c --- /dev/null +++ b/tests/test_device.py @@ -0,0 +1,379 @@ +"""Tests for device.py functionality.""" + +from __future__ import annotations + +import logging +from typing import Any +from unittest.mock import MagicMock, patch + +import aiohttp +import pytest + +from switchbot import fetch_cloud_devices +from switchbot.adv_parser import _MODEL_TO_MAC_CACHE, populate_model_to_mac_cache +from switchbot.const import ( + SwitchbotAccountConnectionError, + SwitchbotAuthenticationError, + SwitchbotModel, +) +from switchbot.devices.device import SwitchbotBaseDevice, _extract_region + + +@pytest.fixture +def mock_auth_response() -> dict[str, Any]: + """Mock authentication response.""" + return { + "access_token": "test_token_123", + "refresh_token": "refresh_token_123", + "expires_in": 3600, + } + + +@pytest.fixture +def mock_user_info() -> dict[str, Any]: + """Mock user info response.""" + return { + "botRegion": "us", + "country": "us", + "email": "test@example.com", + } + + +@pytest.fixture +def mock_device_response() -> dict[str, Any]: + """Mock device list response.""" + return { + "Items": [ + { + "device_mac": "aabbccddeeff", + "device_name": "Test Bot", + "device_detail": { + "device_type": "WoHand", + "version": "1.0.0", + }, + }, + { + "device_mac": "112233445566", + "device_name": "Test Curtain", + "device_detail": { + "device_type": "WoCurtain", + "version": "2.0.0", + }, + }, + { + "device_mac": "778899aabbcc", + "device_name": "Test Lock", + "device_detail": { + "device_type": "WoLock", + "version": "1.5.0", + }, + }, + { + "device_mac": "ddeeff001122", + "device_name": "Unknown Device", + "device_detail": { + "device_type": "WoUnknown", + "version": "1.0.0", + "extra_field": "extra_value", + }, + }, + { + "device_mac": "invalid_device", + # Missing device_detail + }, + { + "device_mac": "another_invalid", + "device_detail": { + # Missing device_type + "version": "1.0.0", + }, + }, + ] + } + + +@pytest.mark.asyncio +async def test_get_devices( + mock_auth_response: dict[str, Any], + mock_user_info: dict[str, Any], + mock_device_response: dict[str, Any], + caplog: pytest.LogCaptureFixture, +) -> None: + """Test get_devices method.""" + caplog.set_level(logging.DEBUG) + + with ( + patch.object( + SwitchbotBaseDevice, "_get_auth_result", return_value=mock_auth_response + ), + patch.object( + SwitchbotBaseDevice, "_async_get_user_info", return_value=mock_user_info + ), + patch.object( + SwitchbotBaseDevice, "api_request", return_value=mock_device_response + ) as mock_api_request, + patch( + "switchbot.devices.device.populate_model_to_mac_cache" + ) as mock_populate_cache, + ): + session = MagicMock(spec=aiohttp.ClientSession) + result = await SwitchbotBaseDevice.get_devices( + session, "test@example.com", "password123" + ) + + # Check that api_request was called with correct parameters + mock_api_request.assert_called_once_with( + session, + "wonderlabs.us", + "wonder/device/v3/getdevice", + {"required_type": "All"}, + {"authorization": "test_token_123"}, + ) + + # Check returned dictionary + assert len(result) == 3 # Only valid devices with known models + assert result["AA:BB:CC:DD:EE:FF"] == SwitchbotModel.BOT + assert result["11:22:33:44:55:66"] == SwitchbotModel.CURTAIN + assert result["77:88:99:AA:BB:CC"] == SwitchbotModel.LOCK + + # Check that cache was populated + assert mock_populate_cache.call_count == 3 + mock_populate_cache.assert_any_call("AA:BB:CC:DD:EE:FF", SwitchbotModel.BOT) + mock_populate_cache.assert_any_call("11:22:33:44:55:66", SwitchbotModel.CURTAIN) + mock_populate_cache.assert_any_call("77:88:99:AA:BB:CC", SwitchbotModel.LOCK) + + # Check that unknown model was logged + assert "Unknown model WoUnknown for device DD:EE:FF:00:11:22" in caplog.text + assert "extra_field" in caplog.text # Full item should be logged + + +@pytest.mark.asyncio +async def test_get_devices_with_region( + mock_auth_response: dict[str, Any], + mock_device_response: dict[str, Any], + caplog: pytest.LogCaptureFixture, +) -> None: + """Test get_devices with different region.""" + mock_user_info_eu = { + "botRegion": "eu", + "country": "de", + "email": "test@example.com", + } + + with ( + patch.object( + SwitchbotBaseDevice, "_get_auth_result", return_value=mock_auth_response + ), + patch.object( + SwitchbotBaseDevice, "_async_get_user_info", return_value=mock_user_info_eu + ), + patch.object( + SwitchbotBaseDevice, "api_request", return_value=mock_device_response + ) as mock_api_request, + patch("switchbot.devices.device.populate_model_to_mac_cache"), + ): + session = MagicMock(spec=aiohttp.ClientSession) + await SwitchbotBaseDevice.get_devices( + session, "test@example.com", "password123" + ) + + # Check that EU region was used + mock_api_request.assert_called_once_with( + session, + "wonderlabs.eu", + "wonder/device/v3/getdevice", + {"required_type": "All"}, + {"authorization": "test_token_123"}, + ) + + +@pytest.mark.asyncio +async def test_get_devices_no_region( + mock_auth_response: dict[str, Any], + mock_device_response: dict[str, Any], +) -> None: + """Test get_devices with no region specified (defaults to us).""" + mock_user_info_no_region = { + "email": "test@example.com", + } + + with ( + patch.object( + SwitchbotBaseDevice, "_get_auth_result", return_value=mock_auth_response + ), + patch.object( + SwitchbotBaseDevice, + "_async_get_user_info", + return_value=mock_user_info_no_region, + ), + patch.object( + SwitchbotBaseDevice, "api_request", return_value=mock_device_response + ) as mock_api_request, + patch("switchbot.devices.device.populate_model_to_mac_cache"), + ): + session = MagicMock(spec=aiohttp.ClientSession) + await SwitchbotBaseDevice.get_devices( + session, "test@example.com", "password123" + ) + + # Check that default US region was used + mock_api_request.assert_called_once_with( + session, + "wonderlabs.us", + "wonder/device/v3/getdevice", + {"required_type": "All"}, + {"authorization": "test_token_123"}, + ) + + +@pytest.mark.asyncio +async def test_get_devices_empty_region( + mock_auth_response: dict[str, Any], + mock_device_response: dict[str, Any], +) -> None: + """Test get_devices with empty region string (defaults to us).""" + mock_user_info_empty_region = { + "botRegion": "", + "email": "test@example.com", + } + + with ( + patch.object( + SwitchbotBaseDevice, "_get_auth_result", return_value=mock_auth_response + ), + patch.object( + SwitchbotBaseDevice, + "_async_get_user_info", + return_value=mock_user_info_empty_region, + ), + patch.object( + SwitchbotBaseDevice, "api_request", return_value=mock_device_response + ) as mock_api_request, + patch("switchbot.devices.device.populate_model_to_mac_cache"), + ): + session = MagicMock(spec=aiohttp.ClientSession) + await SwitchbotBaseDevice.get_devices( + session, "test@example.com", "password123" + ) + + # Check that default US region was used + mock_api_request.assert_called_once_with( + session, + "wonderlabs.us", + "wonder/device/v3/getdevice", + {"required_type": "All"}, + {"authorization": "test_token_123"}, + ) + + +@pytest.mark.asyncio +async def test_fetch_cloud_devices( + mock_auth_response: dict[str, Any], + mock_user_info: dict[str, Any], + mock_device_response: dict[str, Any], +) -> None: + """Test fetch_cloud_devices wrapper function.""" + with ( + patch.object( + SwitchbotBaseDevice, "_get_auth_result", return_value=mock_auth_response + ), + patch.object( + SwitchbotBaseDevice, "_async_get_user_info", return_value=mock_user_info + ), + patch.object( + SwitchbotBaseDevice, "api_request", return_value=mock_device_response + ), + patch( + "switchbot.devices.device.populate_model_to_mac_cache" + ) as mock_populate_cache, + ): + session = MagicMock(spec=aiohttp.ClientSession) + result = await fetch_cloud_devices(session, "test@example.com", "password123") + + # Check returned dictionary + assert len(result) == 3 + assert result["AA:BB:CC:DD:EE:FF"] == SwitchbotModel.BOT + assert result["11:22:33:44:55:66"] == SwitchbotModel.CURTAIN + assert result["77:88:99:AA:BB:CC"] == SwitchbotModel.LOCK + + # Check that cache was populated + assert mock_populate_cache.call_count == 3 + + +@pytest.mark.asyncio +async def test_get_devices_authentication_error() -> None: + """Test get_devices with authentication error.""" + with patch.object( + SwitchbotBaseDevice, + "_get_auth_result", + side_effect=Exception("Auth failed"), + ): + session = MagicMock(spec=aiohttp.ClientSession) + with pytest.raises(SwitchbotAuthenticationError) as exc_info: + await SwitchbotBaseDevice.get_devices( + session, "test@example.com", "wrong_password" + ) + assert "Authentication failed" in str(exc_info.value) + + +@pytest.mark.asyncio +async def test_get_devices_connection_error( + mock_auth_response: dict[str, Any], + mock_user_info: dict[str, Any], +) -> None: + """Test get_devices with connection error.""" + with ( + patch.object( + SwitchbotBaseDevice, "_get_auth_result", return_value=mock_auth_response + ), + patch.object( + SwitchbotBaseDevice, "_async_get_user_info", return_value=mock_user_info + ), + patch.object( + SwitchbotBaseDevice, + "api_request", + side_effect=Exception("Network error"), + ), + ): + session = MagicMock(spec=aiohttp.ClientSession) + with pytest.raises(SwitchbotAccountConnectionError) as exc_info: + await SwitchbotBaseDevice.get_devices( + session, "test@example.com", "password123" + ) + assert "Failed to retrieve devices" in str(exc_info.value) + + +@pytest.mark.asyncio +async def test_populate_model_to_mac_cache() -> None: + """Test the populate_model_to_mac_cache helper function.""" + # Clear the cache first + _MODEL_TO_MAC_CACHE.clear() + + # Populate cache with test data + populate_model_to_mac_cache("AA:BB:CC:DD:EE:FF", SwitchbotModel.BOT) + populate_model_to_mac_cache("11:22:33:44:55:66", SwitchbotModel.CURTAIN) + + # Check cache contents + assert _MODEL_TO_MAC_CACHE["AA:BB:CC:DD:EE:FF"] == SwitchbotModel.BOT + assert _MODEL_TO_MAC_CACHE["11:22:33:44:55:66"] == SwitchbotModel.CURTAIN + assert len(_MODEL_TO_MAC_CACHE) == 2 + + # Clear cache after test + _MODEL_TO_MAC_CACHE.clear() + + +def test_extract_region() -> None: + """Test the _extract_region helper function.""" + # Test with botRegion present and not empty + assert _extract_region({"botRegion": "eu", "country": "de"}) == "eu" + assert _extract_region({"botRegion": "us", "country": "us"}) == "us" + assert _extract_region({"botRegion": "jp", "country": "jp"}) == "jp" + + # Test with botRegion empty string + assert _extract_region({"botRegion": "", "country": "de"}) == "us" + + # Test with botRegion missing + assert _extract_region({"country": "de"}) == "us" + + # Test with empty dict + assert _extract_region({}) == "us" diff --git a/tests/test_utils.py b/tests/test_utils.py new file mode 100644 index 0000000..f9f06d0 --- /dev/null +++ b/tests/test_utils.py @@ -0,0 +1,41 @@ +"""Tests for utils.py functionality.""" + +from __future__ import annotations + +from switchbot.utils import format_mac_upper + + +def test_format_mac_upper() -> None: + """Test the format_mac_upper utility function.""" + # Test already formatted with colons (lowercase) + assert format_mac_upper("aa:bb:cc:dd:ee:ff") == "AA:BB:CC:DD:EE:FF" + + # Test already formatted with colons (uppercase) + assert format_mac_upper("AA:BB:CC:DD:EE:FF") == "AA:BB:CC:DD:EE:FF" + + # Test with dashes + assert format_mac_upper("aa-bb-cc-dd-ee-ff") == "AA:BB:CC:DD:EE:FF" + assert format_mac_upper("AA-BB-CC-DD-EE-FF") == "AA:BB:CC:DD:EE:FF" + + # Test with dots (Cisco format) + assert format_mac_upper("aabb.ccdd.eeff") == "AA:BB:CC:DD:EE:FF" + assert format_mac_upper("AABB.CCDD.EEFF") == "AA:BB:CC:DD:EE:FF" + + # Test without separators + assert format_mac_upper("aabbccddeeff") == "AA:BB:CC:DD:EE:FF" + assert format_mac_upper("AABBCCDDEEFF") == "AA:BB:CC:DD:EE:FF" + + # Test mixed case without separators + assert format_mac_upper("AaBbCcDdEeFf") == "AA:BB:CC:DD:EE:FF" + + # Test invalid formats (should return original in uppercase) + assert format_mac_upper("invalid") == "INVALID" + assert format_mac_upper("aa:bb:cc") == "AA:BB:CC" # Too short + assert ( + format_mac_upper("aa:bb:cc:dd:ee:ff:gg") == "AA:BB:CC:DD:EE:FF:GG" + ) # Too long + + # Test edge cases + assert format_mac_upper("") == "" + assert format_mac_upper("123456789ABC") == "12:34:56:78:9A:BC" + assert format_mac_upper("12:34:56:78:9a:bc") == "12:34:56:78:9A:BC"