Merge e7eb0b8 into 7352bcf

setup.cfg

@@ -42,10 +42,18 @@ warn_redundant_casts = true
 warn_unused_configs = true
 
 # add the modules below once we add typing for them so that we fail the build in the future if someone changes something without updating the typings
-[mypy-xknx.core.*,xknx.devices.travelcalculator,xknx.exceptions.*,xknx.io.gateway_scanner,xknx.remote_value.remote_value,xknx.telegram.*,]
+# fully typechecked modules
+[mypy-xknx.xknx,xknx.core.*,xknx.exceptions.*,xknx.io.*,xknx.knxip.*,xknx.telegram.*,]
+strict = true
+ignore_errors = false
+warn_unreachable = true
+# TODO: turn these off, address issues
+implicit_reexport = true
+
+# partly typechecked modules (extra block for better overview)
+[mypy-xknx.devices.travelcalculator,xknx.remote_value.remote_value]
 strict = true
 ignore_errors = false
 warn_unreachable = true
 # TODO: turn these off, address issues
-allow_any_generics = true
 implicit_reexport = true

test/core_tests/telegram_queue_test.py

@@ -5,7 +5,7 @@
 
 from xknx import XKNX
 from xknx.dpt import DPTBinary
-from xknx.exceptions import CouldNotParseTelegram
+from xknx.exceptions import CommunicationError, CouldNotParseTelegram
 from xknx.telegram import AddressFilter, GroupAddress, Telegram, TelegramDirection
 from xknx.telegram.apci import GroupValueWrite
 
@@ -201,10 +201,8 @@ def test_process_to_callback(self, devices_process):
         devices_process.assert_called_once_with(telegram)
 
     @patch("xknx.io.KNXIPInterface")
-    @patch("logging.Logger.warning")
-    def test_outgoing(self, logger_warning_mock, if_mock):
+    def test_outgoing(self, if_mock):
         """Test outgoing telegrams in telegram queue."""
-        # pylint: disable=no-self-use
         xknx = XKNX()
 
         async_if_send_telegram = asyncio.Future()
@@ -218,10 +216,10 @@ def test_outgoing(self, logger_warning_mock, if_mock):
         )
 
         # log a warning if there is no KNXIP interface instanciated
-        self.loop.run_until_complete(
-            xknx.telegram_queue.process_telegram_outgoing(telegram)
-        )
-        logger_warning_mock.assert_called_once_with("No KNXIP interface defined")
+        with self.assertRaises(CommunicationError):
+            self.loop.run_until_complete(
+                xknx.telegram_queue.process_telegram_outgoing(telegram)
+            )
         if_mock.send_telegram.assert_not_called()
 
         # if we have an interface send the telegram

test/io_tests/gateway_scanner_test.py

@@ -186,7 +186,7 @@ async def async_none():
 
 def fake_router_search_response(xknx: XKNX) -> KNXIPFrame:
     """Return the KNXIPFrame of a KNX/IP Router with a SearchResponse body."""
-    _frame_header = KNXIPHeader(xknx)
+    _frame_header = KNXIPHeader()
     _frame_header.service_type_ident = KNXIPServiceType.SEARCH_RESPONSE
     _frame_body = SearchResponse(xknx)
     _frame_body.control_endpoint = HPAI(ip_addr="192.168.42.10", port=3671)

test/knxip_tests/body_test.py

@@ -1,52 +1,27 @@
-"""Unit test for KNX/IP ConnectionStateRequests."""
-import asyncio
+"""Unit test for KNX/IP Body base class."""
 import unittest
 from unittest.mock import patch
 
 from xknx import XKNX
-from xknx.knxip import KNXIPBody
+from xknx.knxip import KNXIPBody, KNXIPBodyResponse
 
 
 class Test_KNXIPBody(unittest.TestCase):
-    """Test class for KNX/IP ConnectionStateRequests."""
-
-    # pylint: disable=too-many-public-methods,invalid-name
+    """Test base class for KNX/IP bodys."""
 
     def setUp(self):
         """Set up test class."""
-        self.loop = asyncio.new_event_loop()
-        asyncio.set_event_loop(self.loop)
-
-    def tearDown(self):
-        """Tear down test class."""
-        self.loop.close()
-
-    def test_warn_calculated_length(self):
-        """Test correct warn message if calculated_length is missing."""
-        xknx = XKNX()
-        body = KNXIPBody(xknx)
-        with patch("logging.Logger.warning") as mock_warn:
-            body.calculated_length()
-            mock_warn.assert_called_with(
-                "'calculated_length()' not implemented for %s", "KNXIPBody"
-            )
-
-    def test_warn_to_knx(self):
-        """Test correct warn message if to_knx is missing."""
-        xknx = XKNX()
-        body = KNXIPBody(xknx)
-        with patch("logging.Logger.warning") as mock_warn:
-            body.to_knx()
-            mock_warn.assert_called_with(
-                "'to_knx()' not implemented for %s", "KNXIPBody"
-            )
-
-    def test_warn_from_knx(self):
-        """Test correct warn message if from_knx is missing."""
-        xknx = XKNX()
-        body = KNXIPBody(xknx)
-        with patch("logging.Logger.warning") as mock_warn:
-            body.from_knx((0x75, 0x0B, 0x1C, 0x17, 0x07, 0x18, 0x00, 0x00))
-            mock_warn.assert_called_with(
-                "'from_knx()' not implemented for %s", "KNXIPBody"
-            )
+        self.xknx = XKNX()
+
+    @patch.multiple(KNXIPBody, __abstractmethods__=set())
+    def test_body_attributes(self):
+        """Test attributes of KNXIPBody base class."""
+        body = KNXIPBody(self.xknx)
+        self.assertTrue(hasattr(body, "service_type"))
+
+    @patch.multiple(KNXIPBodyResponse, __abstractmethods__=set())
+    def test_response_attributes(self):
+        """Test attributes of KNXIPBodyResponse base class."""
+        response = KNXIPBodyResponse(self.xknx)
+        self.assertTrue(hasattr(response, "service_type"))
+        self.assertTrue(hasattr(response, "status_code"))

test/knxip_tests/header_test.py

@@ -24,8 +24,7 @@ def tearDown(self):
     def test_from_knx(self):
         """Test parsing and streaming wrong Header (wrong length byte)."""
         raw = (0x06, 0x10, 0x04, 0x21, 0x00, 0x0A)
-        xknx = XKNX()
-        header = KNXIPHeader(xknx)
+        header = KNXIPHeader()
         self.assertEqual(header.from_knx(raw), 6)
         self.assertEqual(header.header_length, 6)
         self.assertEqual(header.protocol_version, 16)
@@ -37,47 +36,42 @@ def test_from_knx(self):
     def test_set_length(self):
         """Test setting length."""
         xknx = XKNX()
-        header = KNXIPHeader(xknx)
+        header = KNXIPHeader()
         header.set_length(DisconnectRequest(xknx))
         # 6 (header) + 2 + 8 (HPAI length)
         self.assertEqual(header.total_length, 16)
 
     def test_set_length_error(self):
         """Test setting length with wrong type."""
-        xknx = XKNX()
-        header = KNXIPHeader(xknx)
+        header = KNXIPHeader()
         with self.assertRaises(TypeError):
             header.set_length(2)
 
     def test_from_knx_wrong_header(self):
         """Test parsing and streaming wrong Header (wrong length)."""
         raw = (0x06, 0x10, 0x04, 0x21, 0x00)
-        xknx = XKNX()
-        header = KNXIPHeader(xknx)
+        header = KNXIPHeader()
         with self.assertRaises(CouldNotParseKNXIP):
             header.from_knx(raw)
 
     def test_from_knx_wrong_header2(self):
         """Test parsing and streaming wrong Header (wrong length byte)."""
         raw = (0x05, 0x10, 0x04, 0x21, 0x00, 0x0A)
-        xknx = XKNX()
-        header = KNXIPHeader(xknx)
+        header = KNXIPHeader()
         with self.assertRaises(CouldNotParseKNXIP):
             header.from_knx(raw)
 
     def test_from_knx_wrong_header3(self):
         """Test parsing and streaming wrong Header (wrong protocol version)."""
         raw = (0x06, 0x11, 0x04, 0x21, 0x00, 0x0A)
-        xknx = XKNX()
-        header = KNXIPHeader(xknx)
+        header = KNXIPHeader()
         with self.assertRaises(CouldNotParseKNXIP):
             header.from_knx(raw)
 
     def test_from_knx_wrong_header4(self):
         """Test parsing and streaming wrong Header (unsupported service type)."""
         # 0x0000 as service type
         raw = (0x06, 0x10, 0x00, 0x00, 0x00, 0x0A)
-        xknx = XKNX()
-        header = KNXIPHeader(xknx)
+        header = KNXIPHeader()
         with self.assertRaises(CouldNotParseKNXIP):
             header.from_knx(raw)

test/str_test.py

@@ -534,8 +534,7 @@ def test_hpai(self):
 
     def test_header(self):
         """Test string representation of KNX/IP-Header."""
-        xknx = XKNX()
-        header = KNXIPHeader(xknx)
+        header = KNXIPHeader()
         header.total_length = 42
         self.assertEqual(
             str(header),
@@ -633,8 +632,8 @@ def test_search_response(self):
         self.assertEqual(
             str(search_response),
             '<SearchResponse control_endpoint="<HPAI 192.168.42.1:33941 />" dibs="[\n'
-            '<DIB dtc="None" data="" />,\n'
-            '<DIB dtc="None" data="" />\n'
+            '<DIB dtc="0" data="" />,\n'
+            '<DIB dtc="0" data="" />\n'
             ']" />',
         )
 

xknx/config/__init__.py

@@ -17,3 +17,22 @@
     WeatherSchema,
     XKNXSchema,
 )
+
+__all__ = [
+    "Config",
+    "ConfigV1",
+    "BinarySensorSchema",
+    "ClimateSchema",
+    "ConnectionSchema",
+    "CoverSchema",
+    "DateTimeSchema",
+    "ExposeSchema",
+    "FanSchema",
+    "LightSchema",
+    "NotificationSchema",
+    "SceneSchema",
+    "SensorSchema",
+    "SwitchSchema",
+    "WeatherSchema",
+    "XKNXSchema",
+]

xknx/config/config.py

@@ -6,10 +6,14 @@
 """
 from enum import Enum
 import logging
+from typing import TYPE_CHECKING
 
 from .config_v1 import ConfigV1
 from .yaml_loader import load_yaml
 
+if TYPE_CHECKING:
+    from xknx.xknx import XKNX
+
 logger = logging.getLogger("xknx.log")
 
 
@@ -23,24 +27,24 @@ class Version(Enum):
 class Config:
     """Class for parsing xknx.yaml."""
 
-    def __init__(self, xknx):
+    def __init__(self, xknx: "XKNX") -> None:
         """Initialize Config class."""
         self.xknx = xknx
 
-    def read(self, file="xknx.yaml"):
+    def read(self, file: str = "xknx.yaml") -> None:
         """Read config."""
         logger.debug("Reading %s", file)
         doc = load_yaml(file)
         self.parse(doc)
 
     @staticmethod
-    def parse_version(doc):
+    def parse_version(doc) -> Version:
         """Parse the version of the xknx.yaml."""
         if "version" in doc:
             return Version(doc["version"])
         return Version.VERSION_1
 
-    def parse(self, doc):
+    def parse(self, doc) -> None:
         """Parse the config from the YAML."""
         version = Config.parse_version(doc)
         if version is Version.VERSION_1:

xknx/core/state_updater.py

@@ -157,7 +157,7 @@ def __init__(
         self.tracker_type = tracker_type
         self.update_interval = interval_min * 60
         self._read_state = read_state_awaitable
-        self._task: Optional[asyncio.Task] = None
+        self._task: Optional[asyncio.Task[None]] = None
 
     def start(self) -> None:
         """Start StateTracker - read state on call."""

xknx/core/telegram_queue.py

@@ -9,7 +9,7 @@
 """
 import asyncio
 import logging
-from typing import TYPE_CHECKING, Awaitable, Callable, List, Optional
+from typing import TYPE_CHECKING, Awaitable, Callable, List, Optional, Tuple
 
 from xknx.exceptions import CommunicationError, XKNXException
 from xknx.telegram import AddressFilter, GroupAddress, Telegram, TelegramDirection
@@ -57,7 +57,7 @@ def __init__(self, xknx: "XKNX"):
         self.xknx = xknx
         self.telegram_received_cbs: List[TelegramQueue.Callback] = []
         self.outgoing_queue: asyncio.Queue[Optional[Telegram]] = asyncio.Queue()
-        self._consumer_task: Optional[Awaitable] = None
+        self._consumer_task: Optional[Awaitable[Tuple[None, None]]] = None
 
     def register_telegram_received_cb(
         self,
@@ -137,6 +137,8 @@ async def _process_all_telegrams(self) -> None:
         while not self.xknx.telegrams.empty():
             try:
                 telegram = self.xknx.telegrams.get_nowait()
+                if telegram is None:
+                    return
                 if telegram.direction == TelegramDirection.INCOMING:
                     await self.process_telegram_incoming(telegram)
                 elif telegram.direction == TelegramDirection.OUTGOING:
@@ -154,7 +156,7 @@ async def process_telegram_outgoing(self, telegram: Telegram) -> None:
             if isinstance(telegram.payload, GroupValueWrite):
                 await self.xknx.devices.process(telegram)
         else:
-            logger.warning("No KNXIP interface defined")
+            raise CommunicationError("No KNXIP interface defined")
 
     async def process_telegram_incoming(self, telegram: Telegram) -> None:
         """Process incoming telegram."""

xknx/devices/__init__.py

@@ -17,3 +17,26 @@
 from .switch import Switch
 from .travelcalculator import TravelCalculator, TravelStatus
 from .weather import Weather
+
+__all__ = [
+    "Action",
+    "ActionBase",
+    "ActionCallback",
+    "BinarySensor",
+    "Climate",
+    "ClimateMode",
+    "Cover",
+    "DateTime",
+    "Device",
+    "Devices",
+    "ExposeSensor",
+    "Fan",
+    "Light",
+    "Notification",
+    "Scene",
+    "Sensor",
+    "Switch",
+    "TravelCalculator",
+    "TravelStatus",
+    "Weather",
+]

xknx/devices/devices.py

@@ -3,18 +3,24 @@
 
 More or less an array with devices. Adds some search functionality to find devices.
 """
+from typing import Awaitable, Callable
+
+from xknx.telegram import Telegram
+
 from .device import Device
 
 
 class Devices:
     """Class for handling a vector/array of devices."""
 
-    def __init__(self):
+    def __init__(self) -> None:
         """Initialize Devices class."""
         self.__devices = []
         self.device_updated_cbs = []
 
-    def register_device_updated_cb(self, device_updated_cb):
+    def register_device_updated_cb(
+        self, device_updated_cb: Callable[[Device], Awaitable[None]]
+    ) -> None:
         """Register callback for devices beeing updated."""
         self.device_updated_cbs.append(device_updated_cb)
 
@@ -64,7 +70,7 @@ async def device_updated(self, device):
         for device_updated_cb in self.device_updated_cbs:
             await device_updated_cb(device)
 
-    async def process(self, telegram):
+    async def process(self, telegram: Telegram) -> None:
         """Process telegram."""
         for device in self.devices_by_group_address(telegram.destination_address):
             await device.process(telegram)