Merge branch 'improve-test-coverage-monitor-142' into 'main'

Improve test coverage for monitor

See merge request !39

src/aura_engine/plugins/monitor.py

@@ -112,6 +112,9 @@ class AuraMonitor:
 
         self.engine_id = self.get_engine_id()
 
+    def __del__(self):
+        self.heartbeat_socket.close()
+
     #
     # EVENTS
     #
@@ -309,10 +312,9 @@ class AuraMonitor:
                     {"engine_id": self.engine_id, "status": status}
                 )
 
-        heartbeat_frq = self.config.monitoring.heartbeat.frequency  # default: 1
-        if int(heartbeat_frq or 0) < 1:
-            heartbeat_frq = 1
-        threading.Timer(heartbeat_frq, self.heartbeat).start()
+        heartbeat_frq = self.config.monitoring.heartbeat.frequency  # default: 1, disable: 0
+        if heartbeat_frq > 1:
+            threading.Timer(heartbeat_frq, self.heartbeat).start()
 
     def validate_url_connection(self, url):
         """
@@ -386,3 +388,5 @@ class AuraMonitor:
         except OSError:
             self.logger.critical(SU.red("Error while accessing network via <broadcast>!"))
             return "<UNKNOWN NETWORK>"
+        finally:
+            s.close()

tests/test_plugins_monitor.py

+#
+# Aura Engine (https://gitlab.servus.at/aura/engine)
+#
+# Copyright (C) 2017-now() - The Aura Engine Team.
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+
+import unittest
+
+from aura_engine.base.config import AuraConfig
+from aura_engine.base.utils import SimpleUtil as SU
+from aura_engine.plugins.monitor import AuraMonitor
+
+#
+# Mock
+#
+
+
+class MockedMixer:
+    """
+    Mocked version of mixer.
+    """
+
+    def get_inputs(self) -> dict:
+        state = {
+            "ready": True,
+            "selected": True,
+            "single": True,
+            "volume": 0,
+            "remaining": 0.0,
+        }
+        return {
+            "in_queue_0": state,
+            "in_queue_1": state,
+            "in_stream_0": state,
+            "in_stream_1": state,
+            "in_line_0": state,
+            "in_line_1": state,
+        }
+
+    def get_outputs(self) -> list:
+        return ["lineout_0", "out_http_0"]
+
+
+class MockedPlayer:
+    """
+    Mocked version of engine.
+    """
+
+    mixer = MockedMixer()
+
+
+class MockedEventDispather:
+    """
+    Mocked version of event dispatcher.
+    """
+
+    def on_sick(self, data):
+        pass
+
+
+class MockedEngine:
+    """
+    Mocked version of engine.
+    """
+
+    player = MockedPlayer()
+    event_dispatcher = MockedEventDispather()
+
+    def init_version(self):
+        versions = {"control": 1, "core": 1, "liquidsoap": 1}
+        AuraConfig.instance.init_version(versions)
+
+    def update_playout_state(self):
+        return True
+
+
+class TestPluginsMonitor(unittest.TestCase):
+    """
+    Testing the monitor plugin.
+    """
+
+    monitor: AuraMonitor
+    config: AuraConfig
+
+    # Setup and teardown
+
+    def setUp(self):
+        self.config = AuraConfig.instance.config
+        self.monitor = AuraMonitor(MockedEngine())
+
+    def tearDown(self):
+        pass
+
+    #
+    # Test Cases
+    #
+
+    def test_has_valid_status_vitality_only(self):
+        print(self._testMethodName)
+
+        # will fail due to missing info
+        status = self.monitor.has_valid_status(update_vitality_only=True)
+        self.assertFalse(status)
+
+    def test_has_valid_status(self):
+        print(self._testMethodName)
+
+        # mock valid status by swizzling some methonds
+        def validate_url_connection(url: str) -> bool:
+            return True
+
+        def validate_directory(dir_path: str) -> dict:
+            return {"path": dir_path, "exists": True, "has_content": True}
+
+        def get_url_response(url: str) -> dict:
+            return {"auth": "string", "importer": "string", "store": "string"}
+
+        def get_ip() -> str:
+            return "0.0.0.0"
+
+        self.monitor.validate_url_connection = validate_url_connection
+        self.monitor.validate_directory = validate_directory
+        self.monitor.get_url_response = get_url_response
+        self.monitor.get_ip = get_ip
+
+        status = self.monitor.has_valid_status(update_vitality_only=False)
+        self.assertTrue(status)
+
+    def test_validate_url_connection(self):
+        print(self._testMethodName)
+
+        self.assertTrue(self.monitor.validate_url_connection("https://debian.org/"))
+        self.assertFalse(self.monitor.validate_url_connection("https://debian.123/"))
+
+    def test_validate_directory(self):
+        print(self._testMethodName)
+
+        path = "./tests"
+        status = self.monitor.validate_directory(path)
+        self.assertEqual(status["path"], path)
+        self.assertEqual(status["exists"], True)
+        self.assertEqual(status["has_content"], True)
+
+        path = "./tests_"
+        status = self.monitor.validate_directory(path)
+        self.assertEqual(status["path"], path)
+        self.assertEqual(status["exists"], False)
+        self.assertEqual(status["has_content"], False)
+
+    def test_post_health(self):
+        print(self._testMethodName)
+
+        self.config.monitoring.heartbeat.frequency = 0  # disable timer
+        self.monitor.post_health(data=None, is_healthy=True)
+
+
+if __name__ == "__main__":
+    unittest.main()