chiark - git - mdw - disorder/blob - tests/schedule.py

   1 #! /usr/bin/env python
   2 #
   3 # This file is part of DisOrder.
   4 # Copyright (C) 2008 Richard Kettlewell
   5 #
   6 # This program is free software; you can redistribute it and/or modify
   7 # it under the terms of the GNU General Public License as published by
   8 # the Free Software Foundation; either version 2 of the License, or
   9 # (at your option) any later version.
  10 #
  11 # This program is distributed in the hope that it will be useful, but
  12 # WITHOUT ANY WARRANTY; without even the implied warranty of
  13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14 # General Public License for more details.
  15 #
  16 # You should have received a copy of the GNU General Public License
  17 # along with this program; if not, write to the Free Software
  18 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
  19 # USA
  20 #
  21 import dtest,disorder,time,string
  22
  23 def now():
  24     """Return the current time in whole seconds"""
  25     return int(time.time())
  26
  27 def next_playing(c):
  28     print " waiting for track to play"
  29     p = c.playing()
  30     waited = 0
  31     while p is None and waited < 10:
  32         time.sleep(1)
  33         print "  ."
  34         p = c.playing()
  35     assert waited < 10, "track played in a reasonable time"
  36     return p
  37
  38 def wait_idle(c):
  39     print " waiting for nothing to be playing"
  40     p = c.playing()
  41     waited = 0
  42     while p is not None and waited < 20:
  43         time.sleep(1)
  44         print "  ."
  45         p = c.playing()
  46     assert waited < 20, "idled in a reasonable time"
  47
  48 def test():
  49     """Exercise schedule support"""
  50     dtest.start_daemon()
  51     dtest.create_user()
  52     c = disorder.client()
  53     c.random_disable()
  54     dtest.rescan()
  55     # Wait until there's no track playing
  56     print " waiting for nothing to be playing"
  57     while c.playing() is not None:
  58         time.sleep(1)
  59         print "  ."
  60     track = "%s/Joe Bloggs/First Album/05:Fifth track.ogg" % dtest.tracks
  61     print " scheduling a track for the future"
  62     when = now() + 3
  63     c.schedule_add(when, "normal", "play", track)
  64     print " disorder schedule-list output:"
  65     print string.join(dtest.command(["disorder",
  66                                      "--config", disorder._configfile,
  67                                      "--no-per-user-config",
  68                                      "schedule-list"]), ""),
  69     p = next_playing(c)
  70     assert p["track"] == track, "checking right track played"
  71     assert int(p["when"]) >= when, "checking track played at right time"
  72     assert c.schedule_list() == [], "checking schedule is empty"
  73     wait_idle(c)
  74     print " scheduling an enable-random for the future"
  75     c.schedule_add(now() + 3, "junk", "set-global", "random-play", "yes")
  76     print " disorder schedule-list output:"
  77     print string.join(dtest.command(["disorder",
  78                                      "--config", disorder._configfile,
  79                                      "--no-per-user-config",
  80                                      "schedule-list"]), ""),
  81     next_playing(c)
  82     print " disabling random play"
  83     c.random_disable()
  84     wait_idle(c)
  85     print " scheduling track to play later via command line"
  86     when = now() + 3
  87     dtest.command(["disorder",
  88                    "--config", disorder._configfile,
  89                    "--no-per-user-config",
  90                    "schedule-play",
  91                    time.strftime("%Y-%m-%d %H:%M:%S",
  92                                  time.localtime(when)),
  93                    "normal",
  94                    track])
  95     print " disorder schedule-list output:"
  96     print string.join(dtest.command(["disorder",
  97                                      "--config", disorder._configfile,
  98                                      "--no-per-user-config",
  99                                      "schedule-list"]), ""),
 100     p = next_playing(c)
 101     assert p["track"] == track, "checking right track played"
 102     assert p["when"] >= when, "checking track played at right time"
 103     assert c.schedule_list() == [], "checking schedule is empty"
 104     wait_idle(c)
 105     print " scheduling an enable-random for later via command line"
 106     dtest.command(["disorder",
 107                    "--config", disorder._configfile,
 108                    "--no-per-user-config",
 109                    "schedule-set-global",
 110                    time.strftime("%Y-%m-%d %H:%M:%S",
 111                                  time.localtime(now() + 3)),
 112                    "normal",
 113                    "random-play",
 114                    "yes"])
 115     print " disorder schedule-list output:"
 116     print string.join(dtest.command(["disorder",
 117                                      "--config", disorder._configfile,
 118                                      "--no-per-user-config",
 119                                      "schedule-list"]), ""),
 120     p = next_playing(c)
 121     print " disabling random play"
 122     c.random_disable()
 123     print " waiting for nothing to be playing"
 124     while c.playing() is not None:
 125         time.sleep(1)
 126         print "  ."
 127     print " scheduling a track for the future"
 128     c.schedule_add(now() + 3, "normal", "play", track)
 129     print " schedule via python:"
 130     s = c.schedule_list()
 131     for event in s:
 132         e = c.schedule_get(event)
 133         print "item %s: %s" % (event, e)
 134     print " deleting item %s" % s[0]
 135     c.schedule_del(s[0])
 136     print " checking it's really gone"
 137     s = c.schedule_list()
 138     assert s == [], "checking schedule is empty"
 139     waited = 0
 140     p = c.playing()
 141     while p is None and waited < 5:
 142         time.sleep(1)
 143         print "  ."
 144         waited += 1
 145         p = c.playing()
 146     assert p is None, "checking deleted scheduled event did not run"
 147     print " checking you can't schedule events for the past"
 148     try:
 149         c.schedule_add(now() - 4, "normal", "play", track)
 150         assert False, "checking schedule_add failed"
 151     except disorder.operationError:
 152       pass                              # good
 153     print " checking scheduled events survive restarts"
 154     when = now() + 3
 155     c.schedule_add(when, "normal", "play", track)
 156     dtest.stop_daemon()
 157     print " dumping database"
 158     dump = "%s/dumpfile" % dtest.testroot
 159     print dtest.command(["disorder-dump", "--config", disorder._configfile,
 160                          "--dump", dump])
 161     print "restoring database"
 162     print dtest.command(["disorder-dump", "--config", disorder._configfile,
 163                          "--undump", dump])
 164     dtest.start_daemon()
 165     c = disorder.client()
 166     p = next_playing(c)
 167     print " waiting for track to play"
 168     assert p["track"] == track, "checking right track played"
 169     assert p["when"] >= when, "checking track played at right time"
 170     assert c.schedule_list() == [], "checking schedule is empty"
 171     print " checking junk events do not survive restarts"
 172     c.schedule_add(now() + 2, "junk", "play", track)
 173     s = c.schedule_list()
 174     print s
 175     dtest.stop_daemon()
 176     time.sleep(3)
 177     dtest.start_daemon()
 178     c = disorder.client()
 179     print " checking schedule is empty"
 180     s = c.schedule_list()
 181     print s
 182     assert s == [], "checking schedule is empty"
 183
 184 if __name__ == '__main__':
 185     dtest.run()