mirror of
https://github.com/RustPython/RustPython.git
synced 2026-10-29 03:04:38 +08:00
youknowdot
cd623a9216
* Prevent ContextVar cache hits after Context address reuse Give each Context, including copies, a non-reused cache identity instead of using its heap address. Disable cache hits if the identity counter saturates. This prevents an empty Context from reading a destroyed Context's cached value while preserving existing locking and drop order. Assisted-by: Codex:model-version-unavailable * Cover ContextVar reads after context destruction Exercise fresh and copied contexts with variable defaults, explicit defaults and missing values. The repeated lifecycle check complements canonical context tests without relying on a fixed allocator address-reuse count. Assisted-by: Codex:model-version-unavailable * Move ContextVar cache regression into the contextvars snippet Preserve the fresh and copied Context lifetime checks, defaults and missing-value assertions while removing the redundant Rust interpreter wrapper. Assisted-by: Codex:model-version-unavailable
119 lines
2.7 KiB
Python
119 lines
2.7 KiB
Python
from contextvars import Context, ContextVar, copy_context
|
|
|
|
from testutils import assert_raises
|
|
|
|
name = "\ud800"
|
|
var = ContextVar(name)
|
|
assert var.name == name
|
|
assert var.name is name
|
|
|
|
|
|
class StrSub(str):
|
|
pass
|
|
|
|
|
|
sub = StrSub("foo")
|
|
var_sub = ContextVar(sub)
|
|
assert var_sub.name is sub
|
|
|
|
assert_raises(TypeError, ContextVar, 1)
|
|
|
|
ctx1 = Context()
|
|
ctx2 = Context()
|
|
nested = ContextVar("var")
|
|
|
|
|
|
def func2():
|
|
assert nested.get(None) is None
|
|
|
|
|
|
def func1():
|
|
assert nested.get(None) is None
|
|
nested.set("spam")
|
|
ctx2.run(func2)
|
|
assert nested.get(None) == "spam"
|
|
cur = copy_context()
|
|
assert len(cur) == 1
|
|
assert cur[nested] == "spam"
|
|
return cur
|
|
|
|
|
|
returned_ctx = ctx1.run(func1)
|
|
assert ctx1 == returned_ctx
|
|
assert returned_ctx[nested] == "spam"
|
|
assert nested in returned_ctx
|
|
assert_raises(TypeError, hash, ctx1)
|
|
|
|
|
|
class ReentrantEq:
|
|
def __eq__(self, other):
|
|
ctx1.run(lambda: nested.set(object()))
|
|
return True
|
|
|
|
|
|
ctx1.run(nested.set, ReentrantEq())
|
|
ctx2.run(nested.set, object())
|
|
assert ctx1 == ctx2
|
|
|
|
try:
|
|
import gc
|
|
|
|
import _interpreters
|
|
except ImportError:
|
|
_interpreters = None
|
|
|
|
if _interpreters is not None:
|
|
child = _interpreters.create()
|
|
try:
|
|
assert (
|
|
_interpreters.run_string(
|
|
child,
|
|
"""
|
|
import contextvars
|
|
contextvars.ContextVar('child_marker').set(42)
|
|
""",
|
|
)
|
|
is None
|
|
)
|
|
finally:
|
|
_interpreters.destroy(child)
|
|
|
|
gc.collect()
|
|
assert [
|
|
(var.name, value)
|
|
for var, value in copy_context().items()
|
|
if var.name == "child_marker"
|
|
] == []
|
|
|
|
|
|
def context_cache_does_not_outlive_its_context():
|
|
import contextvars
|
|
|
|
for factory_name in ("new", "copy"):
|
|
for _ in range(100):
|
|
var = contextvars.ContextVar("var", default="default")
|
|
empty = contextvars.Context()
|
|
ctx = contextvars.Context() if factory_name == "new" else empty.copy()
|
|
ctx.run(var.set, "cached")
|
|
del ctx
|
|
ctx = contextvars.Context() if factory_name == "new" else empty.copy()
|
|
assert list(ctx.items()) == []
|
|
assert ctx.run(var.get) == "default"
|
|
assert ctx.run(var.get, "explicit default") == "explicit default"
|
|
|
|
missing = contextvars.ContextVar("missing")
|
|
ctx.run(missing.set, "cached")
|
|
del ctx
|
|
ctx = contextvars.Context() if factory_name == "new" else empty.copy()
|
|
try:
|
|
ctx.run(missing.get)
|
|
except LookupError:
|
|
pass
|
|
else:
|
|
raise AssertionError(
|
|
"an empty context returned a destroyed context value"
|
|
)
|
|
|
|
|
|
context_cache_does_not_outlive_its_context()
|