From 847c7ca190f39a956bb1b154058dcad0732c47ba Mon Sep 17 00:00:00 2001
From: David Trattnig <david.trattnig@o94.at>
Date: Tue, 13 Jul 2021 16:06:42 +0200
Subject: [PATCH] New test cases. #78
---
tests/test.py | 225 -----------------------------
tests/test_config.py | 73 ++++++++++
tests/test_engine_executor.py | 264 ++++++++++++++++++++++++++++++++++
tests/test_logger.py | 47 ++++++
4 files changed, 384 insertions(+), 225 deletions(-)
delete mode 100644 tests/test.py
create mode 100644 tests/test_config.py
create mode 100644 tests/test_engine_executor.py
create mode 100644 tests/test_logger.py
diff --git a/tests/test.py b/tests/test.py
deleted file mode 100644
index 6ef286a9..00000000
--- a/tests/test.py
+++ /dev/null
@@ -1,225 +0,0 @@
-
-#
-# Aura Engine (https://gitlab.servus.at/aura/engine)
-#
-# Copyright (C) 2017-2020 - 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 os
-import unittest
-import validators
-import time
-
-from datetime import datetime
-
-from src.base.utils import SimpleUtil as SU
-from src.base.logger import AuraLogger
-from src.base.config import AuraConfig
-from src.control import EngineExecutor
-
-
-
-class TestLogger(unittest.TestCase):
- """
- Testing the Logger.
- """
- aura_logger = None
-
- def setUp(self):
- self.config = AuraConfig()
- self.aura_logger = AuraLogger(self.config)
-
- def test_logger(self):
- self.assertTrue(self.aura_logger.logger.hasHandlers())
-
-
-class TestConfig(unittest.TestCase):
- """
- Testing the Configuration.
- """
- config = None
-
- def setUp(self):
- self.config = AuraConfig()
-
-
- def test_config(self):
- # Check if config is available
- self.assertIsNotNone(self.config.ini_path)
-
- # Check if "install_dir" is a valid directory (is evaluated at runtime)
- self.assertTrue(os.path.isdir(self.config.get("install_dir")))
-
- # Check API Urls
- self.assertTrue(validators.url(self.config.get("api_steering_status")))
- self.assertTrue(validators.url(self.config.get("api_steering_calendar")))
- self.assertTrue(validators.url(self.config.get("api_tank_status")))
- tank_playlist_url = self.config.get("api_tank_playlist").replace("${ID}", "1")
- self.assertTrue(validators.url(tank_playlist_url))
- self.assertTrue(validators.url(self.config.get("api_engine_status")))
- self.assertTrue(validators.url(self.config.get("api_engine_store_playlog")))
- self.assertTrue(validators.url(self.config.get("api_engine_store_clock")))
- engine_health_url = self.config.get("api_engine_store_health").replace("${ENGINE_NUMBER}", "1")
- self.assertTrue(validators.url(engine_health_url))
-
- # Check if Liquidsoap "socket_dir" is set and a directory
- self.assertTrue(os.path.isdir(self.config.get("socket_dir")))
-
- # Check if database settings are set
- self.assertIsNotNone(self.config.get("db_user"))
- self.assertIsNotNone(self.config.get("db_pass"))
- self.assertIsNotNone(self.config.get("db_name"))
- self.assertIsNotNone(self.config.get("db_host"))
-
-
-
-class TestEngineExecutor(unittest.TestCase):
- """
- Testing the EngineExecutor.
- """
-
- def setUp(self):
- None
-
-
- def test_single_executor(self):
- # Initialize state and executor params
- global_state = ["none"]
- due_time = SU.timestamp() + 2
- def f(param):
- global_state[0] = param
-
- # Before the executor is done there should be the initial value
- e = EngineExecutor("RANDOM_NAMESPACE", None, due_time, f, "hello world")
- self.assertEqual("none", global_state[0])
- self.assertNotEqual("hello world", global_state[0])
-
- # After 3 seconds there should be the updated value
- time.sleep(5)
- self.assertEqual("hello world", global_state[0])
-
-
-
- def test_two_executors(self):
- # Initialize state and executor params
- global_state = ["none"]
- def f(param):
- global_state[0] = param
-
- # Before the executor 1 is done there should be the initial value
- due_time1 = SU.timestamp() + 4
- e1 = EngineExecutor("EXECUTOR_1", None, due_time1, f, "hello world from executor 1")
- self.assertEqual("none", global_state[0])
- self.assertNotEqual("hello world from executor 1", global_state[0])
-
- # Before the executor 2 is done there should be still the initial value
- due_time2 = SU.timestamp() + 2
- e2 = EngineExecutor("EXECUTOR_2", None, due_time2, f, "hello world from executor 2")
- self.assertEqual("none", global_state[0])
- self.assertNotEqual("hello world from executor 2", global_state[0])
-
- # After 1 second there still should be the initial value
- time.sleep(1)
- self.assertEqual("none", global_state[0])
-
- # After 3 seconds max there should be the updated value from executor 2
- time.sleep(2)
- self.assertEqual("hello world from executor 2", global_state[0])
-
- # After 5 seconds max there should be the updated value from executor 1
- time.sleep(5)
- self.assertEqual("hello world from executor 1", global_state[0])
-
-
-
- def test_parent_child_executors_in_order(self):
- # Initialize state and executor params
- global_state = ["none"]
- def f(param):
- global_state[0] = param
-
- # Before the the parent is done there should be the initial value
- due_time1 = SU.timestamp() + 1
- parent = EngineExecutor("EXECUTOR_PARENT", None, due_time1, f, "hello world from parent")
- self.assertEqual("none", global_state[0])
-
- # Before the the child is done there should be the initial value
- due_time2 = SU.timestamp() + 3
- child = EngineExecutor("EXECUTOR_CHILD", parent, due_time2, f, "hello world from child")
- self.assertEqual("none", global_state[0])
-
- # After 0.5 seconds there still should be the initial value
- time.sleep(0.5)
- self.assertEqual("none", global_state[0])
-
- # After 2 seconds max there should be the updated value from parent executor
- time.sleep(2)
- self.assertEqual("hello world from parent", global_state[0])
-
- # After 4 seconds max there should be the updated value from child executor
- time.sleep(4)
- self.assertEqual("hello world from child", global_state[0])
-
-
-
- def test_parent_child_executors_with_child_before(self):
- # Initialize state and executor params
- global_state = ["none", "never called by parent"]
- def f(param):
- global_state[0] = param
- if param == "hello world from parent":
- global_state[1] = param
-
-
- # Before the the parent is done there should be the initial value
- due_time1 = SU.timestamp() + 3
- parent = EngineExecutor("EXECUTOR_PARENT", None, due_time1, f, "hello world from parent")
- self.assertEqual("none", global_state[0])
-
- # Before the the child is done there should be the initial value
- due_time2 = SU.timestamp() + 1
- child = EngineExecutor("EXECUTOR_CHILD", parent, due_time2, f, "hello world from child")
- self.assertEqual("none", global_state[0])
-
- # After 0.5 seconds there still should be the initial value
- time.sleep(0.5)
- self.assertEqual("none", global_state[0])
-
- # After 2 seconds max there isn't a setting from the child yet, because it's waiting for the parent
- time.sleep(2)
- self.assertNotEqual("hello world from child", global_state[0])
-
- # But the parent didn't set anything either, because it's scheduled for later
- self.assertNotEqual("hello world from parent", global_state[0])
- self.assertEqual("none", global_state[0])
-
- # Double check if it has ever been called by parent
- self.assertEqual("never called by parent", global_state[1])
-
- # After 4 seconds max there should be the updated value from parent & child
- # Because the child is due before the parent, it is executed right away,
- # hence overwriting the value just set by the parent
- time.sleep(4)
- self.assertNotEqual("hello world from parent", global_state[0])
- self.assertEqual("hello world from child", global_state[0])
-
- # But we do not just believe what we expect, but check if it really has ever been called by a parent:
- self.assertEqual("hello world from parent", global_state[1])
-
-
-
-if __name__ == '__main__':
- unittest.main()
\ No newline at end of file
diff --git a/tests/test_config.py b/tests/test_config.py
new file mode 100644
index 00000000..392ebc46
--- /dev/null
+++ b/tests/test_config.py
@@ -0,0 +1,73 @@
+
+#
+# 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 os
+import unittest
+import validators
+
+from src.base.utils import SimpleUtil as SU
+from src.base.config import AuraConfig
+
+
+
+
+class TestConfig(unittest.TestCase):
+ """
+ Testing the Configuration.
+ """
+ config = None
+
+ def setUp(self):
+ self.config = AuraConfig()
+
+
+ def test_config(self):
+ # Check if config is available
+ self.assertIsNotNone(self.config.ini_path)
+
+ # Check if "install_dir" is a valid directory (is evaluated at runtime)
+ self.assertTrue(os.path.isdir(self.config.get("install_dir")))
+
+ # Check API Urls
+ self.assertTrue(validators.url(self.config.get("api_steering_status")))
+ self.assertTrue(validators.url(self.config.get("api_steering_calendar")))
+ self.assertTrue(validators.url(self.config.get("api_tank_status")))
+ tank_playlist_url = self.config.get("api_tank_playlist").replace("${ID}", "1")
+ self.assertTrue(validators.url(tank_playlist_url))
+ self.assertTrue(validators.url(self.config.get("api_engine_status")))
+ self.assertTrue(validators.url(self.config.get("api_engine_store_playlog")))
+ self.assertTrue(validators.url(self.config.get("api_engine_store_clock")))
+ engine_health_url = self.config.get("api_engine_store_health").replace("${ENGINE_NUMBER}", "1")
+ self.assertTrue(validators.url(engine_health_url))
+
+ # Check if Liquidsoap "socket_dir" is set and a directory
+ self.assertTrue(os.path.isdir(self.config.get("socket_dir")))
+
+ # Check if database settings are set
+ self.assertIsNotNone(self.config.get("db_user"))
+ self.assertIsNotNone(self.config.get("db_pass"))
+ self.assertIsNotNone(self.config.get("db_name"))
+ self.assertIsNotNone(self.config.get("db_host"))
+
+
+
+
+if __name__ == '__main__':
+ unittest.main()
\ No newline at end of file
diff --git a/tests/test_engine_executor.py b/tests/test_engine_executor.py
new file mode 100644
index 00000000..9977dd5a
--- /dev/null
+++ b/tests/test_engine_executor.py
@@ -0,0 +1,264 @@
+
+#
+# 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/>.
+
+
+from src.engine import Engine
+import unittest
+import time
+
+from src.base.utils import SimpleUtil as SU
+from src.control import EngineExecutor
+
+
+
+
+class TestEngineExecutor(unittest.TestCase):
+ """
+ Testing the EngineExecutor.
+ """
+
+ def setUp(self):
+ None
+
+
+
+ def test_single_executor(self):
+ # Initialize state and executor params
+ EngineExecutor.timer_store = {}
+ global_state = ["none"]
+ due_time = SU.timestamp() + 0.3
+ def f(param):
+ global_state[0] = param
+
+ # Before the executor is done there should be the initial value
+ e = EngineExecutor("RANDOM_NAMESPACE", None, due_time, f, "hello world")
+ self.assertEqual("none", global_state[0])
+ self.assertNotEqual("hello world", global_state[0])
+
+ # After 0.5 seconds there should be the updated value
+ time.sleep(0.5)
+ self.assertEqual("hello world", global_state[0])
+
+
+
+ def test_two_executors(self):
+ # Initialize state and executor params
+ EngineExecutor.timer_store = {}
+ global_state = ["none"]
+ def f(param):
+ global_state[0] = param
+
+ # Before the executor 1 is done there should be the initial value
+ due_time1 = SU.timestamp() + 1
+ e1 = EngineExecutor("EXECUTOR_1", None, due_time1, f, "hello world from executor 1")
+ self.assertEqual("none", global_state[0])
+ self.assertNotEqual("hello world from executor 1", global_state[0])
+
+ # Before the executor 2 is done there should be still the initial value
+ due_time2 = SU.timestamp() + 0.5
+ e2 = EngineExecutor("EXECUTOR_2", None, due_time2, f, "hello world from executor 2")
+ self.assertEqual("none", global_state[0])
+ self.assertNotEqual("hello world from executor 2", global_state[0])
+
+ # After 0.3 seconds there still should be the initial value
+ time.sleep(0.3)
+ self.assertEqual("none", global_state[0])
+
+ # After 0.6 seconds max there should be the updated value from executor 2
+ time.sleep(0.3)
+ self.assertEqual("hello world from executor 2", global_state[0])
+
+ # After 1.1 seconds max there should be the updated value from executor 1
+ time.sleep(0.5)
+ self.assertEqual("hello world from executor 1", global_state[0])
+
+
+
+ def test_parent_child_executors_in_order(self):
+ # Initialize state and executor params
+ EngineExecutor.timer_store = {}
+ global_state = ["none"]
+ def f(param):
+ global_state[0] = param
+
+ # Before the the parent is done there should be the initial value
+ due_time1 = SU.timestamp() + 0.5
+ parent = EngineExecutor("EXECUTOR_PARENT", None, due_time1, f, "hello world from parent")
+ self.assertEqual("none", global_state[0])
+
+ # Before the the child is done there should be the initial value
+ due_time2 = SU.timestamp() + 1
+ child = EngineExecutor("EXECUTOR_CHILD", parent, due_time2, f, "hello world from child")
+ self.assertEqual("none", global_state[0])
+
+ # After 0.3 seconds there still should be the initial value
+ time.sleep(0.3)
+ self.assertEqual("none", global_state[0])
+
+ # After 0.6 seconds max there should be the updated value from parent executor
+ time.sleep(0.3)
+ self.assertEqual("hello world from parent", global_state[0])
+
+ # After 1.2 seconds max there should be the updated value from child executor
+ time.sleep(1.2)
+ self.assertEqual("hello world from child", global_state[0])
+
+
+
+ def test_parent_child_executors_with_child_before(self):
+ # Initialize state and executor params
+ EngineExecutor.timer_store = {}
+ global_state = ["none", "never called by parent"]
+ def f(param):
+ global_state[0] = param
+ if param == "hello world from parent":
+ global_state[1] = param
+
+
+ # Before the the parent is done there should be the initial value
+ due_time1 = SU.timestamp() + 0.5
+ parent = EngineExecutor("EXECUTOR_PARENT", None, due_time1, f, "hello world from parent")
+ self.assertEqual("none", global_state[0])
+
+ # Before the the child is done there should be the initial value
+ due_time2 = SU.timestamp() + 1.5
+ child = EngineExecutor("EXECUTOR_CHILD", parent, due_time2, f, "hello world from child")
+ self.assertEqual("none", global_state[0])
+
+ # After 0.2 seconds there still should be the initial value
+ time.sleep(0.2)
+ self.assertEqual("none", global_state[0])
+
+ # After 0.4 seconds max there isn't a setting from the child yet, because it's waiting for the parent
+ time.sleep(0.2)
+ self.assertNotEqual("hello world from child", global_state[0])
+
+ # But the parent didn't set anything either, because it's scheduled for later
+ self.assertNotEqual("hello world from parent", global_state[0])
+ self.assertEqual("none", global_state[0])
+
+ # Double check if it has ever been called by parent
+ self.assertEqual("never called by parent", global_state[1])
+
+ # After 2.2 seconds max there should be the updated value from parent & child
+ # Because the child is due before the parent, it is executed right away,
+ # hence overwriting the value just set by the parent
+ time.sleep(2.2)
+ self.assertNotEqual("hello world from parent", global_state[0])
+ self.assertEqual("hello world from child", global_state[0])
+
+ # But we do not just believe what we expect, but check if it really has ever been called by a parent
+ self.assertEqual("hello world from parent", global_state[1])
+
+
+
+
+ def test_timer_store_replacement(self):
+ # Initialize state and executor params
+ EngineExecutor.timer_store = {}
+ global_state = ["none"]
+ def f(param):
+ global_state[0] = param
+
+ # There should be a total of 0 timers
+ timers = EngineExecutor.command_history()
+ self.assertEqual(0, len(timers))
+
+ # Before the the parent is done there should be the initial value
+ due_time1 = SU.timestamp() + 0.5
+ parent = EngineExecutor("EXECUTOR_PARENT", None, due_time1, f, "hello world from parent")
+ self.assertEqual("none", global_state[0])
+
+ # Before the the child is done there should be the initial value
+ due_time2 = SU.timestamp() + 1.5
+ child = EngineExecutor("EXECUTOR_CHILD", parent, due_time2, f, "hello world from child")
+ self.assertEqual("none", global_state[0])
+
+ # There should be a total of 2 timers
+ timers = EngineExecutor.command_history()
+ self.assertEqual(2, len(timers))
+
+ # Replacing the parent with a new instance
+ parent = EngineExecutor("EXECUTOR_PARENT", None, due_time1, f, "hello world from alternative parent")
+ self.assertEqual("none", global_state[0])
+
+ # Before the the child is done there should be the initial value
+ child = EngineExecutor("EXECUTOR_CHILD", parent, due_time2, f, "hello world from alternative child")
+ self.assertEqual("none", global_state[0])
+
+ # After 1 seconds max there should be the updated value from the alternative parent
+ time.sleep(1)
+ self.assertEqual("hello world from alternative parent", global_state[0])
+
+ # After 2 seconds max there should be the updated value from the alternative child
+ time.sleep(2)
+ self.assertEqual("hello world from alternative child", global_state[0])
+
+ # There should be a total of 2 timers, even though 4 got instantiated
+ timers = EngineExecutor.command_history()
+ self.assertEqual(2, len(timers))
+
+
+
+
+ def test_dead_parent_with_lively_child(self):
+ # Initialize state and executor params
+ EngineExecutor.timer_store = {}
+ global_state = ["none"]
+ def f(param):
+ global_state[0] = param
+
+ # Before the the parent is done there should be the initial value
+ due_time1 = SU.timestamp() + 0.5
+ parent = EngineExecutor("EXECUTOR_PARENT", None, due_time1, f, "hello world from parent")
+ self.assertEqual("none", global_state[0])
+
+ # Before the the child is done there should be the initial value
+ due_time2 = SU.timestamp() + 1.5
+ child = EngineExecutor("EXECUTOR_CHILD", parent, due_time2, f, "hello world from child")
+ self.assertEqual("none", global_state[0])
+
+ # Wait until the parent timer got executed
+ time.sleep(1)
+ self.assertEqual("hello world from parent", global_state[0])
+
+ # Parent dead - child alive
+ self.assertEqual(False, parent.is_alive())
+ self.assertEqual(True, child.is_alive())
+
+ # Replacing the parent & child with a new instance
+ parent = EngineExecutor("EXECUTOR_PARENT", None, due_time1, f, "hello world from late parent")
+ child = EngineExecutor("EXECUTOR_PARENT", None, due_time2, f, "hello world from alternative child")
+
+ # New parent = dead before finished initialization already, so actually never born
+ self.assertEqual(False, parent.is_alive())
+
+ # Even though the late parent would be executed by now, it wasn't, because the initial parent
+ # was finished at instatiation time already
+ self.assertEqual("hello world from parent", global_state[0])
+
+ # Finally there should be the updated value from the alternative child
+ time.sleep(2)
+ self.assertEqual("hello world from alternative child", global_state[0])
+
+
+
+
+if __name__ == '__main__':
+ unittest.main()
\ No newline at end of file
diff --git a/tests/test_logger.py b/tests/test_logger.py
new file mode 100644
index 00000000..dbbdcf6c
--- /dev/null
+++ b/tests/test_logger.py
@@ -0,0 +1,47 @@
+
+#
+# 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 src.base.utils import SimpleUtil as SU
+from src.base.logger import AuraLogger
+from src.base.config import AuraConfig
+
+
+
+
+class TestLogger(unittest.TestCase):
+ """
+ Testing the Logger.
+ """
+ aura_logger = None
+
+ def setUp(self):
+ self.config = AuraConfig()
+ self.aura_logger = AuraLogger(self.config)
+
+ def test_logger(self):
+ self.assertTrue(self.aura_logger.logger.hasHandlers())
+
+
+
+
+if __name__ == '__main__':
+ unittest.main()
\ No newline at end of file
--
GitLab