cpython/Lib/test/test_free_threading/test_itertools.py at 5ee09e19bd7bdcfd3b37719f9467f441f09e06ac · python/cpython · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
import unittest
from itertools import accumulate, batched, chain, combinations_with_replacement, cycle, islice, permutations, zip_longest
from test.support import threading_helper


threading_helper.requires_working_threading(module=True)


def work_iterator(it):
    while True:
        try:
            next(it)
        except StopIteration:
            break


class ItertoolsThreading(unittest.TestCase):

    @threading_helper.reap_threads
    def test_accumulate(self):
        number_of_iterations = 10
        for _ in range(number_of_iterations):
            it = accumulate(tuple(range(40)))
            threading_helper.run_concurrently(work_iterator, nthreads=10, args=[it])

    @threading_helper.reap_threads
    def test_batched(self):
        number_of_iterations = 10
        for _ in range(number_of_iterations):
            it = batched(tuple(range(1000)), 2)
            threading_helper.run_concurrently(work_iterator, nthreads=10, args=[it])

    @threading_helper.reap_threads
    def test_cycle(self):
        def work(it):
            for _ in range(400):
                next(it)

        number_of_iterations = 6
        for _ in range(number_of_iterations):
            it = cycle((1, 2, 3, 4))
            threading_helper.run_concurrently(work, nthreads=6, args=[it])

    @threading_helper.reap_threads
    def test_chain(self):
        number_of_iterations = 10
        for _ in range(number_of_iterations):
            it = chain(*[(1,)] * 200)
            threading_helper.run_concurrently(work_iterator, nthreads=6, args=[it])

    @threading_helper.reap_threads
    def test_combinations_with_replacement(self):
        number_of_iterations = 6
        for _ in range(number_of_iterations):
            it = combinations_with_replacement(tuple(range(2)), 2)
            threading_helper.run_concurrently(work_iterator, nthreads=6, args=[it])

    @threading_helper.reap_threads
    def test_islice(self):
        number_of_iterations = 6
        for _ in range(number_of_iterations):
            it = islice(tuple(range(10)), 1, 8, 2)
            threading_helper.run_concurrently(work_iterator, nthreads=10, args=[it])

    @threading_helper.reap_threads
    def test_permutations(self):
        number_of_iterations = 6
        for _ in range(number_of_iterations):
            it = permutations(tuple(range(4)), 2)
            threading_helper.run_concurrently(work_iterator, nthreads=6, args=[it])

    @threading_helper.reap_threads
    def test_zip_longest(self):
        number_of_iterations = 10
        for _ in range(number_of_iterations):
            it = zip_longest(list(range(4)), list(range(8)), fillvalue=0)
            threading_helper.run_concurrently(work_iterator, nthreads=10, args=[it])


if __name__ == "__main__":
    unittest.main()