X-Git-Url: http://www.chiark.greenend.org.uk/ucgi/~mdw/git/disorder/blobdiff_plain/067eeb5f497179a9edbfe83bc7f1fb790dd2a9cc..0024bde0fcca14f018c647f41d8192b4564df96e:/tests/schedule.py diff --git a/tests/schedule.py b/tests/schedule.py index ac0b438..25db56d 100755 --- a/tests/schedule.py +++ b/tests/schedule.py @@ -1,25 +1,48 @@ #! /usr/bin/env python # # This file is part of DisOrder. -# Copyright (C) 2008 Richard Kettlewell +# Copyright (C) 2008, 2009 Richard Kettlewell # -# This program is free software; you can redistribute it and/or modify +# This program is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by -# the Free Software Foundation; either version 2 of the License, or +# 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 -# General Public License for more details. -# +# 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 General Public License for more details. +# # You should have received a copy of the GNU General Public License -# along with this program; if not, write to the Free Software -# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 -# USA +# along with this program. If not, see . # import dtest,disorder,time,string +def now(): + """Return the current time in whole seconds""" + return int(time.time()) + +def next_playing(c): + print " waiting for track to play" + p = c.playing() + waited = 0 + while p is None and waited < 10: + time.sleep(1) + print " ." + p = c.playing() + assert waited < 10, "track played in a reasonable time" + return p + +def wait_idle(c): + print " waiting for nothing to be playing" + p = c.playing() + waited = 0 + while p is not None and waited < 20: + time.sleep(1) + print " ." + p = c.playing() + assert waited < 20, "idled in a reasonable time" + def test(): """Exercise schedule support""" dtest.start_daemon() @@ -32,63 +55,40 @@ def test(): while c.playing() is not None: time.sleep(1) print " ." - now = int(time.time()) track = "%s/Joe Bloggs/First Album/05:Fifth track.ogg" % dtest.tracks print " scheduling a track for the future" - c.schedule_add(now + 4, "normal", "play", track) + when = now() + 3 + c.schedule_add(when, "normal", "play", track) print " disorder schedule-list output:" print string.join(dtest.command(["disorder", "--config", disorder._configfile, "--no-per-user-config", "schedule-list"]), ""), - print " waiting for it to play" - waited = 0 - p = c.playing() - while p is None and waited < 10: - time.sleep(1) - print " ." - waited += 1 - p = c.playing() - assert waited < 10, "checking track played within a reasonable period" - assert waited > 2, "checking track didn't play immediately" + p = next_playing(c) assert p["track"] == track, "checking right track played" + print " when=%d expected at least %d" % (int(p["when"]), when) + assert int(p["when"]) >= when, "checking track played at right time" assert c.schedule_list() == [], "checking schedule is empty" - print " waiting for nothing to be playing" - while c.playing() is not None: - time.sleep(1) - print " ." + wait_idle(c) print " scheduling an enable-random for the future" - now = int(time.time()) - c.schedule_add(now + 4, "normal", "set-global", "random-play", "yes") + c.schedule_add(now() + 3, "junk", "set-global", "random-play", "yes") print " disorder schedule-list output:" print string.join(dtest.command(["disorder", "--config", disorder._configfile, "--no-per-user-config", "schedule-list"]), ""), - print " waiting for it to take effect" - waited = 0 - p = c.playing() - while p is None and waited < 10: - time.sleep(1) - print " ." - waited += 1 - p = c.playing() - assert waited < 10, "checking a track played within a reasonable period" - assert waited > 2, "checking a track didn't play immediately" + next_playing(c) print " disabling random play" c.random_disable() - print " waiting for nothing to be playing" - while c.playing() is not None: - time.sleep(1) - print " ." + wait_idle(c) print " scheduling track to play later via command line" - now = int(time.time()) + when = now() + 3 dtest.command(["disorder", "--config", disorder._configfile, "--no-per-user-config", "schedule-play", time.strftime("%Y-%m-%d %H:%M:%S", - time.localtime(now + 4)), + time.localtime(when)), "normal", track]) print " disorder schedule-list output:" @@ -96,30 +96,18 @@ def test(): "--config", disorder._configfile, "--no-per-user-config", "schedule-list"]), ""), - print " waiting for it to play" - waited = 0 - p = c.playing() - while p is None and waited < 10: - time.sleep(1) - print " ." - waited += 1 - p = c.playing() - assert waited < 10, "checking track played within a reasonable period" - assert waited > 2, "checking track didn't play immediately" + p = next_playing(c) assert p["track"] == track, "checking right track played" + assert p["when"] >= when, "checking track played at right time" assert c.schedule_list() == [], "checking schedule is empty" - print " waiting for nothing to be playing" - while c.playing() is not None: - time.sleep(1) - print " ." + wait_idle(c) print " scheduling an enable-random for later via command line" - now = int(time.time()) dtest.command(["disorder", "--config", disorder._configfile, "--no-per-user-config", "schedule-set-global", time.strftime("%Y-%m-%d %H:%M:%S", - time.localtime(now + 4)), + time.localtime(now() + 3)), "normal", "random-play", "yes"]) @@ -128,16 +116,7 @@ def test(): "--config", disorder._configfile, "--no-per-user-config", "schedule-list"]), ""), - print " waiting for it to take effect" - waited = 0 - p = c.playing() - while p is None and waited < 10: - time.sleep(1) - print " ." - waited += 1 - p = c.playing() - assert waited < 10, "checking a track played within a reasonable period" - assert waited > 2, "checking a track didn't play immediately" + p = next_playing(c) print " disabling random play" c.random_disable() print " waiting for nothing to be playing" @@ -145,8 +124,7 @@ def test(): time.sleep(1) print " ." print " scheduling a track for the future" - now = int(time.time()) - c.schedule_add(now + 4, "normal", "play", track) + c.schedule_add(now() + 3, "normal", "play", track) print " schedule via python:" s = c.schedule_list() for event in s: @@ -159,7 +137,7 @@ def test(): assert s == [], "checking schedule is empty" waited = 0 p = c.playing() - while p is None and waited < 10: + while p is None and waited < 5: time.sleep(1) print " ." waited += 1 @@ -167,12 +145,40 @@ def test(): assert p is None, "checking deleted scheduled event did not run" print " checking you can't schedule events for the past" try: - now = int(time.time()) - c.schedule_add(now - 4, "normal", "play", track) + c.schedule_add(now() - 4, "normal", "play", track) assert False, "checking schedule_add failed" except disorder.operationError: pass # good - + print " checking scheduled events survive restarts" + when = now() + 3 + c.schedule_add(when, "normal", "play", track) + dtest.stop_daemon() + print " dumping database" + dump = "%s/dumpfile" % dtest.testroot + print dtest.command(["disorder-dump", "--config", disorder._configfile, + "--dump", dump]) + print "restoring database" + print dtest.command(["disorder-dump", "--config", disorder._configfile, + "--undump", dump]) + dtest.start_daemon() + c = disorder.client() + p = next_playing(c) + print " waiting for track to play" + assert p["track"] == track, "checking right track played" + assert p["when"] >= when, "checking track played at right time" + assert c.schedule_list() == [], "checking schedule is empty" + print " checking junk events do not survive restarts" + c.schedule_add(now() + 2, "junk", "play", track) + s = c.schedule_list() + print s + dtest.stop_daemon() + time.sleep(3) + dtest.start_daemon() + c = disorder.client() + print " checking schedule is empty" + s = c.schedule_list() + print s + assert s == [], "checking schedule is empty" if __name__ == '__main__': dtest.run()