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