Merge branch 'master' of github.com:XKNX/xknx

test/action_test.py

@@ -0,0 +1,106 @@
+"""Unit test for Action objects."""
+import asyncio
+import unittest
+from unittest.mock import Mock, patch
+
+from xknx import XKNX
+from xknx.devices import ActionBase, Action, ActionCallback, BinarySensorState, Light
+
+
+class TestAction(unittest.TestCase):
+    """Class for testing Action 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()
+
+    #
+    # TEST COUNTER
+    #
+    def test_counter(self):
+        """Test counter method."""
+        xknx = XKNX(loop=self.loop)
+        action = ActionBase(xknx, counter=2)
+        self.assertTrue(action.test_counter(None))
+        self.assertFalse(action.test_counter(1))
+        self.assertTrue(action.test_counter(2))
+        self.assertFalse(action.test_counter(3))
+
+    def test_no_counter(self):
+        """Test counter method with no counter set."""
+        xknx = XKNX(loop=self.loop)
+        action = ActionBase(xknx, counter=None)
+        self.assertTrue(action.test_counter(None))
+        self.assertTrue(action.test_counter(1))
+        self.assertTrue(action.test_counter(2))
+        self.assertTrue(action.test_counter(3))
+
+    #
+    # TEST APPLICABLE
+    #
+    def test_if_applicable_hook_on(self):
+        """Test test_if_applicable method with hook set to 'on'."""
+        xknx = XKNX(loop=self.loop)
+        action = ActionBase(xknx, counter=2, hook="on")
+        self.assertTrue(action.test_if_applicable(
+            BinarySensorState.ON, 2))
+        self.assertFalse(action.test_if_applicable(
+            BinarySensorState.ON, 3))
+        self.assertFalse(action.test_if_applicable(
+            BinarySensorState.OFF, 2))
+
+    def test_if_applicable_hook_off(self):
+        """Test test_if_applicable method with hook set to 'off'."""
+        xknx = XKNX(loop=self.loop)
+        action = ActionBase(xknx, counter=2, hook="off")
+        self.assertTrue(action.test_if_applicable(
+            BinarySensorState.OFF, 2))
+        self.assertFalse(action.test_if_applicable(
+            BinarySensorState.OFF, 3))
+        self.assertFalse(action.test_if_applicable(
+            BinarySensorState.ON, 2))
+
+    #
+    # TEST EXECUTE
+    #
+    def test_execute_base_action(self):
+        """Test if execute method of BaseAction shows correct info message."""
+        xknx = XKNX(loop=self.loop)
+        action = ActionBase(xknx)
+        with patch('logging.Logger.info') as mock_info:
+            self.loop.run_until_complete(asyncio.Task(action.execute()))
+            mock_info.assert_called_with('Execute not implemented for %s', 'ActionBase')
+
+    def test_execute_action(self):
+        """Test if execute method of Action calls correct do method of device."""
+        xknx = XKNX(loop=self.loop)
+        light = Light(
+            xknx,
+            'Light1',
+            group_address_switch='1/6/4')
+        xknx.devices.add(light)
+        action = Action(xknx, target='Light1', method='on')
+        with patch('xknx.devices.Light.do') as mock_do:
+            fut = asyncio.Future()
+            fut.set_result(None)
+            mock_do.return_value = fut
+            self.loop.run_until_complete(asyncio.Task(action.execute()))
+            mock_do.assert_called_with('on')
+
+    def test_execute_action_callback(self):
+        """Test if execute method of ActionCallback calls correct callback method."""
+        xknx = XKNX(loop=self.loop)
+        callback = Mock()
+
+        async def async_callback():
+            """Async callback."""
+            callback()
+
+        action = ActionCallback(xknx, async_callback)
+        self.loop.run_until_complete(asyncio.Task(action.execute()))
+        callback.assert_called_with()

test/address_filter_test.py

@@ -1,7 +1,8 @@
 """Unit test for Address class."""
 import unittest
 
-from xknx.knx import AddressFilter
+from xknx.knx import AddressFilter, GroupAddress
+from xknx.exceptions import ConversionError
 
 
 class TestAddressFilter(unittest.TestCase):
@@ -137,3 +138,14 @@ def test_address_combined(self):
         self.assertFalse(af1.match("1/7/10"))
         self.assertFalse(af1.match("2/4/10"))
         self.assertFalse(af1.match("2/1/10"))
+
+    def test_initialize_wrong_format(self):
+        """Test if wrong address format raises exception."""
+        with self.assertRaises(ConversionError):
+            AddressFilter("2-/2,3/4/5/1,5-/*")
+
+    def test_adjust_range(self):
+        """Test helper function _adjust_range."""
+        # pylint: disable=protected-access
+        self.assertEqual(AddressFilter.Range._adjust_range(GroupAddress.MAX_FREE+1), GroupAddress.MAX_FREE)
+        self.assertEqual(AddressFilter.Range._adjust_range(-1), 0)

test/binary_sensor_test.py

@@ -1,7 +1,7 @@
 """Unit test for BinarySensor objects."""
 import asyncio
 import unittest
-from unittest.mock import Mock
+from unittest.mock import Mock, patch
 
 from xknx import XKNX
 from xknx.devices import Action, BinarySensor, BinarySensorState, Switch
@@ -185,6 +185,35 @@ async def async_after_update_callback(device):
         self.loop.run_until_complete(asyncio.Task(switch.process(telegram)))
         after_update_callback.assert_called_with(switch)
 
+    #
+    # TEST COUNTER
+    #
+    def test_counter(self):
+        """Test counter functionality."""
+        xknx = XKNX(loop=self.loop)
+        switch = BinarySensor(xknx, 'TestInput', group_address='1/2/3')
+        with patch('time.time') as mock_time:
+            mock_time.return_value = 1517000000.0
+            self.assertEqual(switch.bump_and_get_counter(BinarySensorState.ON), 1)
+
+            mock_time.return_value = 1517000000.1
+            self.assertEqual(switch.bump_and_get_counter(BinarySensorState.ON), 2)
+
+            mock_time.return_value = 1517000000.2
+            self.assertEqual(switch.bump_and_get_counter(BinarySensorState.OFF), 1)
+
+            mock_time.return_value = 1517000000.3
+            self.assertEqual(switch.bump_and_get_counter(BinarySensorState.ON), 3)
+
+            mock_time.return_value = 1517000000.4
+            self.assertEqual(switch.bump_and_get_counter(BinarySensorState.OFF), 2)
+
+            mock_time.return_value = 1517000002.0  # TIME OUT ...
+            self.assertEqual(switch.bump_and_get_counter(BinarySensorState.ON), 1)
+
+            mock_time.return_value = 1517000004.1  # TIME OUT ...
+            self.assertEqual(switch.bump_and_get_counter(BinarySensorState.OFF), 1)
+
     #
     # STATE ADDRESSES
     #

test/config_test.py

@@ -1,12 +1,14 @@
 """Unit test for Configuration logic."""
 import asyncio
 import unittest
+from unittest.mock import patch
 
 from xknx import XKNX
 from xknx.devices import (Action, BinarySensor, Climate, Cover, DateTime,
                           ExposeSensor, Light, Notification, Scene, Sensor,
                           Switch)
 from xknx.devices.datetime import DateTimeBroadcastType
+from xknx.exceptions import XKNXException
 
 
 # pylint: disable=too-many-public-methods,invalid-name
@@ -304,3 +306,17 @@ def test_config_scene(self):
                 group_address='7/0/1',
                 scene_number=23,
                 device_updated_cb=xknx.devices.device_updated))
+
+    def test_config_file_not_found(self):
+        """Test error message when reading a non exisiting config file."""
+        with patch('logging.Logger.error') as mock_err:
+            XKNX(config='xknx_does_not_exist.yaml', loop=self.loop)
+            self.assertEqual(mock_err.call_count, 1)
+
+    def test_config_file_error(self):
+        """Test error message when reading an errornous config file."""
+        with patch('logging.Logger.error') as mock_err, \
+                patch('xknx.core.Config.parse_group_light') as mock_parse:
+            mock_parse.side_effect = XKNXException()
+            XKNX(config='xknx.yaml', loop=self.loop)
+            self.assertEqual(mock_err.call_count, 1)