(self, step=1)
| 604 | |
| 605 | @threading_helper.requires_working_threading() |
| 606 | def test_count_threading(self, step=1): |
| 607 | # this test verifies multithreading consistency, which is |
| 608 | # mostly for testing builds without GIL, but nice to test anyway |
| 609 | count_to = 10_000 |
| 610 | num_threads = 10 |
| 611 | c = count(step=step) |
| 612 | def counting_thread(): |
| 613 | for i in range(count_to): |
| 614 | next(c) |
| 615 | threads = [] |
| 616 | for i in range(num_threads): |
| 617 | thread = threading.Thread(target=counting_thread) |
| 618 | thread.start() |
| 619 | threads.append(thread) |
| 620 | for thread in threads: |
| 621 | thread.join() |
| 622 | self.assertEqual(next(c), count_to * num_threads * step) |
| 623 | |
| 624 | def test_count_with_step_threading(self): |
| 625 | self.test_count_threading(step=5) |
no test coverage detected