Commit 847c7ca1 authored by david's avatar david
Browse files

New test cases. #78

parent f593acdd
#
# 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
......@@ -2,89 +2,31 @@
#
# Aura Engine (https://gitlab.servus.at/aura/engine)
#
# Copyright (C) 2017-2020 - The Aura Engine Team.
#
# 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
from src.engine import Engine
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):
"""
......@@ -95,10 +37,12 @@ class TestEngineExecutor(unittest.TestCase):
None
def test_single_executor(self):
# Initialize state and executor params
EngineExecutor.timer_store = {}
global_state = ["none"]
due_time = SU.timestamp() + 2
due_time = SU.timestamp() + 0.3
def f(param):
global_state[0] = param
......@@ -107,76 +51,79 @@ class TestEngineExecutor(unittest.TestCase):
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)
# 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() + 4
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() + 2
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 1 second there still should be the initial value
time.sleep(1)
# After 0.3 seconds there still should be the initial value
time.sleep(0.3)
self.assertEqual("none", global_state[0])
# After 3 seconds max there should be the updated value from executor 2
time.sleep(2)
# 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 5 seconds max there should be the updated value from executor 1
time.sleep(5)
# 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() + 1
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() + 3
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)
# After 0.3 seconds there still should be the initial value
time.sleep(0.3)
self.assertEqual("none", global_state[0])
# After 2 seconds max there should be the updated value from parent executor
time.sleep(2)
# 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 4 seconds max there should be the updated value from child executor
time.sleep(4)
# 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
......@@ -185,21 +132,21 @@ class TestEngineExecutor(unittest.TestCase):
# Before the the parent is done there should be the initial value
due_time1 = SU.timestamp() + 3
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
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.5 seconds there still should be the initial value
time.sleep(0.5)
# After 0.2 seconds there still should be the initial value
time.sleep(0.2)
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)
# 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
......@@ -209,17 +156,109 @@ class TestEngineExecutor(unittest.TestCase):
# 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
# 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(4)
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:
# 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
#
# 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
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment