Changeset d42236e for server/test/server/printer.py

Timestamp:

Jan 21, 2012, 11:14:07 PM (12 years ago)

Author:

Jessica B. Hamrick <jhamrick@…>

Branches:

no-cups

Children:

75928fe

Parents:

3cb6c7b

git-author:

Jessica B. Hamrick <jhamrick@…> (01/21/12 23:12:03)

git-committer:

Jessica B. Hamrick <jhamrick@…> (01/21/12 23:14:07)

Message:

Fix a few more bugs in printer.py and update test cases for printer.py

File:

• server/test/server/printer.py (modified) (2 diffs)

server/test/server/printer.py

@@ -3 +3 @@
 from gutenbach.server import errors
 
-from gutenbach.ipp import JobStates as States
+from gutenbach.ipp import PrinterStates as States
 import unittest
 import tempfile
 import time
+import logging
+import threading
 
 def make_tempfile():
@@ -14 +16 @@
     return fh
 
-class TestGutenbachPrinter(unittest.TestCase):
+class TestEmptyGutenbachPrinter(unittest.TestCase):
 
     testJobName = "unique_test_job"
     testJobRequestingUserName = "unique_test_username"
+    testPrinterConfig = {
+        'ipp-versions': ["1.0", "1.1"]
+        }
 
     def setUp(self):
-        self.printer = GutenbachPrinter("test")
+        self.printer = GutenbachPrinter("test", self.testPrinterConfig)
+
+    def testName(self):
+        self.assertEqual(self.printer.name, "test")
+    def testConfig(self):
+        self.assertEqual(self.printer.config, self.testPrinterConfig)
+    def testTimeCreated(self):
+        self.assertTrue(self.printer.time_created is not None)
+    def testFinishedJobs(self):
+        self.assertEqual(self.printer.finished_jobs, [])
+    def testPendingJobs(self):
+        self.assertEqual(self.printer.pending_jobs, [])
+    def testCurrentJob(self):
+        self.assertEqual(self.printer.current_job, None)
+    def testJobs(self):
+        self.assertEqual(self.printer.jobs, {})
+    def testActiveJobs(self):
+        self.assertEqual(self.printer.active_jobs, [])
+    def testRunning(self):
+        self.assertEqual(self.printer.is_running, False)
+    def testPaused(self):
+        self.assertEqual(self.printer.paused, False)
+    def testNextJobId(self):
+        self.assertEqual(self.printer._next_job_id, 1)
+    def testState(self):
+        self.assertEqual(self.printer.state, States.STOPPED)
+
+    def testStartJob(self):
+        self.assertRaises(RuntimeError, self.printer.start_job)
+    def testCompleteJob(self):
+        self.assertRaises(RuntimeError, self.printer.complete_job)
+    def testGetJob(self):
+        self.assertRaises(RuntimeError, self.printer.get_job, 0)
+
+    def testIPPAttributes(self):
+        for attribute in self.printer.printer_attributes:
+            attr = attribute.replace("-", "_")
+            self.assertRaises(RuntimeError, getattr, self.printer, attr)
+            self.assertRaises(RuntimeError, setattr, self.printer, attr, None)
+        for attribute in self.printer.job_attributes:
+            attr = getattr(self.printer, attribute.replace("-", "_"))
+            self.assertRaises(RuntimeError, attr, 0)
+
+    def testIPPOperations(self):
+        self.assertRaises(RuntimeError, self.printer.print_job, None)
+        self.assertRaises(RuntimeError, self.printer.validate_job)
+        self.assertRaises(RuntimeError, self.printer.get_jobs)
+        self.assertRaises(RuntimeError, self.printer.print_uri)
+        self.assertRaises(RuntimeError, self.printer.create_job)
+        self.assertRaises(RuntimeError, self.printer.pause_printer)
+        self.assertRaises(RuntimeError, self.printer.resume_printer)
+        self.assertRaises(RuntimeError, self.printer.get_printer_attributes)
+        self.assertRaises(RuntimeError, self.printer.set_printer_attributes, {})
+        self.assertRaises(RuntimeError, self.printer.cancel_job, 0)
+        self.assertRaises(RuntimeError, self.printer.send_document, 0, None)
+        self.assertRaises(RuntimeError, self.printer.send_uri, 0, "")
+        self.assertRaises(RuntimeError, self.printer.get_job_attributes, 0)
+        self.assertRaises(RuntimeError, self.printer.set_job_attributes, 0, {})
+        self.assertRaises(RuntimeError, self.printer.restart_job, 0)
+        self.assertRaises(RuntimeError, self.printer.promote_job, 0)
+
+class TestBadEmptyGutenbachPrinter(unittest.TestCase):
+
+    testJobName = "unique_test_job"
+    testJobRequestingUserName = "unique_test_username"
+    testPrinterConfig = {
+        'ipp-versions': ["1.0", "1.1"]
+        }
+
+    def testBadPrinterName(self):
+        printer = GutenbachPrinter(None, self.testPrinterConfig)
+        self.assertEqual(printer.name, "None")
+        printer.name = "foo"
+        self.assertEqual(printer.name, "foo")
+        printer.name = 1234
+        self.assertEqual(printer.name, "1234")
+        printer.name = []
+        self.assertEqual(printer.name, "[]")
+
+    def testBadConfig(self):
+        self.assertRaises(ValueError, GutenbachPrinter, "test", {})
+        self.assertRaises(ValueError, GutenbachPrinter, "test", {"hello": 1234})
+        self.assertRaises(ValueError, GutenbachPrinter, "test", [])
+        self.assertRaises(ValueError, GutenbachPrinter, "test", 1234)
+        self.assertRaises(ValueError, GutenbachPrinter, "test", "hello")
+        conf = self.testPrinterConfig.copy()
+        printer = GutenbachPrinter("test", conf)
+        printer.config = conf.items()
+        self.assertEqual(printer.config, conf)
+        conf["hello"] = "goodbye"
+        self.assertNotEqual(printer.config, conf)
+        printer.config = conf
+        self.assertEqual(printer.config, conf)
+
+class TestGutenbachPrinter(unittest.TestCase):
+
+    testJobName = "unique_test_job"
+    testJobRequestingUserName = "unique_test_username"
+    testPrinterConfig = {
+        'ipp-versions': ["1.0", "1.1"],
+        'dryrun': True
+        }
+
+    def setUp(self):
+        self.printer = GutenbachPrinter("test", self.testPrinterConfig)
         self.printer.start()
+        while not self.printer.is_running:
+            time.sleep(0.01)
 
     def tearDown(self):
-        self.printer.running = False
-        self.printer.join()
-
-    # nb: this test assumes that pause_printer will block until it has paused.
+        self.printer.stop()
+
+    def createTestJob(self):
+        job_id = self.printer.create_job(
+            self.testJobRequestingUserName,
+            self.testJobName)
+        return job_id
+
+    def testPrintJob(self):
+        pass
+
+    def testValidateJob(self):
+        pass
+
+    def testGetJobs(self):
+        pass
+
+    def testPrintUri(self):
+        pass
+
+    def testCreateJob(self):
+        for i in xrange(2):
+            countBeforeAdded = len(self.printer.jobs)
+            job = self.createTestJob()
+            self.assertEqual(len(self.printer.jobs) - 1, countBeforeAdded)
+            self.printer.send_document(job, make_tempfile())
+            self.assertTrue(job in self.printer.pending_jobs)
+            self.assertFalse(job in self.printer.finished_jobs)
+            self.assertEqual(self.printer.state, States.PROCESSING)
+
+        while len(self.printer.active_jobs) > 0:
+            time.sleep(0.01)
+        self.assertEqual(self.printer.state, States.IDLE)
+
     def testPausePrinter(self):
+        job = self.createTestJob()
+        self.printer.send_document(job, make_tempfile())
+        job = self.createTestJob()
+        self.printer.send_document(job, make_tempfile())
+        self.assertTrue(self.printer.is_running)
+        self.assertEqual(self.printer.state, States.PROCESSING)
+        self.assertFalse(self.printer.paused)
+
+        while self.printer.current_job is None:
+            time.sleep(0.01)
+
         self.printer.pause_printer()
-        self.assertEqual(self.printer.paused, True)
+        self.assertTrue(self.printer.paused)
+        time.sleep(0.6)
+        self.assertTrue(self.printer.current_job is not None)
+        self.assertEqual(self.printer.state, States.STOPPED)
 
         # check that no jobs are playing
         for job_id, job in self.printer.jobs.items():
-            self.assertEqual(job.is_playing(), False)
-
-    def createTestJob(self):
-        return self.printer.create_job(self.testJobRequestingUserName, self.testJobName)
-  
-    def testCreateJob(self):
-        countBeforeAdded = len(self.printer.jobs)
-        job_id = self.createTestJob()
-        self.assertEqual(len(self.printer.jobs) - 1, countBeforeAdded)
-
-        queued_job = self.printer.get_job(job_id)
-        b = queued_job is not None
-        self.assertTrue(b)
-        self.assertEqual(queued_job.name, self.testJobName)
-        self.assertEqual(queued_job.creator, self.testJobRequestingUserName)
-        self.assertEqual(queued_job.state, States.HELD)
+            self.assertEqual(job.is_paused, job.is_playing)
 
     def testResumePrinter(self):
+        job = self.createTestJob()
+        self.printer.send_document(job, make_tempfile())
+        job = self.createTestJob()
+        self.printer.send_document(job, make_tempfile())
+        self.assertTrue(self.printer.is_running)
+        self.assertEqual(self.printer.state, States.PROCESSING)
+        self.assertFalse(self.printer.paused)
+
+        while self.printer.current_job is None:
+            time.sleep(0.01)
+
+        self.printer.pause_printer()
+        self.assertTrue(self.printer.paused)
+        time.sleep(0.6)
+        self.assertTrue(self.printer.current_job is not None)
+        self.assertEqual(self.printer.state, States.STOPPED)
+
+        # check that no jobs are playing
+        for job_id, job in self.printer.jobs.items():
+            self.assertEqual(job.is_paused, job.is_playing)
+
         self.printer.resume_printer()
-        self.assertEqual(self.printer.paused, False)
-
-    def testJobPlays(self):
-        job_id = self.createTestJob()
-        fh = make_tempfile()
-        self.printer.get_job(job_id).spool(fh)
-        self.assertTrue(job_id in self.printer.pending_jobs)
-        self.assertFalse(job_id in self.printer.finished_jobs)
-        self.printer.complete_job()
-        while (self.printer.get_job(job_id).state == States.PENDING):
-            time.sleep(0.1)
-            continue
-        
-        non_pending_job = self.printer.get_job(job_id)
-        self.assertTrue(non_pending_job.state == States.COMPLETE)
-
-        self.assertFalse(job_id in self.printer.pending_jobs)
-        self.assertTrue(job_id in self.printer.finished_jobs)
+        self.assertFalse(self.printer.paused)
+        self.assertEqual(self.printer.state, States.PROCESSING)
+
+        while len(self.printer.active_jobs) > 0:
+            time.sleep(0.01)
+        self.assertEqual(self.printer.state, States.IDLE)
+
+    def testGetPrinterAttributes(self):
+        pass
+
+    def testSetPrinterAttributes(self):
+        pass
+
+    def testCancelJob(self):
+        pass
+
+    def testSendDocument(self):
+        pass
+
+    def testSendUri(self):
+        pass
+
+    def testGetJobAttributes(self):
+        pass
+
+    def testSetJobAttributes(self):
+        pass
+
+    def testRestartJob(self):
+        pass
+
+    def testPromoteJob(self):
+        pass
 
 if __name__ == "__main__":
+    logging.basicConfig(loglevel=logging.DEBUG)
     unittest.main()