(self, q)
| 731 | super().setUp() |
| 732 | |
| 733 | def failing_queue_test(self, q): |
| 734 | if q.qsize(): |
| 735 | raise RuntimeError("Call this function with an empty queue") |
| 736 | for i in range(QUEUE_SIZE-1): |
| 737 | q.put(i) |
| 738 | # Test a failing non-blocking put. |
| 739 | q.fail_next_put = True |
| 740 | try: |
| 741 | q.put("oops", block=0) |
| 742 | self.fail("The queue didn't fail when it should have") |
| 743 | except FailingQueueException: |
| 744 | pass |
| 745 | q.fail_next_put = True |
| 746 | try: |
| 747 | q.put("oops", timeout=0.1) |
| 748 | self.fail("The queue didn't fail when it should have") |
| 749 | except FailingQueueException: |
| 750 | pass |
| 751 | q.put("last") |
| 752 | self.assertTrue(qfull(q), "Queue should be full") |
| 753 | # Test a failing blocking put |
| 754 | q.fail_next_put = True |
| 755 | try: |
| 756 | self.do_blocking_test(q.put, ("full",), q.get, ()) |
| 757 | self.fail("The queue didn't fail when it should have") |
| 758 | except FailingQueueException: |
| 759 | pass |
| 760 | # Check the Queue isn't damaged. |
| 761 | # put failed, but get succeeded - re-add |
| 762 | q.put("last") |
| 763 | # Test a failing timeout put |
| 764 | q.fail_next_put = True |
| 765 | try: |
| 766 | self.do_exceptional_blocking_test(q.put, ("full", True, 10), q.get, (), |
| 767 | FailingQueueException) |
| 768 | self.fail("The queue didn't fail when it should have") |
| 769 | except FailingQueueException: |
| 770 | pass |
| 771 | # Check the Queue isn't damaged. |
| 772 | # put failed, but get succeeded - re-add |
| 773 | q.put("last") |
| 774 | self.assertTrue(qfull(q), "Queue should be full") |
| 775 | q.get() |
| 776 | self.assertTrue(not qfull(q), "Queue should not be full") |
| 777 | q.put("last") |
| 778 | self.assertTrue(qfull(q), "Queue should be full") |
| 779 | # Test a blocking put |
| 780 | self.do_blocking_test(q.put, ("full",), q.get, ()) |
| 781 | # Empty it |
| 782 | for i in range(QUEUE_SIZE): |
| 783 | q.get() |
| 784 | self.assertTrue(not q.qsize(), "Queue should be empty") |
| 785 | q.put("first") |
| 786 | q.fail_next_get = True |
| 787 | try: |
| 788 | q.get() |
| 789 | self.fail("The queue didn't fail when it should have") |
| 790 | except FailingQueueException: |
no test coverage detected