Repository: XKNX/xknx
Pull request #430 (2020-09-26 13:44:39)

Increase test coverage
- Module remote_value: Adds tests for uncovered code.
- Module knxip: Increase test coverage

coveralls: 
[![Coverage Status](https://coveralls.io/builds/33741081/badge)](https://coveralls.io/builds/33741081)

Coverage increased (+0.2%) to 92.945% when pulling **61733025213dd5b69601c8192a9195f69c439ab9 on marvin-w:feature/test_coverage** into **e1560a1c843ad7bddc873140139d9d285db26c54 on XKNX:master**.


Patch:
diff --git a/xknx/knxip/cemi_frame.py b/xknx/knxip/cemi_frame.py
index f950b178..aa8fbbb6 100644
--- a/xknx/knxip/cemi_frame.py
+++ b/xknx/knxip/cemi_frame.py
@@ -203,9 +203,9 @@ class CEMIFrame:
     def to_knx(self):
         """Serialize to KNX/IP raw data."""
         if not isinstance(self.src_addr, (GroupAddress, PhysicalAddress)):
-            raise ConversionError("src_add not set")
+            raise ConversionError("src_addr not set")
         if not isinstance(self.dst_addr, (GroupAddress, PhysicalAddress)):
-            raise ConversionError("dst_add not set")
+            raise ConversionError("dst_addr not set")
 
         data = []
 


Test patch:
diff --git a/test/knxip_tests/cemi_frame_test.py b/test/knxip_tests/cemi_frame_test.py
index d1c90722..0b1fa670 100644
--- a/test/knxip_tests/cemi_frame_test.py
+++ b/test/knxip_tests/cemi_frame_test.py
@@ -3,11 +3,11 @@
 from unittest.mock import MagicMock
 
 from pytest import fixture, raises
-from xknx.dpt import DPTBinary
-from xknx.exceptions import CouldNotParseKNXIP, UnsupportedCEMIMessage
+from xknx.dpt import DPTBinary, DPTComparator
+from xknx.exceptions import ConversionError, CouldNotParseKNXIP, UnsupportedCEMIMessage
 from xknx.knxip.cemi_frame import CEMIFrame
 from xknx.knxip.knxip_enum import APCICommand, CEMIMessageCode
-from xknx.telegram import PhysicalAddress
+from xknx.telegram import PhysicalAddress, Telegram
 
 
 def get_data(code, adil, flags, src, dst, mpdu_len, tpci_apci, payload):
@@ -59,6 +59,92 @@ def test_invalid_apdu_len(frame):
         frame.from_knx(get_data(0x29, 0, 0, 0, 0, 2, 0, []))
 
 
+def test_invalid_src_addr(frame):
+    """Test for invalid src addr"""
+    frame.code = CEMIMessageCode.L_DATA_IND
+    frame.cmd = APCICommand.GROUP_READ
+    frame.flags = 0
+    frame.mpdu_len = 1
+    frame.payload = DPTBinary(0)
+    frame.src_addr = None
+    frame.dst_addr = PhysicalAddress(0)
+
+    with raises(ConversionError, match=r"src_addr not set"):
+        frame.to_knx()
+
+
+def test_invalid_dst_addr(frame):
+    """Test for invalid dst addr"""
+    frame.code = CEMIMessageCode.L_DATA_IND
+    frame.cmd = APCICommand.GROUP_READ
+    frame.flags = 0
+    frame.mpdu_len = 1
+    frame.payload = DPTBinary(0)
+    frame.src_addr = PhysicalAddress(0)
+    frame.dst_addr = None
+
+    with raises(ConversionError, match=r"dst_addr not set"):
+        frame.to_knx()
+
+
+def test_no_payload(frame):
+    """Test for having no payload set"""
+    frame.code = CEMIMessageCode.L_DATA_IND
+    frame.cmd = APCICommand.GROUP_READ
+    frame.flags = 0
+    frame.mpdu_len = 1
+    frame.payload = None
+    frame.src_addr = PhysicalAddress(0)
+    frame.dst_addr = PhysicalAddress(0)
+
+    assert [41, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0] == frame.to_knx()
+    assert 11 == frame.calculated_length()
+
+
+def test_invalid_payload(frame):
+    """Test for having wrong payload set"""
+    frame.code = CEMIMessageCode.L_DATA_IND
+    frame.cmd = APCICommand.GROUP_READ
+    frame.flags = 0
+    frame.mpdu_len = 1
+    frame.payload = DPTComparator()
+    frame.src_addr = PhysicalAddress(0)
+    frame.dst_addr = PhysicalAddress(0)
+
+    with raises(TypeError):
+        frame.calculated_length()
+
+    with raises(TypeError):
+        frame.to_knx()
+
+
+def test_from_knx_with_not_handleable_cemi(frame):
+    """Test for having unhandlebale cemi set"""
+    with raises(UnsupportedCEMIMessage, match=r".*CEMIMessageCode not implemented:.*"):
+        frame.from_knx(get_data(0x30, 0, 0, 0, 0, 2, 0, []))
+
+
+def test_from_knx_with_not_implemented_cemi(frame):
+    """Test for having not implemented CEMI set"""
+    with raises(UnsupportedCEMIMessage, match=r".*Could not handle CEMIMessageCode:.*"):
+        frame.from_knx(get_data(CEMIMessageCode.L_BUSMON_IND, 0, 0, 0, 0, 2, 0, []))
+
+
+def test_invalid_telegram(frame):
+    """Test for invalid telegram type"""
+    packet_len = frame.from_knx(get_data(0x29, 0, 0, 0, 0, 1, 0, []))
+    frame.cmd = None
+    with raises(ConversionError, match=r".*Telegram not implemented for.*"):
+        tg = frame.telegram
+
+
+def test_set_invalid_telegram(frame):
+    """Test for setting invalid telegram type"""
+    tg = Telegram(telegramtype=None)
+    with raises(TypeError):
+        frame.telegram = tg
+
+
 def test_invalid_invalid_len(frame):
     """Test for invalid cemi len"""
     with raises(UnsupportedCEMIMessage, match=r".*CEMI too small.*"):
diff --git a/test/knxip_tests/routing_indication_test.py b/test/knxip_tests/routing_indication_test.py
index c40f28a6..d62f696e 100644
--- a/test/knxip_tests/routing_indication_test.py
+++ b/test/knxip_tests/routing_indication_test.py
@@ -464,3 +464,10 @@ class Test_KNXIP(unittest.TestCase):
         knxipframe2.init(KNXIPServiceType.ROUTING_INDICATION)
         knxipframe2.from_knx(knxipframe.to_knx())
         self.assertEqual(knxipframe2.body.cemi.telegram, telegram)
+
+    def test_from_knx_invalid_cemi(self):
+        """Test parsing and streaming CEMIFrame KNX/IP packet with unsupported CEMICode."""
+        xknx = XKNX(loop=self.loop)
+        ri = RoutingIndication(xknx)
+
+        self.assertEqual(11, ri.from_knx([43, 0, 0, 0, 0, 0, 0, 0, 2, 0, 0]))
diff --git a/test/remote_value_tests/remote_value_climate_mode_test.py b/test/remote_value_tests/remote_value_climate_mode_test.py
index 4602da5a..2fe0872c 100644
--- a/test/remote_value_tests/remote_value_climate_mode_test.py
+++ b/test/remote_value_tests/remote_value_climate_mode_test.py
@@ -10,6 +10,7 @@ from xknx.remote_value import (
     RemoteValueBinaryOperationMode,
     RemoteValueClimateMode,
 )
+from xknx.remote_value.remote_value_climate_mode import _RemoteValueBinaryClimateMode
 from xknx.telegram import GroupAddress, Telegram
 
 
@@ -46,6 +47,15 @@ class TestRemoteValueDptValue1Ucount(unittest.TestCase):
         )
         self.assertEqual(remote_value.to_knx(HVACOperationMode.NIGHT), DPTBinary(False))
 
+    def test_from_knx_binary_error(self):
+        """Test from_knx function with invalid payload."""
+        xknx = XKNX(loop=self.loop)
+        remote_value = RemoteValueBinaryOperationMode(
+            xknx, operation_mode=HVACOperationMode.COMFORT
+        )
+        with self.assertRaises(CouldNotParseTelegram):
+            remote_value.from_knx(DPTArray((0x9, 0xF)))
+
     def test_to_knx_heat_cool(self):
         """Test to_knx function with normal operation."""
         xknx = XKNX(loop=self.loop)
@@ -65,6 +75,21 @@ class TestRemoteValueDptValue1Ucount(unittest.TestCase):
             remote_value.from_knx(DPTArray((0x02,))), HVACOperationMode.STANDBY
         )
 
+    def test_from_knx_binary_heat_cool(self):
+        """Test from_knx function with invalid payload."""
+        xknx = XKNX(loop=self.loop)
+        remote_value = RemoteValueBinaryHeatCool(
+            xknx, operation_mode=HVACOperationMode.HEAT
+        )
+        with self.assertRaises(CouldNotParseTelegram):
+            remote_value.from_knx(DPTArray((0x9, 0xF)))
+
+    def test_from_knx_operation_mode_error(self):
+        """Test from_knx function with invalid payload."""
+        xknx = XKNX(loop=self.loop)
+        with self.assertRaises(ConversionError):
+            RemoteValueClimateMode(xknx, climate_mode_type=None)
+
     def test_from_knx_binary(self):
         """Test from_knx function with normal operation."""
         xknx = XKNX(loop=self.loop)
@@ -87,6 +112,24 @@ class TestRemoteValueDptValue1Ucount(unittest.TestCase):
             remote_value.from_knx(DPTBinary(False)), HVACOperationMode.COOL
         )
 
+    def test_from_knx_unsupported_operation_mode(self):
+        """Test from_knx function with unsupported operation."""
+        xknx = XKNX(loop=self.loop)
+        with self.assertRaises(ConversionError):
+            RemoteValueBinaryHeatCool(xknx, operation_mode=HVACOperationMode.NODEM)
+
+    def test_from_knx_unknown_operation_mode(self):
+        """Test from_knx function with unsupported operation."""
+        xknx = XKNX(loop=self.loop)
+        with self.assertRaises(ConversionError):
+            RemoteValueBinaryHeatCool(xknx, operation_mode=None)
+
+    def test_supported_operation_modes_not_implemented(self):
+        """Test from_knx function with unsupported operation."""
+        xknx = XKNX(loop=self.loop)
+        with self.assertRaises(NotImplementedError):
+            _RemoteValueBinaryClimateMode.supported_operation_modes()
+
     def test_to_knx_error_operation_mode(self):
         """Test to_knx function with wrong parameter."""
         xknx = XKNX(loop=self.loop)
diff --git a/test/remote_value_tests/remote_value_datetime_test.py b/test/remote_value_tests/remote_value_datetime_test.py
new file mode 100644
index 00000000..c083a071
--- /dev/null
+++ b/test/remote_value_tests/remote_value_datetime_test.py
@@ -0,0 +1,67 @@
+"""Unit test for RemoteValueDateTime objects."""
+import asyncio
+import time
+import unittest
+
+from xknx import XKNX
+from xknx.dpt import DPTArray, DPTBinary, HVACOperationMode
+from xknx.exceptions import ConversionError, CouldNotParseTelegram
+from xknx.remote_value import (
+    RemoteValueBinaryHeatCool,
+    RemoteValueBinaryOperationMode,
+    RemoteValueClimateMode,
+    RemoteValueDateTime,
+)
+from xknx.remote_value.remote_value_climate_mode import _RemoteValueBinaryClimateMode
+from xknx.telegram import GroupAddress, Telegram
+
+
+class TestRemoteValueDateTime(unittest.TestCase):
+    """Test class for RemoteValueDateTime objects."""
+
+    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_init_raises_keyerror(self):
+        """Test init raises KeyError if not supported."""
+        xknx = XKNX(loop=self.loop)
+        with self.assertRaises(ConversionError):
+            RemoteValueDateTime(xknx, value_type="trees_are_important")
+
+    def test_from_knx(self):
+        """Test parsing of RV with datetime object."""
+        xknx = XKNX(loop=self.loop)
+        rv = RemoteValueDateTime(xknx, value_type="datetime")
+        self.assertEqual(
+            rv.from_knx(DPTArray((0x75, 0x0B, 0x1C, 0x17, 0x07, 0x18, 0x20, 0x80))),
+            time.strptime("2017-11-28 23:7:24", "%Y-%m-%d %H:%M:%S"),
+        )
+
+    def test_to_knx(self):
+        """Testing date time object."""
+        xknx = XKNX(loop=self.loop)
+        rv = RemoteValueDateTime(xknx, value_type="datetime")
+        array = rv.to_knx(time.strptime("2017-11-28 23:7:24", "%Y-%m-%d %H:%M:%S"))
+        self.assertEqual(array.value, (0x75, 0x0B, 0x1C, 0x57, 0x07, 0x18, 0x20, 0x80))
+
+    def test_payload_valid(self):
+        """Testing KNX/Byte representation of DPTDateTime object."""
+        xknx = XKNX(loop=self.loop)
+        rv = RemoteValueDateTime(xknx, value_type="datetime")
+        self.assertTrue(
+            rv.payload_valid(DPTArray((0x75, 0x0B, 0x1C, 0x57, 0x07, 0x18, 0x20, 0x80)))
+        )
+
+    def test_payload_invalid(self):
+        """Testing KNX/Byte representation of DPTDateTime object."""
+        xknx = XKNX(loop=self.loop)
+        rv = RemoteValueDateTime(xknx, value_type="datetime")
+        self.assertFalse(
+            rv.payload_valid(DPTArray((0x0B, 0x1C, 0x57, 0x07, 0x18, 0x20, 0x80)))
+        )
diff --git a/test/str_test.py b/test/str_test.py
index e1a45863..dc0e0221 100644
--- a/test/str_test.py
+++ b/test/str_test.py
@@ -48,6 +48,7 @@ from xknx.knxip import (
     KNXIPHeader,
     KNXIPServiceType,
     KNXMedium,
+    RoutingIndication,
     SearchRequest,
     SearchResponse,
     TunnellingAck,
@@ -696,3 +697,15 @@ class TestStringRepresentations(unittest.TestCase):
             str(gateway_descriptor),
             '<GatewayDescriptor name="KNX-Interface" addr="192.168.2.3:1234" local="192.168.2.50@en1" routing="False" tunnelling="True" />',
         )
+
+    #
+    # Routing Indication
+    #
+    def test_routing_indication_str(self):
+        """Test string representation of GatewayDescriptor."""
+        xknx = XKNX(loop=self.loop)
+        ri = RoutingIndication(xknx)
+        self.assertEqual(
+            str(ri),
+            '<RoutingIndication cemi="<CEMIFrame SourceAddress="PhysicalAddress("0.0.0")" DestinationAddress="GroupAddress("0/0/0")" Flags="               0" Command="APCICommand.GROUP_READ" payload="None" />" />',
+        )