Back to index

python3.2  3.2.2
Public Member Functions | Public Attributes | Static Public Attributes
test.test_queue.LifoQueueTest Class Reference
Inheritance diagram for test.test_queue.LifoQueueTest:
Inheritance graph
[legend]
Collaboration diagram for test.test_queue.LifoQueueTest:
Collaboration graph
[legend]

List of all members.

Public Member Functions

def setUp
def simple_queue_test
def worker
def queue_join_test
def test_queue_task_done
def test_queue_join
def test_simple_queue
def test_negative_timeout_raises_exception
def test_nowait
def test_shrinking_queue
def do_blocking_test
def do_exceptional_blocking_test

Public Attributes

 cum
 cumlock
 t
 result

Static Public Attributes

 type2test = queue.LifoQueue

Detailed Description

Definition at line 235 of file test_queue.py.


Member Function Documentation

def test.test_queue.BlockingTestMixin.do_blocking_test (   self,
  block_func,
  block_args,
  trigger_func,
  trigger_args 
) [inherited]

Definition at line 49 of file test_queue.py.

00049 
00050     def do_blocking_test(self, block_func, block_args, trigger_func, trigger_args):
00051         self.t = _TriggerThread(trigger_func, trigger_args)
00052         self.t.start()
00053         self.result = block_func(*block_args)
00054         # If block_func returned before our thread made the call, we failed!
00055         if not self.t.startedEvent.is_set():
00056             self.fail("blocking function '%r' appeared not to block" %
00057                       block_func)
00058         self.t.join(10) # make sure the thread terminates
00059         if self.t.is_alive():
00060             self.fail("trigger function '%r' appeared to not return" %
00061                       trigger_func)
00062         return self.result

Here is the caller graph for this function:

def test.test_queue.BlockingTestMixin.do_exceptional_blocking_test (   self,
  block_func,
  block_args,
  trigger_func,
  trigger_args,
  expected_exception_class 
) [inherited]

Definition at line 65 of file test_queue.py.

00065 
00066                                    trigger_args, expected_exception_class):
00067         self.t = _TriggerThread(trigger_func, trigger_args)
00068         self.t.start()
00069         try:
00070             try:
00071                 block_func(*block_args)
00072             except expected_exception_class:
00073                 raise
00074             else:
00075                 self.fail("expected exception of kind %r" %
00076                                  expected_exception_class)
00077         finally:
00078             self.t.join(10) # make sure the thread terminates
00079             if self.t.is_alive():
00080                 self.fail("trigger function '%r' appeared to not return" %
00081                                  trigger_func)
00082             if not self.t.startedEvent.is_set():
00083                 self.fail("trigger thread ended but event never set")
00084 

Here is the call graph for this function:

Here is the caller graph for this function:

def test.test_queue.BaseQueueTest.queue_join_test (   self,
  q 
) [inherited]

Definition at line 157 of file test_queue.py.

00157 
00158     def queue_join_test(self, q):
00159         self.cum = 0
00160         for i in (0,1):
00161             threading.Thread(target=self.worker, args=(q,)).start()
00162         for i in range(100):
00163             q.put(i)
00164         q.join()
00165         self.assertEqual(self.cum, sum(range(100)),
00166                          "q.join() did not block until all tasks were done")
00167         for i in (0,1):
00168             q.put(-1)         # instruct the threads to close
00169         q.join()                # verify that you can join twice

Here is the call graph for this function:

Here is the caller graph for this function:

def test.test_queue.BaseQueueTest.setUp (   self) [inherited]

Definition at line 86 of file test_queue.py.

00086 
00087     def setUp(self):
00088         self.cum = 0
00089         self.cumlock = threading.Lock()

Here is the caller graph for this function:

def test.test_queue.BaseQueueTest.simple_queue_test (   self,
  q 
) [inherited]

Definition at line 90 of file test_queue.py.

00090 
00091     def simple_queue_test(self, q):
00092         if q.qsize():
00093             raise RuntimeError("Call this function with an empty queue")
00094         self.assertTrue(q.empty())
00095         self.assertFalse(q.full())
00096         # I guess we better check things actually queue correctly a little :)
00097         q.put(111)
00098         q.put(333)
00099         q.put(222)
00100         target_order = dict(Queue = [111, 333, 222],
00101                             LifoQueue = [222, 333, 111],
00102                             PriorityQueue = [111, 222, 333])
00103         actual_order = [q.get(), q.get(), q.get()]
00104         self.assertEqual(actual_order, target_order[q.__class__.__name__],
00105                          "Didn't seem to queue the correct data!")
00106         for i in range(QUEUE_SIZE-1):
00107             q.put(i)
00108             self.assertTrue(q.qsize(), "Queue should not be empty")
00109         self.assertTrue(not qfull(q), "Queue should not be full")
00110         last = 2 * QUEUE_SIZE
00111         full = 3 * 2 * QUEUE_SIZE
00112         q.put(last)
00113         self.assertTrue(qfull(q), "Queue should be full")
00114         self.assertFalse(q.empty())
00115         self.assertTrue(q.full())
00116         try:
00117             q.put(full, block=0)
00118             self.fail("Didn't appear to block with a full queue")
00119         except queue.Full:
00120             pass
00121         try:
00122             q.put(full, timeout=0.01)
00123             self.fail("Didn't appear to time-out with a full queue")
00124         except queue.Full:
00125             pass
00126         # Test a blocking put
00127         self.do_blocking_test(q.put, (full,), q.get, ())
00128         self.do_blocking_test(q.put, (full, True, 10), q.get, ())
00129         # Empty it
00130         for i in range(QUEUE_SIZE):
00131             q.get()
00132         self.assertTrue(not q.qsize(), "Queue should be empty")
00133         try:
00134             q.get(block=0)
00135             self.fail("Didn't appear to block with an empty queue")
00136         except queue.Empty:
00137             pass
00138         try:
00139             q.get(timeout=0.01)
00140             self.fail("Didn't appear to time-out with an empty queue")
00141         except queue.Empty:
00142             pass
00143         # Test a blocking get
00144         self.do_blocking_test(q.get, (), q.put, ('empty',))
00145         self.do_blocking_test(q.get, (True, 10), q.put, ('empty',))
00146 

Here is the call graph for this function:

Here is the caller graph for this function:

Definition at line 200 of file test_queue.py.

00200 
00201     def test_negative_timeout_raises_exception(self):
00202         q = self.type2test(QUEUE_SIZE)
00203         with self.assertRaises(ValueError):
00204             q.put(1, timeout=-1)
00205         with self.assertRaises(ValueError):
00206             q.get(1, timeout=-1)

Here is the call graph for this function:

def test.test_queue.BaseQueueTest.test_nowait (   self) [inherited]

Definition at line 207 of file test_queue.py.

00207 
00208     def test_nowait(self):
00209         q = self.type2test(QUEUE_SIZE)
00210         for i in range(QUEUE_SIZE):
00211             q.put_nowait(1)
00212         with self.assertRaises(queue.Full):
00213             q.put_nowait(1)
00214 
00215         for i in range(QUEUE_SIZE):
00216             q.get_nowait()
00217         with self.assertRaises(queue.Empty):
00218             q.get_nowait()

Here is the call graph for this function:

Definition at line 180 of file test_queue.py.

00180 
00181     def test_queue_join(self):
00182         # Test that a queue join()s successfully, and before anything else
00183         # (done twice for insurance).
00184         q = self.type2test()
00185         self.queue_join_test(q)
00186         self.queue_join_test(q)
00187         try:
00188             q.task_done()
00189         except ValueError:
00190             pass
00191         else:
00192             self.fail("Did not detect task count going negative")

Here is the call graph for this function:

Definition at line 170 of file test_queue.py.

00170 
00171     def test_queue_task_done(self):
00172         # Test to make sure a queue task completed successfully.
00173         q = self.type2test()
00174         try:
00175             q.task_done()
00176         except ValueError:
00177             pass
00178         else:
00179             self.fail("Did not detect task count going negative")

Here is the call graph for this function:

Definition at line 219 of file test_queue.py.

00219 
00220     def test_shrinking_queue(self):
00221         # issue 10110
00222         q = self.type2test(3)
00223         q.put(1)
00224         q.put(2)
00225         q.put(3)
00226         with self.assertRaises(queue.Full):
00227             q.put_nowait(4)
00228         self.assertEqual(q.qsize(), 3)
00229         q.maxsize = 2                       # shrink the queue
00230         with self.assertRaises(queue.Full):
00231             q.put_nowait(4)

Here is the call graph for this function:

Definition at line 193 of file test_queue.py.

00193 
00194     def test_simple_queue(self):
00195         # Do it a couple of times on the same queue.
00196         # Done twice to make sure works with same instance reused.
00197         q = self.type2test(QUEUE_SIZE)
00198         self.simple_queue_test(q)
00199         self.simple_queue_test(q)

Here is the call graph for this function:

def test.test_queue.BaseQueueTest.worker (   self,
  q 
) [inherited]

Definition at line 147 of file test_queue.py.

00147 
00148     def worker(self, q):
00149         while True:
00150             x = q.get()
00151             if x < 0:
00152                 q.task_done()
00153                 return
00154             with self.cumlock:
00155                 self.cum += x
00156             q.task_done()

Here is the caller graph for this function:


Member Data Documentation

Definition at line 87 of file test_queue.py.

Definition at line 88 of file test_queue.py.

Definition at line 52 of file test_queue.py.

Definition at line 50 of file test_queue.py.

Definition at line 236 of file test_queue.py.


The documentation for this class was generated from the following file: