forked from retoor/devplacepy
fix: remove project_set_private from confirm list and fix async test helpers for event loop safety
This commit is contained in:
+21
-11
@@ -36,23 +36,33 @@ os.environ["DEVPLACE_SITEMAP_TTL"] = "0"
|
||||
|
||||
|
||||
def run_async(coro):
|
||||
box = {}
|
||||
"""Run a coroutine from synchronous test code, regardless of event loop state."""
|
||||
try:
|
||||
loop = asyncio.get_running_loop()
|
||||
except RuntimeError:
|
||||
return asyncio.run(coro)
|
||||
|
||||
def runner():
|
||||
result_box = []
|
||||
exc_box = []
|
||||
|
||||
def _run():
|
||||
try:
|
||||
box["value"] = asyncio.run(coro)
|
||||
except BaseException as exc:
|
||||
box["error"] = exc
|
||||
future = asyncio.run_coroutine_threadsafe(coro, loop)
|
||||
result_box.append(future.result())
|
||||
except BaseException as e:
|
||||
exc_box.append(e)
|
||||
|
||||
thread = threading.Thread(target=runner)
|
||||
thread.start()
|
||||
thread.join()
|
||||
t = threading.Thread(target=_run, daemon=True)
|
||||
t.start()
|
||||
t.join(timeout=30)
|
||||
from devplacepy.database import refresh_snapshot
|
||||
|
||||
refresh_snapshot()
|
||||
if "error" in box:
|
||||
raise box["error"]
|
||||
return box["value"]
|
||||
if exc_box:
|
||||
raise exc_box[0]
|
||||
if result_box:
|
||||
return result_box[0]
|
||||
raise TimeoutError("run_async timed out waiting for coroutine to complete")
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
|
||||
Reference in New Issue
Block a user