test(core): harden concurrent-mutation test for safe_json_dumps

- Loop 50 iterations to increase race trigger probability
- Catch all exceptions (not just RuntimeError) to avoid silent swallowing
- Assert result is a non-empty string (correctness check)
This commit is contained in:
yryzhan 2026-05-20 15:51:06 +02:00
parent 3ed3b861fa
commit 4980905fee

View file

@ -177,28 +177,35 @@ def test_pydantic_base_model():
def test_no_runtime_error_on_concurrent_dict_mutation():
"""safe_dumps must not raise RuntimeError when another thread mutates the dict."""
data = {f"key_{i}": f"value_{i}" for i in range(200)}
barrier = threading.Barrier(2, timeout=5)
errors = []
errors: list = []
def mutator():
barrier.wait()
for i in range(200, 400):
data[f"key_{i}"] = f"value_{i}"
for _ in range(50):
data = {f"key_{i}": f"value_{i}" for i in range(200)}
barrier = threading.Barrier(2, timeout=5)
results: list = []
def serializer():
barrier.wait()
try:
safe_dumps(data)
except RuntimeError as e:
if "dictionary changed size during iteration" in str(e):
def mutator():
barrier.wait()
for i in range(200, 400):
data[f"key_{i}"] = f"value_{i}"
def serializer():
barrier.wait()
try:
result = safe_dumps(data)
results.append(result)
except Exception as e:
errors.append(e)
t1 = threading.Thread(target=mutator)
t2 = threading.Thread(target=serializer)
t1.start()
t2.start()
t1.join(timeout=5)
t2.join(timeout=5)
t1 = threading.Thread(target=mutator)
t2 = threading.Thread(target=serializer)
t1.start()
t2.start()
t1.join(timeout=5)
t2.join(timeout=5)
assert not errors, "safe_dumps raised RuntimeError on concurrent mutation"
if results:
assert isinstance(results[0], str)
assert len(results[0]) > 0
assert not errors, f"safe_dumps raised {errors[0]!r} on concurrent mutation"