source: server/lib/gutenbach_test/server/printer.py @ 5ffffe0

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

Reorganize test suite
  • Property mode set to 100644
File size: 8.8 KB
RevLine 
[5ffffe0]1__all__ = [
2    'TestEmptyGutenbachPrinter',
3    'TestBadEmptyGutenbachPrinter',
4    'TestGutenbachPrinter',
5    ]
6
[a2b0582]7from gutenbach.server import GutenbachJob
8from gutenbach.server import GutenbachPrinter
9from gutenbach.server import errors
10
[d42236e]11from gutenbach.ipp import PrinterStates as States
[a2b0582]12import unittest
13import tempfile
[d518298]14import time
[d42236e]15import logging
16import threading
[d518298]17
18def make_tempfile():
19    fh = tempfile.NamedTemporaryFile()
20    fh.write("test\n")
21    fh.seek(0)
22    return fh
[a2b0582]23
[d42236e]24class TestEmptyGutenbachPrinter(unittest.TestCase):
25
26    testJobName = "unique_test_job"
27    testJobRequestingUserName = "unique_test_username"
28    testPrinterConfig = {
29        'ipp-versions': ["1.0", "1.1"]
30        }
31
32    def setUp(self):
33        self.printer = GutenbachPrinter("test", self.testPrinterConfig)
34
35    def testName(self):
36        self.assertEqual(self.printer.name, "test")
37    def testConfig(self):
38        self.assertEqual(self.printer.config, self.testPrinterConfig)
39    def testTimeCreated(self):
40        self.assertTrue(self.printer.time_created is not None)
41    def testFinishedJobs(self):
42        self.assertEqual(self.printer.finished_jobs, [])
43    def testPendingJobs(self):
44        self.assertEqual(self.printer.pending_jobs, [])
45    def testCurrentJob(self):
46        self.assertEqual(self.printer.current_job, None)
47    def testJobs(self):
48        self.assertEqual(self.printer.jobs, {})
49    def testActiveJobs(self):
50        self.assertEqual(self.printer.active_jobs, [])
51    def testRunning(self):
52        self.assertEqual(self.printer.is_running, False)
53    def testPaused(self):
54        self.assertEqual(self.printer.paused, False)
55    def testNextJobId(self):
56        self.assertEqual(self.printer._next_job_id, 1)
57    def testState(self):
58        self.assertEqual(self.printer.state, States.STOPPED)
59
60    def testStartJob(self):
61        self.assertRaises(RuntimeError, self.printer.start_job)
62    def testCompleteJob(self):
63        self.assertRaises(RuntimeError, self.printer.complete_job)
64    def testGetJob(self):
65        self.assertRaises(RuntimeError, self.printer.get_job, 0)
66
67    def testIPPAttributes(self):
68        for attribute in self.printer.printer_attributes:
69            attr = attribute.replace("-", "_")
70            self.assertRaises(RuntimeError, getattr, self.printer, attr)
71            self.assertRaises(RuntimeError, setattr, self.printer, attr, None)
72        for attribute in self.printer.job_attributes:
73            attr = getattr(self.printer, attribute.replace("-", "_"))
74            self.assertRaises(RuntimeError, attr, 0)
75
76    def testIPPOperations(self):
77        self.assertRaises(RuntimeError, self.printer.print_job, None)
78        self.assertRaises(RuntimeError, self.printer.validate_job)
79        self.assertRaises(RuntimeError, self.printer.get_jobs)
80        self.assertRaises(RuntimeError, self.printer.print_uri)
81        self.assertRaises(RuntimeError, self.printer.create_job)
82        self.assertRaises(RuntimeError, self.printer.pause_printer)
83        self.assertRaises(RuntimeError, self.printer.resume_printer)
84        self.assertRaises(RuntimeError, self.printer.get_printer_attributes)
85        self.assertRaises(RuntimeError, self.printer.set_printer_attributes, {})
86        self.assertRaises(RuntimeError, self.printer.cancel_job, 0)
87        self.assertRaises(RuntimeError, self.printer.send_document, 0, None)
88        self.assertRaises(RuntimeError, self.printer.send_uri, 0, "")
89        self.assertRaises(RuntimeError, self.printer.get_job_attributes, 0)
90        self.assertRaises(RuntimeError, self.printer.set_job_attributes, 0, {})
91        self.assertRaises(RuntimeError, self.printer.restart_job, 0)
92        self.assertRaises(RuntimeError, self.printer.promote_job, 0)
93
94class TestBadEmptyGutenbachPrinter(unittest.TestCase):
95
96    testJobName = "unique_test_job"
97    testJobRequestingUserName = "unique_test_username"
98    testPrinterConfig = {
99        'ipp-versions': ["1.0", "1.1"]
100        }
101
102    def testBadPrinterName(self):
103        printer = GutenbachPrinter(None, self.testPrinterConfig)
104        self.assertEqual(printer.name, "None")
105        printer.name = "foo"
106        self.assertEqual(printer.name, "foo")
107        printer.name = 1234
108        self.assertEqual(printer.name, "1234")
109        printer.name = []
110        self.assertEqual(printer.name, "[]")
111
112    def testBadConfig(self):
113        self.assertRaises(ValueError, GutenbachPrinter, "test", {})
114        self.assertRaises(ValueError, GutenbachPrinter, "test", {"hello": 1234})
115        self.assertRaises(ValueError, GutenbachPrinter, "test", [])
116        self.assertRaises(ValueError, GutenbachPrinter, "test", 1234)
117        self.assertRaises(ValueError, GutenbachPrinter, "test", "hello")
118        conf = self.testPrinterConfig.copy()
119        printer = GutenbachPrinter("test", conf)
120        printer.config = conf.items()
121        self.assertEqual(printer.config, conf)
122        conf["hello"] = "goodbye"
123        self.assertNotEqual(printer.config, conf)
124        printer.config = conf
125        self.assertEqual(printer.config, conf)
126
[a2b0582]127class TestGutenbachPrinter(unittest.TestCase):
128
129    testJobName = "unique_test_job"
130    testJobRequestingUserName = "unique_test_username"
[d42236e]131    testPrinterConfig = {
132        'ipp-versions': ["1.0", "1.1"],
133        'dryrun': True
134        }
[a2b0582]135
136    def setUp(self):
[d42236e]137        self.printer = GutenbachPrinter("test", self.testPrinterConfig)
[a2b0582]138        self.printer.start()
[d42236e]139        while not self.printer.is_running:
140            time.sleep(0.01)
[a2b0582]141
142    def tearDown(self):
[d42236e]143        self.printer.stop()
144
145    def createTestJob(self):
146        job_id = self.printer.create_job(
147            self.testJobRequestingUserName,
148            self.testJobName)
149        return job_id
150
151    def testPrintJob(self):
152        pass
153
154    def testValidateJob(self):
155        pass
156
157    def testGetJobs(self):
158        pass
159
160    def testPrintUri(self):
161        pass
162
163    def testCreateJob(self):
164        for i in xrange(2):
165            countBeforeAdded = len(self.printer.jobs)
166            job = self.createTestJob()
167            self.assertEqual(len(self.printer.jobs) - 1, countBeforeAdded)
168            self.printer.send_document(job, make_tempfile())
169            self.assertTrue(job in self.printer.pending_jobs)
170            self.assertFalse(job in self.printer.finished_jobs)
171            self.assertEqual(self.printer.state, States.PROCESSING)
172
173        while len(self.printer.active_jobs) > 0:
174            time.sleep(0.01)
175        self.assertEqual(self.printer.state, States.IDLE)
[a2b0582]176
177    def testPausePrinter(self):
[d42236e]178        job = self.createTestJob()
179        self.printer.send_document(job, make_tempfile())
180        job = self.createTestJob()
181        self.printer.send_document(job, make_tempfile())
182        self.assertTrue(self.printer.is_running)
183        self.assertEqual(self.printer.state, States.PROCESSING)
184        self.assertFalse(self.printer.paused)
185
186        while self.printer.current_job is None:
187            time.sleep(0.01)
188
[a2b0582]189        self.printer.pause_printer()
[d42236e]190        self.assertTrue(self.printer.paused)
191        time.sleep(0.6)
192        self.assertTrue(self.printer.current_job is not None)
193        self.assertEqual(self.printer.state, States.STOPPED)
[a2b0582]194
195        # check that no jobs are playing
196        for job_id, job in self.printer.jobs.items():
[d42236e]197            self.assertEqual(job.is_paused, job.is_playing)
[a2b0582]198
[d42236e]199    def testResumePrinter(self):
200        job = self.createTestJob()
201        self.printer.send_document(job, make_tempfile())
202        job = self.createTestJob()
203        self.printer.send_document(job, make_tempfile())
204        self.assertTrue(self.printer.is_running)
205        self.assertEqual(self.printer.state, States.PROCESSING)
206        self.assertFalse(self.printer.paused)
[a2b0582]207
[d42236e]208        while self.printer.current_job is None:
209            time.sleep(0.01)
210
211        self.printer.pause_printer()
212        self.assertTrue(self.printer.paused)
213        time.sleep(0.6)
214        self.assertTrue(self.printer.current_job is not None)
215        self.assertEqual(self.printer.state, States.STOPPED)
216
217        # check that no jobs are playing
218        for job_id, job in self.printer.jobs.items():
219            self.assertEqual(job.is_paused, job.is_playing)
[a2b0582]220
221        self.printer.resume_printer()
[d42236e]222        self.assertFalse(self.printer.paused)
223        self.assertEqual(self.printer.state, States.PROCESSING)
224
225        while len(self.printer.active_jobs) > 0:
226            time.sleep(0.01)
227        self.assertEqual(self.printer.state, States.IDLE)
228
229    def testGetPrinterAttributes(self):
230        pass
231
232    def testSetPrinterAttributes(self):
233        pass
234
235    def testCancelJob(self):
236        pass
237
238    def testSendDocument(self):
239        pass
240
241    def testSendUri(self):
242        pass
243
244    def testGetJobAttributes(self):
245        pass
246
247    def testSetJobAttributes(self):
248        pass
249
250    def testRestartJob(self):
251        pass
[a2b0582]252
[d42236e]253    def testPromoteJob(self):
254        pass
[a2b0582]255
256if __name__ == "__main__":
[d42236e]257    logging.basicConfig(loglevel=logging.DEBUG)
[a2b0582]258    unittest.main()
Note: See TracBrowser for help on using the repository browser.