Completed Coverage for dbus_opendtu.py: 96%

run_coverage.sh

@@ -1,4 +1,6 @@
 #!/bin/bash
 
 python3-coverage run -m unittest discover -s tests -p "test_*.py"
-python3-coverage report -m
+python3-coverage html
+#python3-coverage report -m
+#zip -r htmlcov.zip htmlcov

tests/test_dbus_opendtu.py

@@ -1,7 +1,7 @@
 """ Unit tests for the dbus_opendtu.py module """
 
 import unittest
-from unittest.mock import patch, MagicMock, mock_open
+from unittest.mock import patch, MagicMock, mock_open, ANY
 import sys
 import os
 import configparser
@@ -12,11 +12,21 @@
 # Mocking the dbus and other dependencies before importing the module to test
 sys.modules['dbus'] = MagicMock()
 sys.modules['vedbus'] = MagicMock()
+# Mock the gi.repository.GLib module
+sys.modules['gi'] = MagicMock()
 sys.modules['gi.repository'] = MagicMock()
+sys.modules['gi.repository.GLib'] = MagicMock()
+sys.modules['gi.repository.GLib.MainLoop'] = MagicMock()
 sys.modules['dbus.mainloop.glib'] = MagicMock()
 
 
-from dbus_opendtu import get_DbusServices, getConfig  # pylint: disable=E0401,C0413 # noqa: E402
+from dbus_opendtu import (  # pylint: disable=E0401,C0413
+    get_DbusServices,
+    getConfig,
+    sign_of_life_all_services,
+    update_all_services,
+    main
+)  # noqa
 
 
 class TestDbusOpendtu(unittest.TestCase):
@@ -151,6 +161,29 @@ def test_get_dbus_services_with_no_inverters_or_templates(self, mock_dbus_servic
         self.assertEqual(len(services), 0)
         mock_dbus_service.assert_not_called()
 
+    @patch("dbus_opendtu.DbusService")
+    def test_get_config_with_invalid_NumberOfInverter_and_Template_values(self, mock_dbus_service):
+        """ Test get_DbusServices with invalid NumberOfInverter and NumberOfTemplate values """
+        # Create a mock config with the required values
+        config = {
+            "DEFAULT": {
+                "NumberOfInvertersToQuery": "invalid",
+                "NumberOfTemplates": "invalid",
+                "DTU": "openDTU"
+            },
+            "INVERTER0": {},  # Add the required key to avoid KeyError
+            "TEMPLATE0": {}   # Add the required key to avoid KeyError
+        }
+
+        # Mock the get_number_of_inverters method to return 0
+        mock_dbus_service_instance = mock_dbus_service.return_value
+        mock_dbus_service_instance.get_number_of_inverters.return_value = 0
+
+        services = get_DbusServices(config)
+
+        self.assertEqual(len(services), 0)
+        mock_dbus_service.assert_not_called()
+
     @patch('dbus_opendtu.DbusService')
     @patch('dbus_opendtu.get_config_value')
     def test_get_dbus_services_with_missing_dtu_key(self, mock_get_config_value, mock_dbus_service):
@@ -170,8 +203,192 @@ def test_get_dbus_services_with_missing_dtu_key(self, mock_get_config_value, moc
         self.assertIsNone(services)
         mock_dbus_service.assert_not_called()
 
-        if __name__ == '__main__':
-            unittest.main()
+    def test_sign_of_life_all_services(self):
+        """ Test sign_of_life_all_services with a list of mock services """
+        # Create a list of mock services
+        mock_service_1 = MagicMock()
+        mock_service_2 = MagicMock()
+        services = [mock_service_1, mock_service_2]
+
+        # Call the function
+        result = sign_of_life_all_services(services)
+
+        # Verify that the sign_of_life method was called on each service
+        mock_service_1.sign_of_life.assert_called_once()
+        mock_service_2.sign_of_life.assert_called_once()
+
+        # Verify the return value
+        self.assertTrue(result)
+
+    def test_sign_of_life_all_services_with_empty_list(self):
+        """ Test sign_of_life_all_services with an empty list """
+        services = []
+
+        # Call the function
+        result = sign_of_life_all_services(services)
+
+        # Verify the return value
+        self.assertTrue(result)
+
+    def test_sign_of_life_all_services_with_no_sign_of_life_method(self):
+        """ Test sign_of_life_all_services with services missing sign_of_life method """
+        # Create a list of mock services, one without sign_of_life method
+        mock_service_1 = MagicMock()
+        mock_service_2 = MagicMock()
+        del mock_service_2.sign_of_life
+        services = [mock_service_1, mock_service_2]
+
+        # Call the function and expect an AttributeError
+        with self.assertRaises(AttributeError):
+            sign_of_life_all_services(services)
+
+    @patch('dbus_opendtu.gobject')
+    def test_update_all_services(self, mock_gobject):
+        """ Test update_all_services with valid services """
+        # Mock the current time
+        mock_gobject.get_real_time.return_value = 2000000
+
+        # Create mock services
+        mock_service_1 = MagicMock()
+        mock_service_1.polling_interval = 1000
+        mock_service_1.last_polling = 1000
+
+        mock_service_2 = MagicMock()
+        mock_service_2.polling_interval = 2000
+        mock_service_2.last_polling = 1000
+
+        services = [mock_service_1, mock_service_2]
+
+        # Call the function
+        result = update_all_services(services)
+
+        # Verify that the update method was called on each service
+        mock_service_1.update.assert_called_once()
+        mock_service_2.update.assert_not_called()
+
+        # Verify that the last_polling attribute was updated
+        self.assertEqual(mock_service_1.last_polling, 2000)
+        self.assertEqual(mock_service_2.last_polling, 1000)
+
+        # Verify the return value
+        self.assertTrue(result)
+
+    @patch('dbus_opendtu.gobject')
+    def test_update_all_services_with_no_update_needed(self, mock_gobject):
+        """ Test update_all_services when no update is needed """
+        # Mock the current time
+        mock_gobject.get_real_time.return_value = 2000000
+
+        # Create mock services
+        mock_service_1 = MagicMock()
+        mock_service_1.polling_interval = 1000
+        mock_service_1.last_polling = 1999
+
+        mock_service_2 = MagicMock()
+        mock_service_2.polling_interval = 2000
+        mock_service_2.last_polling = 1999
+
+        services = [mock_service_1, mock_service_2]
+
+        # Call the function
+        result = update_all_services(services)
+
+        # Verify that the update method was not called on any service
+        mock_service_1.update.assert_not_called()
+        mock_service_2.update.assert_not_called()
+
+        # Verify the return value
+        self.assertTrue(result)
+
+    @patch('dbus_opendtu.gobject')
+    def test_update_all_services_with_empty_list(self, mock_gobject):
+        """ Test update_all_services with an empty list """
+        services = []
+
+        # Call the function
+        result = update_all_services(services)
+
+        # Verify the return value
+        self.assertTrue(result)
+
+    @patch('dbus_opendtu.gobject')
+    def test_update_all_services_with_missing_attributes(self, mock_gobject):
+        """ Test update_all_services with services missing required attributes """
+        # Mock the current time
+        mock_gobject.get_real_time.return_value = 2000000
+
+        # Create mock services, one missing required attributes
+        mock_service_1 = MagicMock()
+        mock_service_1.polling_interval = 1000
+        mock_service_1.last_polling = 1000
+
+        mock_service_2 = MagicMock()
+        del mock_service_2.polling_interval
+        del mock_service_2.last_polling
+
+        services = [mock_service_1, mock_service_2]
+
+        # Call the function and expect an AttributeError
+        with self.assertRaises(AttributeError):
+            update_all_services(services)
+
+    @patch('dbus_opendtu.getConfig')
+    @patch('dbus_opendtu.get_config_value')
+    @patch('dbus_opendtu.get_DbusServices')
+    @patch('dbus_opendtu.sign_of_life_all_services')
+    @patch('dbus_opendtu.update_all_services')
+    @patch('dbus_opendtu.gobject')
+    def test_main(
+        self,
+        mock_gobject,
+        mock_update_all_services,
+        mock_sign_of_life_all_services,
+        mock_get_dbus_services,
+        mock_get_config_value,
+        mock_get_config,
+    ):
+        """ Test the main function """
+        # Mock the configuration
+        mock_config = MagicMock()
+        mock_get_config.return_value = mock_config
+        mock_get_config_value.return_value = 1
+
+        # Mock the services
+        mock_services = [MagicMock()]
+        mock_get_dbus_services.return_value = mock_services
+
+        # Mock the timeout_add method
+        def timeout_add_mock(interval, callback, *args, **kwargs):
+            callback(*args, **kwargs)
+            return True
+
+        mock_gobject.timeout_add.side_effect = timeout_add_mock
+
+        # Call the main function
+        main()
+
+        # Assertions to verify the behavior
+        mock_get_config.assert_called_once()
+        mock_get_dbus_services.assert_called_once_with(mock_config)
+        mock_update_all_services.assert_called_once_with(mock_services)
+        mock_sign_of_life_all_services.assert_called_once_with(mock_services)
+        mock_gobject.MainLoop.assert_called_once()
+
+    @patch('dbus_opendtu.get_DbusServices')
+    @patch('dbus_opendtu.gobject')
+    @patch('dbus_opendtu.logging')
+    def test_main_exception(
+        self,
+        mock_logging,
+        mock_gobject,
+        mock_get_dbus_services,  # pylint: disable=W0613
+    ):
+        """ Test the main function with exception """
+        # Mock gobject.MainLoop to raise an exception
+        mock_gobject.MainLoop.side_effect = Exception("Test exception")
+
+        main()
+        mock_logging.critical.assert_called_once_with("Error at %s", "main",  exc_info=ANY)
 
 
 if __name__ == '__main__':