mirror of
https://github.com/python/cpython
synced 2026-11-06 09:02:40 +03:00
29e9fe279a
Fix @reap_threads usage in test_free_threading.test_json (GH-158937)
Use the decorator on test methods, not on the whole class.
Previously, test_json ran no tests:
$ ./python -m test -v test_free_threading.test_json -v
...
Ran 0 tests in 0.000s
NO TESTS RAN
Add also a check in @reap_threads to detect misuse (on a class).
(cherry picked from commit 565f76fee1)
Co-authored-by: Victor Stinner <vstinner@python.org>
81 lines
2.3 KiB
Python
81 lines
2.3 KiB
Python
from threading import Barrier, Thread
|
|
from test.test_json import CTest
|
|
from test.support import threading_helper
|
|
|
|
|
|
def encode_json_helper(
|
|
json, worker, data, number_of_threads=12, number_of_json_encodings=100
|
|
):
|
|
worker_threads = []
|
|
barrier = Barrier(number_of_threads)
|
|
for index in range(number_of_threads):
|
|
worker_threads.append(
|
|
Thread(target=worker, args=[barrier, data, index])
|
|
)
|
|
for t in worker_threads:
|
|
t.start()
|
|
for ii in range(number_of_json_encodings):
|
|
json.dumps(data)
|
|
data.clear()
|
|
for t in worker_threads:
|
|
t.join()
|
|
|
|
|
|
class MyMapping(dict):
|
|
def __init__(self):
|
|
self.mapping = []
|
|
|
|
def items(self):
|
|
return self.mapping
|
|
|
|
|
|
@threading_helper.requires_working_threading()
|
|
class TestJsonEncoding(CTest):
|
|
# Test encoding json with concurrent threads modifying the data cannot
|
|
# corrupt the interpreter
|
|
|
|
@threading_helper.reap_threads
|
|
def test_json_mutating_list(self):
|
|
def worker(barrier, data, index):
|
|
barrier.wait()
|
|
while data:
|
|
for d in data:
|
|
if len(d) > 5:
|
|
d.clear()
|
|
else:
|
|
d.append(index)
|
|
|
|
data = [[], []]
|
|
encode_json_helper(self.json, worker, data)
|
|
|
|
@threading_helper.reap_threads
|
|
def test_json_mutating_exact_dict(self):
|
|
def worker(barrier, data, index):
|
|
barrier.wait()
|
|
while data:
|
|
for d in data:
|
|
if len(d) > 5:
|
|
try:
|
|
key = list(d)[0]
|
|
d.pop(key)
|
|
except (KeyError, IndexError):
|
|
pass
|
|
else:
|
|
d[index] = index
|
|
|
|
data = [{}, {}]
|
|
encode_json_helper(self.json, worker, data)
|
|
|
|
@threading_helper.reap_threads
|
|
def test_json_mutating_mapping(self):
|
|
def worker(barrier, data, index):
|
|
barrier.wait()
|
|
while data:
|
|
for d in data:
|
|
if len(d.mapping) > 3:
|
|
d.mapping.clear()
|
|
else:
|
|
d.mapping.append((index, index))
|
|
|
|
data = [MyMapping(), MyMapping()]
|
|
encode_json_helper(self.json, worker, data)
|