152 lines
4.9 KiB
Python
152 lines
4.9 KiB
Python
from __future__ import annotations
|
|
|
|
import threading
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from deerflow.sandbox import sandbox_provider
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def isolated_provider(monkeypatch):
|
|
monkeypatch.setattr(sandbox_provider, "_default_sandbox_provider", None)
|
|
monkeypatch.setattr(sandbox_provider, "_sandbox_provider_lock", threading.RLock())
|
|
monkeypatch.setattr(
|
|
sandbox_provider,
|
|
"get_app_config",
|
|
lambda: SimpleNamespace(sandbox=SimpleNamespace(use="test-provider")),
|
|
)
|
|
|
|
|
|
def test_parallel_first_get_constructs_only_one_provider(monkeypatch):
|
|
entered = threading.Event()
|
|
release = threading.Event()
|
|
second_attempted = threading.Event()
|
|
duplicate_constructor = threading.Event()
|
|
constructed = []
|
|
|
|
class Provider:
|
|
def __init__(self, *, marker):
|
|
self.marker = marker
|
|
constructed.append(self)
|
|
if len(constructed) > 1:
|
|
duplicate_constructor.set()
|
|
entered.set()
|
|
assert release.wait(3), "test did not release provider construction"
|
|
|
|
monkeypatch.setattr(sandbox_provider, "resolve_class", lambda *_: Provider)
|
|
|
|
def second_get():
|
|
second_attempted.set()
|
|
return sandbox_provider.get_sandbox_provider(marker="second")
|
|
|
|
with ThreadPoolExecutor(max_workers=2) as executor:
|
|
first = executor.submit(sandbox_provider.get_sandbox_provider, marker="first")
|
|
try:
|
|
assert entered.wait(3)
|
|
second = executor.submit(second_get)
|
|
assert second_attempted.wait(3)
|
|
assert not duplicate_constructor.wait(0.1)
|
|
finally:
|
|
release.set()
|
|
provider = first.result(timeout=3)
|
|
assert second.result(timeout=3) is provider
|
|
|
|
assert constructed == [provider]
|
|
assert provider.marker == "first"
|
|
assert sandbox_provider.get_sandbox_provider(marker="later") is provider
|
|
|
|
|
|
def test_failed_constructor_can_be_retried(monkeypatch):
|
|
calls = 0
|
|
|
|
class Provider:
|
|
def __init__(self):
|
|
nonlocal calls
|
|
calls += 1
|
|
if calls == 1:
|
|
raise RuntimeError("temporary initialization failure")
|
|
|
|
monkeypatch.setattr(sandbox_provider, "resolve_class", lambda *_: Provider)
|
|
with pytest.raises(RuntimeError, match="temporary initialization failure"):
|
|
sandbox_provider.get_sandbox_provider()
|
|
|
|
assert sandbox_provider._default_sandbox_provider is None
|
|
provider = sandbox_provider.get_sandbox_provider()
|
|
assert sandbox_provider.get_sandbox_provider() is provider
|
|
assert calls == 2
|
|
|
|
|
|
@pytest.mark.parametrize("operation", ["reset", "set", "shutdown"])
|
|
def test_lifecycle_mutation_waits_for_construction(monkeypatch, operation):
|
|
entered = threading.Event()
|
|
release = threading.Event()
|
|
mutation_started = threading.Event()
|
|
mutation_finished = threading.Event()
|
|
replacement = object()
|
|
|
|
class Provider:
|
|
def __init__(self):
|
|
self.stopped = False
|
|
entered.set()
|
|
assert release.wait(3), "test did not release provider construction"
|
|
|
|
def shutdown(self):
|
|
self.stopped = True
|
|
|
|
monkeypatch.setattr(sandbox_provider, "resolve_class", lambda *_: Provider)
|
|
|
|
def mutate():
|
|
mutation_started.set()
|
|
if operation == "reset":
|
|
sandbox_provider.reset_sandbox_provider()
|
|
elif operation == "set":
|
|
sandbox_provider.set_sandbox_provider(replacement)
|
|
else:
|
|
sandbox_provider.shutdown_sandbox_provider()
|
|
mutation_finished.set()
|
|
|
|
with ThreadPoolExecutor(max_workers=2) as executor:
|
|
first = executor.submit(sandbox_provider.get_sandbox_provider)
|
|
try:
|
|
assert entered.wait(3)
|
|
mutation = executor.submit(mutate)
|
|
assert mutation_started.wait(3)
|
|
assert not mutation_finished.wait(0.1)
|
|
finally:
|
|
release.set()
|
|
provider = first.result(timeout=3)
|
|
mutation.result(timeout=3)
|
|
|
|
expected = replacement if operation == "set" else None
|
|
assert sandbox_provider._default_sandbox_provider is expected
|
|
assert provider.stopped is (operation == "shutdown")
|
|
|
|
|
|
def test_shutdown_callback_can_reenter_lifecycle_helpers():
|
|
finished = threading.Event()
|
|
observed = []
|
|
|
|
class Provider:
|
|
def shutdown(self):
|
|
observed.append(sandbox_provider.get_sandbox_provider())
|
|
sandbox_provider.reset_sandbox_provider()
|
|
|
|
provider = Provider()
|
|
sandbox_provider.set_sandbox_provider(provider)
|
|
|
|
def shutdown():
|
|
try:
|
|
sandbox_provider.shutdown_sandbox_provider()
|
|
finally:
|
|
finished.set()
|
|
|
|
worker = threading.Thread(target=shutdown, daemon=True)
|
|
worker.start()
|
|
assert finished.wait(3), "same-thread lifecycle callback deadlocked"
|
|
worker.join(timeout=3)
|
|
assert observed == [provider]
|
|
assert sandbox_provider._default_sandbox_provider is None
|