systemd service/aura-engine.service

+[Unit]
+Description=Aura Engine Playout Server
+After=network.target
+
+[Service]
+Type=simple
+User=gg
+WorkingDirectory=/home/gg/PycharmProjects/engine
+ExecStart=/home/gg/PycharmProjects/engine/aura.py
+ExecStop=/home/gg/PycharmProjects/engine/guru.py --shutdown --quiet
+Restart=always
+
+[Install]
+WantedBy=multi-user.target

systemd service/aura-lqs.service

+[Unit]
+Description=Aura Engine Playout Server
+After=network.target aura-engine.service
+Wants=aura-engine.service
+
+[Service]
+Type=simple
+User=gg
+ExecStart=/usr/bin/liquidsoap /home/gg/PycharmProjects/engine/modules/liquidsoap/engine.liq
+Restart=always
+
+[Install]
+WantedBy=multi-user.target

test.py

+#!/usr/bin/python3
+
+from libraries.database.broadcasts import *
+import simplejson
+import logging
+import sqlalchemy
+
+from modules.communication.liquidsoap.communicator import LiquidSoapCommunicator
+from modules.monitoring.diskspace_watcher import DiskSpaceWatcher
+from libraries.base.config import AuraConfig
+
+def alchemyencoder(obj):
+    """JSON encoder function for SQLAlchemy special classes."""
+    if isinstance(obj, datetime.date):
+        return obj.isoformat()
+    elif isinstance(obj, decimal.Decimal):
+        return float(obj)
+    elif isinstance(obj, sqlalchemy.orm.state.InstanceState):
+        return ""
+    #elif isinstance(obj, Schedule):
+    #    return simplejson.dumps([obj._asdict()], default=alchemyencoder)
+    else:
+        return str(obj)
+
+def select_with_relationship():
+    se = TrackServiceScheduleEntry.select_all()
+    for e in se:
+        print(e._asdict())
+
+#    programme_as_string = simplejson.dumps([se[0]._asdict()], default=alchemyencoder)
+#    print(programme_as_string)
+
+def start_diskspace_watcher():
+    config = AuraConfig()
+    config.read_config()
+    diskspace_watcher = DiskSpaceWatcher(config.config, logging.getLogger("AuraEngine"), LiquidSoapCommunicator(config.config))
+    diskspace_watcher.run()
+
+# # ## ## ## ## ## # #
+# # ENTRY FUNCTION # #
+# # ## ## ## ## ## # #
+def main():
+    #select_with_relationship()
+    start_diskspace_watcher()
+# # ## ## ## ## ## ## # #
+# # End ENTRY FUNCTION # #
+# # ## ## ## ## ## ## # #
+
+
+if __name__ == "__main__":
+    main()
+

testing/test.py

+import os
+import unittest
+import validators
+
+from datetime import datetime
+
+# libraries.base
+from libraries.base.logger import AuraLogger
+from libraries.base.config import AuraConfig
+# libraries.database
+from libraries.database.broadcasts import Schedule, ScheduleEntry, TrackService
+# libraries.security
+from libraries.security.user import AuraUser
+
+# modules
+from modules.communication.liquidsoap.communicator import LiquidSoapCommunicator
+from modules.scheduling.scheduler import AuraScheduler
+
+class TestLogger(unittest.TestCase):
+    aura_logger = None
+
+    def setUp(self):
+        self.aura_logger = AuraLogger()
+
+    def test_logger(self):
+        self.assertTrue(self.aura_logger.logger.hasHandlers())
+
+
+class TestConfig(unittest.TestCase):
+    aura_config = None
+
+    def setUp(self):
+        self.aura_config = AuraConfig()
+
+    def test_config(self):
+        # is ini path correct set?
+        self.assertEqual(self.aura_config.config.ini_path, "/etc/aura/engine.ini")
+
+        # install_dir is set by runtime. is it a directory?
+        self.assertTrue(os.path.isdir(self.aura_config.config.get("install_dir")))
+
+        # calendarurl and importerurl set and valid urls?
+        self.assertTrue(validators.url(self.aura_config.config.get("calendarurl")))
+        self.assertTrue(validators.url(self.aura_config.config.get("importerurl")))
+
+        # is liquidsoap socketdir set and a directory?
+        self.assertTrue(os.path.isdir(self.aura_config.config.get("socketdir")))
+
+        # database settings set?
+        self.assertIsNotNone(self.aura_config.config.get("db_user"))
+        self.assertIsNotNone(self.aura_config.config.get("db_pass"))
+        self.assertIsNotNone(self.aura_config.config.get("db_name"))
+        self.assertIsNotNone(self.aura_config.config.get("db_host"))
+
+class TestSchedule(unittest.TestCase):
+    schedule = None
+
+    def setUp(self):
+        self.schedule = Schedule()
+
+    def test_schedule(self):
+        # select one and check if its not None and a Schedule
+        entry = self.schedule.select_by_id(1)
+        self.assertIsNotNone(entry)
+        self.assertIsInstance(entry, Schedule)
+
+
+class TestScheduleEntry(unittest.TestCase):
+    schedule_entry = None
+
+    def setUp(self):
+        self.schedule_entry = ScheduleEntry()
+
+    def test_schedule_entry(self):
+        # select one playlist and check if its not None, a ScheduleEntry
+        entry = self.schedule_entry.select_playlist(2)
+
+        self.assertIsNotNone(entry)
+        self.assertIsInstance(entry, list)
+        self.assertGreaterEqual(len(entry), 1)
+
+
+class TestTrackService(unittest.TestCase):
+    track_service = None
+
+    def setUp(self):
+        self.track_service = TrackService()
+
+    def test_track_service(self):
+        day = datetime.strptime("19.03.2018", "%d.%m.%Y")
+        entry = self.track_service.select_by_day(day)
+        self.assertIsNotNone(entry)
+        self.assertIsInstance(entry, list)
+
+
+class TestAuraUser(unittest.TestCase):
+    aura_user = None
+
+    def setUp(self):
+        self.aura_user = AuraUser()
+
+    def test_add_user(self):
+        username = "user"
+        password = "password"
+        role = "admin"
+
+        login_cnt = len(self.aura_user.getLogins())
+
+        # insert user
+        key = self.aura_user.insertUser(username, password, role)
+
+        self.assertGreaterEqual(len(self.aura_user.getLogins()), login_cnt)
+
+        # selecting user and check data
+        user = self.aura_user.getUserByKey(key)
+
+        self.assertEqual(user["username"], username)
+        # TODO: no encrypted storage.., but usermgm not really in use
+        self.assertEqual(user["password"], password)
+        self.assertEqual(user["role"], role)
+
+
+class TestLQSComm(unittest.TestCase):
+    comm = None
+
+    def setUp(self):
+        # wosn do passiert?
+        p = AuraConfig().config
+
+        self.comm = LiquidSoapCommunicator(p)
+        self.comm.scheduler = AuraScheduler(p)
+
+        self.comm.init_player()
+
+
+    def test_get_active_channel(self):
+        active_channel = self.comm.get_active_channel()
+
+        print(active_channel)
+
+
+
+if __name__ == '__main__':
+    unittest.main()
\ No newline at end of file