File size: 3,455 Bytes
7acbd80 | 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 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 | """Tests for plugins/plugin_utils.py — thread-safe lazy singleton helpers.
These exercise the actual concurrency guarantee with real threads (not mocks):
a barrier releases N threads simultaneously into the accessor, and we assert
the factory ran exactly once.
"""
import threading
import pytest
from plugins.plugin_utils import SingletonSlot, lazy_singleton
# --- lazy_singleton -------------------------------------------------------
def test_lazy_singleton_builds_once_and_returns_same_instance():
calls = []
@lazy_singleton
def get():
calls.append(1)
return object()
a = get()
b = get()
assert a is b
assert len(calls) == 1
def test_lazy_singleton_reset_rebuilds():
counter = {"n": 0}
@lazy_singleton
def get():
counter["n"] += 1
return counter["n"]
assert get() == 1
assert get() == 1
get.reset()
assert get() == 2
def test_lazy_singleton_concurrent_first_call_builds_once():
build_count = {"n": 0}
build_lock = threading.Lock()
barrier = threading.Barrier(16)
results = []
results_lock = threading.Lock()
@lazy_singleton
def get():
# Count builds under a lock so the assertion is exact even if the
# double-checked lock had a bug and let two through.
with build_lock:
build_count["n"] += 1
# Simulate an expensive build so threads genuinely overlap.
import time
time.sleep(0.01)
return object()
def worker():
barrier.wait() # release all threads at once
obj = get()
with results_lock:
results.append(obj)
threads = [threading.Thread(target=worker) for _ in range(16)]
for t in threads:
t.start()
for t in threads:
t.join()
assert build_count["n"] == 1, "factory must run exactly once under race"
assert len(results) == 16
assert all(r is results[0] for r in results), "all callers share one instance"
# --- SingletonSlot --------------------------------------------------------
def test_slot_caches_first_value():
slot: SingletonSlot = SingletonSlot()
assert slot.peek() is None
v1 = slot.get(lambda: "first")
assert slot.peek() == "first"
# Subsequent factory is ignored — first value wins.
v2 = slot.get(lambda: "second")
assert v1 == v2 == "first"
def test_slot_factory_exception_not_cached():
slot: SingletonSlot = SingletonSlot()
def boom():
raise ValueError("nope")
with pytest.raises(ValueError):
slot.get(boom)
assert slot.peek() is None
assert slot.get(lambda: "recovered") == "recovered"
def test_slot_concurrent_first_call_builds_once():
build_count = {"n": 0}
build_lock = threading.Lock()
barrier = threading.Barrier(16)
slot: SingletonSlot = SingletonSlot()
results = []
results_lock = threading.Lock()
def factory():
with build_lock:
build_count["n"] += 1
import time
time.sleep(0.01)
return object()
def worker():
barrier.wait()
obj = slot.get(factory)
with results_lock:
results.append(obj)
threads = [threading.Thread(target=worker) for _ in range(16)]
for t in threads:
t.start()
for t in threads:
t.join()
assert build_count["n"] == 1
assert len(results) == 16
assert all(r is results[0] for r in results)
|