(self)
| 47 | self.test_heapq.check_invariant(heap) |
| 48 | |
| 49 | def test_racing_heappop(self): |
| 50 | heap = self.create_heap(OBJECT_COUNT, Heap.MIN) |
| 51 | |
| 52 | # Each thread pops (OBJECT_COUNT / NTHREADS) items |
| 53 | self.assertEqual(OBJECT_COUNT % NTHREADS, 0) |
| 54 | per_thread_pop_count = OBJECT_COUNT // NTHREADS |
| 55 | |
| 56 | def heappop_func(heap, pop_count): |
| 57 | local_list = [] |
| 58 | for _ in range(pop_count): |
| 59 | item = heapq.heappop(heap) |
| 60 | local_list.append(item) |
| 61 | |
| 62 | # Each local list should be sorted |
| 63 | self.assertTrue(self.is_sorted_ascending(local_list)) |
| 64 | |
| 65 | run_concurrently( |
| 66 | worker_func=heappop_func, |
| 67 | nthreads=NTHREADS, |
| 68 | args=(heap, per_thread_pop_count), |
| 69 | ) |
| 70 | self.assertEqual(len(heap), 0) |
| 71 | |
| 72 | def test_racing_heappushpop(self): |
| 73 | heap = self.create_heap(OBJECT_COUNT, Heap.MIN) |
nothing calls this directly
no test coverage detected