source: server/test/server/job.py @ bd5bffc

no-cups
Last change on this file since bd5bffc was bd5bffc, checked in by Jessica B. Hamrick <jhamrick@…>, 12 years ago

Fix logger issues in job and player
  • Property mode set to 100644
File size: 6.2 KB
Line 
1from gutenbach.server import GutenbachJob
2from gutenbach.server import errors
3from gutenbach.ipp import JobStates as States
4import unittest
5import tempfile
6import sys
7import time
8import logging
9
10def make_tempfile():
11    fh = tempfile.NamedTemporaryFile()
12    fh.write("test\n")
13    fh.seek(0)
14    return fh
15
16class TestEmptyGutenbachJob(unittest.TestCase):
17
18    def setUp(self):
19        self.job = GutenbachJob()
20
21    def testPlayer(self):
22        self.assertEqual(self.job.player, None)
23    def testId(self):
24        self.assertEqual(self.job.id, -1)
25    def testCreator(self):
26        self.assertEqual(self.job.creator, "")
27    def testName(self):
28        self.assertEqual(self.job.name, "")
29    def testSize(self):
30        self.assertEqual(self.job.size, 0)
31    def testState(self):
32        self.assertEqual(self.job.state, States.HELD)
33    def testPriority(self):
34        self.assertEqual(self.job.priority, 1)
35
36    def testStateProperties(self):
37        self.assertFalse(self.job.is_valid)
38        self.assertFalse(self.job.is_ready)
39        self.assertFalse(self.job.is_playing)
40        self.assertFalse(self.job.is_paused)
41        self.assertFalse(self.job.is_done)
42        self.assertFalse(self.job.is_completed)
43        self.assertFalse(self.job.is_cancelled)
44        self.assertFalse(self.job.is_aborted)
45
46    def testSpool(self):
47        fh = tempfile.NamedTemporaryFile()
48        self.assertRaises(errors.InvalidJobStateException, self.job.spool, fh)
49        fh.close()
50    def testPlay(self):
51        self.assertRaises(errors.InvalidJobStateException, self.job.play)
52    def testPause(self):
53        self.assertRaises(errors.InvalidJobStateException, self.job.pause)
54    def testCancel(self):
55        self.job.cancel()
56        self.assertTrue(self.job.is_cancelled)
57        self.assertEqual(self.job.state, States.CANCELLED)
58    def testAbort(self):
59        self.job.abort()
60        self.assertTrue(self.job.is_aborted)
61        self.assertEqual(self.job.state, States.ABORTED)
62
63class TestBadGutenbachJob(unittest.TestCase):
64
65    def testBadJobId(self):
66        job = GutenbachJob(job_id=-2)
67        self.assertEqual(job.id, -1)
68        job.id = -2
69        self.assertEqual(job.id, -1)
70
71    def testBadCreator(self):
72        job = GutenbachJob(
73            job_id=1,
74            creator=12345)
75        self.assertEqual(job.creator, "12345")
76        job.creator = None
77        self.assertEqual(job.creator, "")
78        job.creator = []
79        self.assertEqual(job.creator, "[]")
80
81    def testBadName(self):
82        job = GutenbachJob(
83            job_id=1,
84            creator="foo",
85            name=12345)
86        self.assertEqual(job.name, "12345")
87        job.name = None
88        self.assertEqual(job.name, "")
89        job.name = []
90        self.assertEqual(job.name, "[]")
91
92    def testBadPriority(self):
93        job = GutenbachJob(
94            job_id=1,
95            creator="foo",
96            name="test",
97            priority=-1)
98        self.assertEqual(job.priority, 1)
99        job.priority = 0
100        self.assertEqual(job.priority, 1)
101        job.priority = 1
102        self.assertEqual(job.priority, 1)
103        job.priority = sys.maxint
104        self.assertEqual(job.priority, sys.maxint)
105        job.priority = "hello"
106        self.assertEqual(job.priority, 1)
107        job.priority = []
108        self.assertEqual(job.priority, 1)
109
110    def testBadDocument(self):
111        job = GutenbachJob(
112            job_id=1,
113            creator="foo",
114            name="test",
115            priority=1)
116        self.assertRaises(errors.InvalidDocument, job.spool, "hello")
117        self.assertRaises(errors.InvalidDocument, job.spool, [])
118        self.assertRaises(errors.InvalidDocument, job.spool, 12345)
119
120        fh = make_tempfile()
121        job = GutenbachJob(
122            job_id=1,
123            creator="foo",
124            name="test",
125            priority=1,
126            document=fh)
127        self.assertRaises(errors.InvalidJobStateException, job.spool, "hello")
128        self.assertRaises(errors.InvalidJobStateException, job.spool, [])
129        self.assertRaises(errors.InvalidJobStateException, job.spool, 12345)
130
131class TestOperations(unittest.TestCase):
132
133    def setUp(self):
134        fh = make_tempfile()
135        self.job = GutenbachJob(
136            job_id=1,
137            creator="foo",
138            name="test",
139            priority=1,
140            document=fh)
141        self.job.player._dryrun = True
142
143    def testPlay(self):
144        self.job.play()
145        self.assertTrue(self.job.is_playing)
146
147        while self.job.is_playing:
148            time.sleep(0.1)
149
150        self.assertTrue(self.job.is_done)
151        self.assertTrue(self.job.is_completed)
152        self.assertFalse(self.job.is_aborted)
153        self.assertFalse(self.job.is_cancelled)
154
155    def testPause(self):
156        self.job.play()
157        self.assertTrue(self.job.is_playing)
158        self.assertFalse(self.job.is_paused)
159
160        self.job.pause()
161        self.assertTrue(self.job.is_playing)
162        self.assertTrue(self.job.is_paused)
163
164        time.sleep(0.6)
165        self.assertTrue(self.job.is_playing)
166        self.assertTrue(self.job.is_paused)
167
168        self.job.pause()
169        self.assertTrue(self.job.is_playing)
170        self.assertFalse(self.job.is_paused)
171       
172        while self.job.is_playing:
173            time.sleep(0.1)
174           
175        self.assertTrue(self.job.is_done)
176        self.assertTrue(self.job.is_completed)
177        self.assertFalse(self.job.is_aborted)
178        self.assertFalse(self.job.is_cancelled)
179
180    def testCancel(self):
181        self.job.play()
182        self.assertTrue(self.job.is_playing)
183        self.assertFalse(self.job.is_cancelled)
184
185        self.job.cancel()
186        self.assertFalse(self.job.is_playing)
187        self.assertTrue(self.job.is_done)
188        self.assertTrue(self.job.is_cancelled)
189        self.assertFalse(self.job.is_aborted)
190
191    def testAbort(self):
192        self.job.play()
193        self.assertTrue(self.job.is_playing)
194        self.assertFalse(self.job.is_aborted)
195
196        self.job.abort()
197        self.assertFalse(self.job.is_playing)
198        self.assertTrue(self.job.is_done)
199        self.assertFalse(self.job.is_cancelled)
200        self.assertTrue(self.job.is_aborted)
201
202    def testRestart(self):
203        # XXX: Todo
204        pass
205
206if __name__ == "__main__":
207    logging.basicConfig(loglevel=logging.CRITICAL)
208    unittest.main()
Note: See TracBrowser for help on using the repository browser.