Cleanup member access (#1698)

* Reduce number of required characters for an address identification

* Cleanup member access

changelog.md

@@ -1,6 +1,7 @@
 # Version 2024.9.5 (2024-09-03)
 
 - Improve device_description usage
+- Reduce number of required characters for an address identification
 
 # Version 2024.9.4 (2024-09-03)
 

hahomematic/caches/dynamic.py

@@ -432,7 +432,7 @@ def _fire_event(mismatch_count: int) -> None:
                             EVENT_INTERFACE_ID: self._interface_id,
                             EVENT_TYPE: event_type,
                             EVENT_DATA: {
-                                EVENT_INSTANCE_NAME: self._central.config.name,
+                                EVENT_INSTANCE_NAME: self._central.name,
                                 EVENT_PONG_MISMATCH_COUNT: mismatch_count,
                             },
                         }

hahomematic/client/__init__.py

@@ -68,29 +68,22 @@ class Client(ABC):
     def __init__(self, client_config: _ClientConfig) -> None:
         """Initialize the Client."""
         self._config: Final = client_config
-        self.central: Final[hmcu.CentralUnit] = client_config.central
         self._last_value_send_cache = CommandCache(interface_id=client_config.interface_id)
-
-        self._json_rpc_client: Final = client_config.central.json_rpc_client
-
-        self.interface: Final[str] = client_config.interface
-        self.interface_id: Final[str] = client_config.interface_id
-        self.version: Final[str] = client_config.version
+        self._json_rpc_client: Final = client_config.central.config.json_rpc_client
         self._available: bool = True
         self._connection_error_count: int = 0
         self._is_callback_alive: bool = True
-        self.modified_at: datetime = INIT_DATETIME
         self._ping_pong_cache: Final = PingPongCache(
             central=client_config.central, interface_id=client_config.interface_id
         )
-
         self._proxy: XmlRpcProxy
         self._proxy_read: XmlRpcProxy
-        self.system_information: SystemInformation
+        self._system_information: SystemInformation
+        self.modified_at: datetime = INIT_DATETIME
 
     async def init_client(self) -> None:
         """Init the client."""
-        self.system_information = await self._get_system_information()
+        self._system_information = await self._get_system_information()
         self._proxy = await self._config.get_xml_rpc_proxy(
             auth_enabled=self.system_information.auth_enabled
         )
@@ -103,6 +96,21 @@ def available(self) -> bool:
         """Return the availability of the client."""
         return self._available
 
+    @property
+    def central(self) -> hmcu.CentralUnit:
+        """Return the central of the client."""
+        return self._config.central
+
+    @property
+    def interface(self) -> str:
+        """Return the interface of the client."""
+        return self._config.interface
+
+    @property
+    def interface_id(self) -> str:
+        """Return the interface id of the client."""
+        return self._config.interface_id
+
     @property
     def last_value_send_cache(self) -> CommandCache:
         """Return the last value send cache."""
@@ -118,6 +126,16 @@ def ping_pong_cache(self) -> PingPongCache:
         """Return the ping pong cache."""
         return self._ping_pong_cache
 
+    @property
+    def system_information(self) -> SystemInformation:
+        """Return the system_information of the client."""
+        return self._system_information
+
+    @property
+    def version(self) -> str:
+        """Return the version id of the client."""
+        return self._config.version
+
     def get_product_group(self, device_type: str) -> ProductGroup:
         """Return the product group."""
         if self.interface == InterfaceName.HMIP_RF:
@@ -1113,20 +1131,14 @@ def __init__(
         self.interface_config: Final = interface_config
         self.interface: Final = interface_config.interface
         self.interface_id: Final = interface_config.interface_id
-        self._callback_host: Final[str] = (
-            central.config.callback_host
-            if central.config.callback_host
-            else central.xml_rpc_server_ip_addr
-        )
-        self._callback_port: Final[int] = (
-            central.config.callback_port
-            if central.config.callback_port
-            else central.xml_rpc_server_port
-        )
         self.has_credentials: Final[bool] = (
             central.config.username is not None and central.config.password is not None
         )
-        self.init_url: Final[str] = f"http://{self._callback_host}:{self._callback_port}"
+        self.init_url: Final[str] = f"http://{central.config.callback_host
+            if central.config.callback_host
+            else central.xml_rpc_server_ip_addr}:{central.config.callback_port
+            if central.config.callback_port
+            else central.xml_rpc_server_port}"
         self.xml_rpc_uri: Final = build_xml_rpc_uri(
             host=central.config.host,
             port=interface_config.port,

hahomematic/const.py

@@ -37,8 +37,8 @@
 # The CCU WebUI also supports ÄäÖöÜüß, but these characters are not supported by the XmlRPC servers
 CCU_PASSWORD_PATTERN: Final = re.compile(r"[A-Za-z0-9.!$():;#-]{0,}")
 # Pattern is bigger than needed
-CHANNEL_ADDRESS_PATTERN: Final = re.compile(r"^[0-9a-zA-Z]{10,20}:[0-9]{1,3}$")
-DEVICE_ADDRESS_PATTERN: Final = re.compile(r"^[0-9a-zA-Z]{10,20}$")
+CHANNEL_ADDRESS_PATTERN: Final = re.compile(r"^[0-9a-zA-Z-]{5,20}:[0-9]{1,3}$")
+DEVICE_ADDRESS_PATTERN: Final = re.compile(r"^[0-9a-zA-Z-]{5,20}$")
 ALLOWED_HOSTNAME_PATTERN: Final = re.compile(r"(?!-)[a-z0-9-]{1,63}(?<!-)$", re.IGNORECASE)
 HTMLTAG_PATTERN: Final = re.compile(r"<.*?>|&([a-z0-9]+|#[0-9]{1,6}|#x[0-9a-f]{1,6});")
 

hahomematic/platforms/device.py

@@ -71,7 +71,7 @@ def __init__(self, central: hmcu.CentralUnit, interface_id: str, device_address:
         self._sub_device_channels: Final[dict[int, int]] = {}
         self._central: Final = central
         self._interface_id: Final = interface_id
-        self._interface: Final = central.device_details.get_interface(device_address)
+        self._interface: Final = central.device_details.get_interface(address=device_address)
         self._client: Final = central.get_client(interface_id=interface_id)
         self._device_address: Final = device_address
         self._device_description = self._get_device_description(
@@ -114,7 +114,7 @@ def __init__(self, central: hmcu.CentralUnit, interface_id: str, device_address:
             device_address=device_address,
             device_type=self._device_type,
         )
-        self.value_cache: Final = ValueCache(device=self)
+        self._value_cache: Final[ValueCache] = ValueCache(device=self)
         self._rooms: Final = central.device_details.get_device_rooms(device_address=device_address)
         self._update_entity: Final = HmUpdate(device=self) if self.is_updatable else None  # pylint: disable=using-constant-test
         _LOGGER.debug(
@@ -223,6 +223,11 @@ def generic_entities(self) -> tuple[GenericEntity, ...]:
         """Return the generic entities."""
         return tuple(self._generic_entities.values())
 
+    @property
+    def has_custom_entity_definition(self) -> bool:
+        """Return if custom_entity definition is available for the device."""
+        return self._has_custom_entity_definition
+
     @config_property
     def has_sub_devices(self) -> bool:
         """Return if device has multiple sub device channels."""
@@ -248,10 +253,10 @@ def ignore_for_custom_entity(self) -> bool:
         """Return if device should be ignored for custom entity."""
         return self._ignore_for_custom_entity
 
-    @property
-    def has_custom_entity_definition(self) -> bool:
-        """Return if custom_entity definition is available for the device."""
-        return self._has_custom_entity_definition
+    @config_property
+    def is_updatable(self) -> bool:
+        """Return if the device is updatable."""
+        return self._is_updatable
 
     @config_property
     def manufacturer(self) -> str:
@@ -290,16 +295,16 @@ def sub_type(self) -> str | None:
         """Return the sub_type of the device."""
         return self._sub_type
 
-    @config_property
-    def is_updatable(self) -> bool:
-        """Return if the device is updatable."""
-        return self._is_updatable
-
     @property
     def update_entity(self) -> HmUpdate | None:
         """Return the device firmware update entity of the device."""
         return self._update_entity
 
+    @property
+    def value_cache(self) -> ValueCache:
+        """Return the value_cache of the device."""
+        return self._value_cache
+
     @property
     def _e_unreach(self) -> GenericEntity | None:
         """Return th UNREACH entity."""
@@ -570,9 +575,9 @@ async def refresh_data() -> None:
     async def load_value_cache(self) -> None:
         """Init the parameter cache."""
         if len(self._generic_entities) > 0:
-            await self.value_cache.init_base_entities()
+            await self._value_cache.init_base_entities()
         if len(self._generic_events) > 0:
-            await self.value_cache.init_readable_events()
+            await self._value_cache.init_readable_events()
         _LOGGER.debug(
             "INIT_DATA: Skipping load_data, missing entities for %s",
             self._device_address,

hahomematic/platforms/entity.py

@@ -293,9 +293,7 @@ def __init__(
         self._function: Final = self._central.device_details.get_function_text(
             address=self._channel_address
         )
-        self._client: Final[hmcl.Client] = device.central.get_client(
-            interface_id=device.interface_id
-        )
+        self._client: Final[hmcl.Client] = device.client
 
         self._forced_usage: EntityUsage | None = None
         self._entity_name_data: Final = self._get_entity_name()

hahomematic_support/client_local.py

@@ -45,7 +45,7 @@ def __init__(self, client_config: _ClientConfig, local_resources: LocalRessource
 
     async def init_client(self) -> None:
         """Init the client."""
-        self.system_information = await self._get_system_information()
+        self._system_information = await self._get_system_information()
 
     @property
     def available(self) -> bool:

tests/helper.py

@@ -6,7 +6,7 @@
 import importlib.resources
 import logging
 import os
-from typing import Any
+from typing import Any, Final
 from unittest.mock import MagicMock, Mock, patch
 
 from aiohttp import ClientSession
@@ -17,17 +17,19 @@
 from hahomematic.client import Client, InterfaceConfig, _ClientConfig
 from hahomematic.const import BackendSystemEvent, InterfaceName
 from hahomematic.platforms.custom.entity import CustomEntity
+from hahomematic.platforms.decorators import _get_public_attributes_by_decorator
 from hahomematic_support.client_local import ClientLocal, LocalRessources
 
 from tests import const
 
 _LOGGER = logging.getLogger(__name__)
 
+EXCLUDE_METHODS_FROM_MOCKS: Final = []
+INCLUDE_PROPERTIES_IN_MOCKS: Final = []
 GOT_DEVICES = False
 
-# pylint: disable=protected-access
-
 
+# pylint: disable=protected-access
 class Factory:
     """Factory for a central with one local client."""
 
@@ -113,7 +115,7 @@ async def get_default_central(
         """Return a central based on give address_device_translation."""
         central, client = await self.get_unpatched_default_central(
             address_device_translation=address_device_translation,
-            do_mock_client=do_mock_client,
+            do_mock_client=True,
             ignore_devices_on_create=ignore_devices_on_create,
             un_ignore_list=un_ignore_list,
         )
@@ -162,15 +164,34 @@ def load_device_description(central: CentralUnit, filename: str) -> Any:
     return dev_desc
 
 
-def get_mock(instance, **kwargs):
+def get_mock(instance: Any, **kwargs):
     """Create a mock and copy instance attributes over mock."""
     if isinstance(instance, Mock):
         instance.__dict__.update(instance._mock_wraps.__dict__)
         return instance
-
     mock = MagicMock(spec=instance, wraps=instance, **kwargs)
     mock.__dict__.update(instance.__dict__)
-    return mock
+    try:
+        for method_name in [
+            prop
+            for prop in _get_not_mockable_method_names(instance)
+            if prop not in INCLUDE_PROPERTIES_IN_MOCKS and prop not in kwargs
+        ]:
+            setattr(mock, method_name, getattr(instance, method_name))
+    except Exception:
+        pass
+    finally:
+        return mock
+
+
+def _get_not_mockable_method_names(instance: Any) -> set[str]:
+    """Return all relevant method names for mocking."""
+    methods: set[str] = set(_get_public_attributes_by_decorator(instance, property))
+
+    for method in dir(instance):
+        if method in EXCLUDE_METHODS_FROM_MOCKS:
+            methods.add(method)
+    return methods
 
 
 def _load_json_file(anchor: str, resource: str, filename: str) -> Any | None: