Add coverage config (#76)

* Add coverage config

- small fixes
- refactor tests

* Update changelog.txt

.coveragerc

@@ -0,0 +1,23 @@
+[run]
+source = hahomematic
+
+omit =
+    hahomematic/__init__.py
+
+
+[report]
+# Regexes for lines to exclude from consideration
+exclude_lines =
+    # Have to re-enable the standard pragma
+    pragma: no cover
+
+    # Don't complain about missing debug-only code:
+    def __repr__
+
+    # Don't complain if tests don't hit defensive assertion code:
+    raise AssertionError
+    raise NotImplementedError
+
+    # TYPE_CHECKING and @overload blocks are never executed during pytest run
+    if TYPE_CHECKING:
+    @overload

changelog.txt

@@ -1,8 +1,8 @@
 Version 0.1.0  (2021-12-xx)
 - Bump version to 0.1.0
-- Remove interface_id from get_entity_name
-- Remove interface_id from get_custom_entity_name
+- Remove interface_id from get_entity_name and get_custom_entity_name
 - Add initial test
+- Add coverage config
 
 Version 0.0.22 (2021-12-16)
 - Resolve names without interface

cov_hahomematic.sh

@@ -0,0 +1 @@
+pytest --cov=hahomematic --cov-config=.coveragerc --cov-report html tests/

hahomematic/entity.py

@@ -421,7 +421,7 @@ def __init__(
         self._device_enum = device_enum
         self._device_desc = device_def
         self._entity_def = entity_def
-        self._channel_no = channel_no
+        self.channel_no = channel_no
         self.name = get_custom_entity_name(
             central=self._central,
             address=self.address,
@@ -437,7 +437,7 @@ def _init_entities(self) -> None:
         fields_rep = self._device_desc.get(hm_entity_definition.ED_FIELDS_REP, {})
         # Add repeating fields
         for (f_name, p_name) in fields_rep.items():
-            f_address = f"{self.address}:{self._channel_no}"
+            f_address = f"{self.address}:{self.channel_no}"
             entity = self._device.get_hm_entity(f_address, p_name)
             self._add_entity(f_name, entity)
         # Add device fields

tests/conftest.py

@@ -1,20 +1,25 @@
+from __future__ import annotations
+
 import asyncio
 import logging
+from typing import Any
 
 import pydevccu
 import pytest
 
 from hahomematic import config, const
-from hahomematic.central_unit import CentralConfig
+from hahomematic.central_unit import CentralConfig, CentralUnit
 from hahomematic.client import ClientConfig
+from hahomematic.device import HmDevice
+from hahomematic.entity import CustomEntity, GenericEntity
 from hahomematic.helpers import get_device_address
 from hahomematic.xml_rpc_server import register_xml_rpc_server
 
 logging.basicConfig(level=logging.DEBUG)
 CCU_HOST = "127.0.0.1"
 CCU_USERNAME = None
 CCU_PASSWORD = None
-got_devices = False
+GOT_DEVICES = False
 
 # content of conftest.py
 def pytest_configure(config):
@@ -29,26 +34,38 @@ def pytest_unconfigure(config):  # pragma: no cover
     del sys._called_from_test
 
 
+@pytest.yield_fixture(scope="session")
+def loop() -> asyncio.AbstractEventLoop:
+    """Yield running event_loop"""
+    event_loop = asyncio.get_event_loop_policy().new_event_loop()
+    yield event_loop
+    event_loop.close()
+
+
 @pytest.fixture
-async def pydev_ccu(loop):
+def pydev_ccu() -> pydevccu.Server:
     """Defines the virtual ccu"""
-    ccu = pydevccu.Server(persistance=True, logic={"startupdelay": 1, "interval": 30})
+    ccu = pydevccu.Server()
     ccu.start()
     yield ccu
     ccu.stop()
 
 
 @pytest.fixture
-async def central(loop, pydev_ccu):
+async def central(
+    loop: asyncio.AbstractEventLoop, pydev_ccu: pydevccu.Server
+) -> CentralUnit:
     """Yield central"""
-    SLEEPCOUNTER = 0
+    sleep_counter = 0
+    global GOT_DEVICES
+    GOT_DEVICES = False
 
-    def systemcallback(self, src, *args):
-        if args and args[0] and len(args[0]) > 0:
-            global got_devices
-            got_devices = True
+    def systemcallback(src, *args):
+        if src == "devicesCreated" and args and args[0] and len(args[0]) > 0:
+            global GOT_DEVICES
+            GOT_DEVICES = True
 
-    central = CentralConfig(
+    central_unit = CentralConfig(
         name="ccu-dev",
         entry_id="123",
         loop=loop,
@@ -60,59 +77,74 @@ def systemcallback(self, src, *args):
     ).get_central()
     config.INIT_TIMEOUT = 10
     config.CACHE_DIR = "cache"
-    central.callback_system_event = systemcallback
+    central_unit.callback_system_event = systemcallback
     client1 = await ClientConfig(
-        central=central,
+        central=central_unit,
         name="hm",
         port=2001,
     ).get_client()
 
     # Clients have to exist prior to creating the devices
-    central.create_devices()
+    central_unit.create_devices()
     # Once the central_1 is running we subscribe to receive messages.
     await client1.proxy_init()
-    await central.init_hub()
-    while not got_devices and SLEEPCOUNTER < 300:
+    await central_unit.init_hub()
+    while not GOT_DEVICES and sleep_counter < 300:
         print("Waiting for devices")
-        SLEEPCOUNTER += 1
+        sleep_counter += 1
         await asyncio.sleep(1)
 
-    yield central
+    yield central_unit
 
-    await central.stop()
+    await central_unit.stop()
 
 
-@pytest.yield_fixture(scope="session")
-def loop(request):
-    """Yield running event_loop"""
-    event_loop = asyncio.get_event_loop_policy().new_event_loop()
-    yield event_loop
-    event_loop.close()
-
-
-async def get_value_from_central(central, address, parameter, do_load=False):
+async def get_value_from_generic_entity(
+    central_unit: CentralUnit, address: str, parameter: str, do_load: bool = False
+) -> Any:
     """Return the device value."""
-    hm_entity = get_hm_entity(central, address, parameter)
+    hm_entity = await get_hm_genertic_entity(central_unit, address, parameter)
     assert hm_entity
     if do_load:
         await hm_entity.load_data()
         assert hm_entity.state
     return hm_entity.state
 
 
-def get_hm_device(central, address):
+def get_hm_device(central_unit: CentralUnit, address: str) -> HmDevice | None:
     """Return the hm_device."""
     d_address = get_device_address(address)
-    return central.hm_devices.get(d_address)
+    return central_unit.hm_devices.get(d_address)
+
+
+async def get_hm_genertic_entity(
+    central_unit: CentralUnit, address: str, parameter: str, do_load: bool = False
+) -> GenericEntity | None:
+    """Return the hm generic_entity."""
+    hm_device = get_hm_device(central_unit, address)
+    assert hm_device
+    hm_entity = hm_device.entities.get((address, parameter))
+    if hm_entity and do_load:
+        await hm_entity.load_data()
+    return hm_entity
 
 
-def get_hm_entity(central, address, parameter):
-    """Return the hm_entity."""
-    hm_device = get_hm_device(central, address)
+async def get_hm_custom_entity(
+    central_unit: CentralUnit, address: str, channel_no: int, do_load: bool = False
+) -> CustomEntity | None:
+    """Return the hm custom_entity."""
+    hm_device = get_hm_device(central_unit, address)
     assert hm_device
-    return hm_device.entities.get((address, parameter))
+    for custom_entity in hm_device.custom_entities:
+        if custom_entity.channel_no == channel_no:
+            if do_load:
+                await custom_entity.load_data()
+            return custom_entity
+    return None
 
 
-def send_device_value_to_ccu(pydev_ccu, address, parameter, value):
+def send_device_value_to_ccu(
+    pydev_ccu: pydevccu.Server, address: str, parameter: str, value: Any
+) -> None:
     """Send the device value to ccu."""
     pydev_ccu.setValue(address, parameter, value, force=False)

tests/test_central.py

@@ -2,7 +2,7 @@
 from typing import Any
 from unittest.mock import patch
 
-from conftest import get_value_from_central, send_device_value_to_ccu
+from conftest import get_value_from_generic_entity, send_device_value_to_ccu
 import pytest
 
 from hahomematic.helpers import get_device_address
@@ -26,12 +26,12 @@ async def test_device_set_data(central, pydev_ccu, loop) -> None:
     """Test callback."""
     assert central
     assert pydev_ccu
-    old_value = await get_value_from_central(
+    old_value = await get_value_from_generic_entity(
         central, "VCU6354483:1", "SET_POINT_TEMPERATURE"
     )
     assert old_value is None
     send_device_value_to_ccu(pydev_ccu, "VCU6354483:1", "SET_POINT_TEMPERATURE", 19.0)
-    new_value = await get_value_from_central(
+    new_value = await get_value_from_generic_entity(
         central, "VCU6354483:1", "SET_POINT_TEMPERATURE"
     )
     assert new_value == 19.0