CC-4905: Unit tests for Pypo

This commit is contained in:
Martin Konecny 2013-01-31 12:33:14 -05:00
parent 3553ebc2ee
commit 8adb4b259c
5 changed files with 67 additions and 5 deletions

View File

@ -176,7 +176,7 @@ if __name__ == '__main__':
sys.exit()
api_client = api_client.AirtimeApiClient()
ReplayGainUpdater.start_reply_gain(api_client)
api_client.register_component("pypo")

View File

@ -18,7 +18,8 @@ from std_err_override import LogWriter
from configobj import ConfigObj
# configure logging
logging.config.fileConfig("logging.cfg")
logging_cfg = os.path.join(os.path.dirname(__file__), "logging.cfg")
logging.config.fileConfig(logging_cfg)
logger = logging.getLogger()
LogWriter.override_std_err(logger)

View File

@ -9,6 +9,7 @@ import logging.config
import telnetlib
import calendar
import math
import os
from pypofetch import PypoFetch
from Queue import Empty
@ -21,7 +22,8 @@ from configobj import ConfigObj
# configure logging
logging.config.fileConfig("logging.cfg")
logging_cfg = os.path.join(os.path.dirname(__file__), "logging.cfg")
logging.config.fileConfig(logging_cfg)
logger = logging.getLogger()
LogWriter.override_std_err(logger)
@ -249,7 +251,7 @@ class PypoPush(Thread):
self.start_web_stream_buffer(current_item)
self.start_web_stream(current_item)
if is_file(current_item):
self.modify_cue_point(file_chain[0])
file_chain = self.modify_first_link_cue_point(file_chain)
self.push_to_liquidsoap(file_chain)
#we've changed the queue, so let's refetch it
liquidsoap_queue_approx = self.get_queue_items_from_liquidsoap()
@ -279,7 +281,7 @@ class PypoPush(Thread):
chain_to_push = file_chain[problem_at_iteration:]
if len(chain_to_push) > 0:
self.modify_cue_point(chain_to_push[0])
chain_to_push = self.modify_first_link_cue_point(chain_to_push)
self.push_to_liquidsoap(chain_to_push)
@ -363,6 +365,18 @@ class PypoPush(Thread):
original_cue_in_td = timedelta(seconds=float(link['cue_in']))
link['cue_in'] = self.date_interval_to_seconds(original_cue_in_td) + diff_sec
def modify_first_link_cue_point(self, chain):
if not len(chain):
return []
first_link = chain[0]
self.modify_cue_point(first_link)
if float(first_link['cue_in']) >= float(first_link['cue_out']):
chain = chain [1:]
return chain
"""
Returns two chains, original chain and current_chain. current_chain is a subset of
original_chain but can also be equal to original chain.

View File

@ -0,0 +1,18 @@
#!/bin/bash
which py.test
pytest_exist=$?
if [ "$pytest_exist" != "0" ]; then
echo "Need to have py.test installed. Exiting..."
exit 1
fi
SCRIPT=`readlink -f $0`
# Absolute directory this script is in
SCRIPTPATH=`dirname $SCRIPT`
export PYTHONPATH=$PYTHONPATH:$SCRIPTPATH/..:$SCRIPTPATH/../..
py.test

View File

@ -0,0 +1,29 @@
from pypopush import PypoPush
from threading import Lock
from Queue import Queue
import datetime
pypoPush_q = Queue()
telnet_lock = Lock()
pp = PypoPush(pypoPush_q, telnet_lock)
def test_modify_cue_in():
link = pp.modify_first_link_cue_point([])
assert len(link) == 0
min_ago = datetime.datetime.utcnow() - datetime.timedelta(minutes = 1)
link = [{"start":min_ago.strftime("%Y-%m-%d-%H-%M-%S"),
"cue_in":"0", "cue_out":"30"}]
link = pp.modify_first_link_cue_point(link)
assert len(link) == 0
link = [{"start":min_ago.strftime("%Y-%m-%d-%H-%M-%S"),
"cue_in":"0", "cue_out":"70"}]
link = pp.modify_first_link_cue_point(link)
assert len(link) == 1