Commit 5007ded0 authored by Gottfried Gaisbauer's avatar Gottfried Gaisbauer
Browse files

unittests

parent b4a70b06
......@@ -18,6 +18,9 @@ before_script:
simple_guru_help:
stage: test
script:
- whoami
- python3 guru.py -h
print_connection_status:
stage: test
script:
- python3 guru.py -pcs
\ No newline at end of file
#!/usr/bin/python3
import signal
import os
import sys
import signal
import unittest
from modules.scheduling.scheduler import AuraScheduler
from modules.communication.liquidsoap.communicator import LiquidSoapCommunicator
from modules.communication.redis.adapter import ServerRedisAdapter
from modules.web.routes import Routes
from libraries.base.config import AuraConfig
from libraries.base.logger import AuraLogger
class Aura(AuraConfig, AuraLogger):
from testing.test import TestConfig, TestLogger
class Aura(AuraLogger):
server = None
messenger = None
controller = None
......@@ -19,9 +22,19 @@ class Aura(AuraConfig, AuraLogger):
def __init__(self):
super(Aura, self).__init__()
def test_yourself(self):
self.run_test(TestConfig)
self.run_test(TestLogger)
def run_test(self, testingClass):
runner = unittest.TextTestRunner()
result = runner.run(unittest.makeSuite(testingClass))
def startup(self):
server = object
# self.controller = AuraController(self.config)
# self.controller = AuraController(self.config)
# create scheduler and ls_communicator
self.liquidsoapcommunicator = LiquidSoapCommunicator(self.config)
self.scheduler = AuraScheduler(self.config)
......@@ -42,6 +55,9 @@ class Aura(AuraConfig, AuraLogger):
signal.signal(signal.SIGUSR1, receive_signal)
self.join_comm()
self.start_web_service()
def join_comm(self):
# start listener thread
self.messenger.start()
......@@ -60,8 +76,13 @@ class Aura(AuraConfig, AuraLogger):
# # ## ## ## ## ## # #
def main():
aura = Aura()
aura.join_comm()
aura.start_web_service()
if len(sys.argv) >= 2 and sys.argv[1] == "--na-gspiast-di-?":
aura.test_yourself()
else:
aura.startup()
# # ## ## ## ## ## ## # #
# # End ENTRY FUNCTION # #
# # ## ## ## ## ## ## # #
......
......@@ -73,7 +73,6 @@ importerurl="http://localhost:8008/api/v1/groups/_public/playlists/"
audiobase="/var/audio/rec"
altaudiobase="/var/audio/preprod"
playlistdir="/var/audio/playlists/"
install_dir="/home/gg/PycharmProjects/engine"
[stream]
stream="y"
......
......@@ -12,6 +12,8 @@ class AuraConfig:
config = None
def __init__(self):
super(AuraConfig, self).__init__()
self.config = ConfigReader()
self.read_config()
def read_config(self):
......@@ -19,5 +21,4 @@ class AuraConfig:
reads aura.ini
:return:
"""
self.config = ConfigReader()
self.config.load_config()
......@@ -4,11 +4,14 @@ Aura logger functions
import logging
from libraries.base.config import AuraConfig
class AuraLogger:
class AuraLogger(AuraConfig):
logger = None
def __init__(self):
super(AuraLogger, self).__init__()
self.__create_logger("AuraEngine")
def __create_logger(self, name):
......
......@@ -86,6 +86,17 @@ class Schedule(DB.Model, AuraDatabaseModel):
show_fallback_id = Column(Integer)
station_fallback_id = Column(Integer)
@staticmethod
def select_all():
# fetching all entries
all_entries = DB.session.query(Schedule).filter().order_by(Schedule.schedule_start).all()
return all_entries
@staticmethod
def select_by_id(id):
entry = DB.session.query(Schedule).filter(Schedule.schedule_id == id).first()
return entry
def get_length(self):
sec1 = int(datetime.datetime.strptime(self.start[0:16].replace(" ", "T"), "%Y-%m-%dT%H:%M").strftime("%s"))
sec2 = int(datetime.datetime.strptime(self.end[0:16].replace(" ", "T"), "%Y-%m-%dT%H:%M").strftime("%s"))
......@@ -181,6 +192,11 @@ class ScheduleEntry(DB.Model, AuraDatabaseModel):
def select_one(playlist_id, entry_num):
return DB.session.query(ScheduleEntry).filter(ScheduleEntry.playlist_id == playlist_id, ScheduleEntry.entry_num == entry_num).first()
# ------------------------------------------------------------------------------------------ #
@staticmethod
def select_playlist(playlist_id):
return DB.session.query(ScheduleEntry).filter(ScheduleEntry.playlist_id == playlist_id).all()
# ------------------------------------------------------------------------------------------ #
def __str__(self):
return "ScheduleID: #" + str(self.schedule_id) + " Showname: " + self.schedule.show_name + " starts @ " + str(self.entry_start) + " and plays " + self.source
......@@ -206,9 +222,21 @@ class TrackService(DB.Model, AuraDatabaseModel):
schedule_entry = relationship("ScheduleEntry", primaryjoin="and_(TrackService.playlist_id==ScheduleEntry.playlist_id, TrackService.entry_num==ScheduleEntry.entry_num)", lazy="joined")
@staticmethod
# ------------------------------------------------------------------------------------------ #
def select_one(trackservice_id):
return DB.session.query(TrackService).filter(TrackService.trackservice_id == trackservice_id).first()
@staticmethod
# ------------------------------------------------------------------------------------------ #
def select_by_day(day):
day_plus_one = day + datetime.timedelta(days=1)
tracks = DB.session.query(TrackService).filter(TrackService.start >= str(day), TrackService.start < str(day_plus_one)).all()
return tracks
# ------------------------------------------------------------------------------------------ #
def __str__(self):
return "TrackServiceID: #" + str(self.trackservice_id) + " playlist_id: " + str(self.playlist_id) + " started @ " + str(self.start) + " and played " + self.source
# ------------------------------------------------------------------------------------------ #
# class TrackServiceSchedule(db.Model, AuraDatabaseModel):
# """
......
......@@ -6,8 +6,10 @@ class NoProgrammeLoadedException(Exception):
class LQConnectionError(Exception):
pass
class RedisConnectionException(Exception):
pass
class PlaylistException(Exception):
pass
......@@ -124,4 +124,6 @@ class ConfigReader(object):
v = config_parser.get(section, key).replace('"', '').strip()
self.set(key, v)
self.set("install_dir", os.path.realpath(__file__ + "../../../.."))
......@@ -7,3 +7,4 @@ redis==2.10.5
simplejson==3.11.1
mutagen==1.38
python-dateutil==2.6.0
validators==0.12.1
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
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)
if __name__ == '__main__':
unittest.main()
\ No newline at end of file
Supports Markdown
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