← Back to branch summary

~vcs-imports/quotient/main

« back to all changes in this revision

Viewing changes to quotient/test/test_sched.py

  • Committer: glyph
  • Date: 2003-10-26 23:44:25 UTC
  • Revision ID: Arch-1:unnamed@bazaar.ubuntu.com%series--4208--patch-749
whitespace

Show diffs side-by-side

added added

removed removed

Lines of Context:
quotient/test/test_sched.py
1
 
 
2
 
from os.path import join as opj
3
 
from twisted.trial import unittest
4
 
from quotient.sched import SystemScheduler, SubScheduler, SchedulerService, SchedulerBase, Schedulable
5
 
from quotient.items.powerup import IPowerStation
6
 
from quotient.storq import Store, SubStore, Item
7
 
 
8
 
class MySch(Item, Schedulable):
9
 
    count = 0
10
 
    def run(self):
11
 
        self.count += 1
12
 
 
13
 
class SchedTest(unittest.TestCase):
14
 
 
15
 
    def setUp(self):
16
 
        self.times = [0.0, 1.0, 2.0, 3.0, 10.0, 20.0, 30.0]
17
 
        self.calls = []
18
 
        SchedulerBase.clock = staticmethod(lambda : self.times[0])
19
 
        SystemScheduler.callLater = staticmethod(lambda t, f, *a, **b: self.calls.append((t, f)))
20
 
        dbpath = self.mktemp()
21
 
        self.store = Store(opj(dbpath, "db"), opj(dbpath, "files"))
22
 
        def _():
23
 
            self.svc = SchedulerService(self.store)
24
 
            self.getTopSched = self.svc.ref.getItem
25
 
            self.sub = SubStore(self.store)
26
 
            IPowerStation(self.sub).installPowerup(SubScheduler)
27
 
            self.svc.startService()
28
 
        self.store.transact(_)
29
 
 
30
 
    def testTopScheduler(self):
31
 
        def _():
32
 
            msc = MySch(self.store)
33
 
            msc.schedule(2.5)
34
 
            for x in range(10):
35
 
                t, f = self.calls.pop(0)
36
 
                f()
37
 
            self.assertEquals(msc.count, 0)
38
 
            for x in range(3):
39
 
                self.times.pop(0)
40
 
                t, f = self.calls.pop(0)
41
 
                f()
42
 
            self.assertEquals(msc.count, 1)
43
 
            self.assertEquals(self.calls, [])
44
 
        self.store.transact(_)
45
 
 
46
 
    def testSubScheduler(self):
47
 
        def _():
48
 
            msc = MySch(self.sub)
49
 
            msc.schedule(2.5)
50
 
            for x in range(10):
51
 
                t, f = self.calls.pop(0)
52
 
                f()
53
 
            self.assertEquals(msc.count, 0)
54
 
            for x in range(3):
55
 
                self.times.pop(0)
56
 
                t, f = self.calls.pop(0)
57
 
                f()
58
 
            self.assertEquals(msc.count, 1)
59
 
            self.assertEquals(self.calls, [])
60
 
        self.store.transact(_)