mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-10-30 12:25:29 +08:00
103 lines
3.7 KiB
Python
103 lines
3.7 KiB
Python
"""Failure reproductions and bounded subprocess transport tests."""
|
|
|
|
import asyncio
|
|
import importlib
|
|
import os
|
|
import threading
|
|
|
|
import pytest
|
|
from test_isolation import load
|
|
|
|
|
|
@pytest.mark.parametrize("name", ["dashscope", "tbox"])
|
|
def test_cancel_hung_vendor_does_not_leave_threads(name):
|
|
with load(name):
|
|
module = importlib.import_module("pkg." + name + "_client")
|
|
# RED on the old implementation. New transport removes this unsafe API.
|
|
if not hasattr(module, "_iterate_sync_in_thread"):
|
|
assert hasattr(module, "vendor_stream")
|
|
return
|
|
release = threading.Event()
|
|
baseline = set(threading.enumerate())
|
|
|
|
def stuck():
|
|
release.wait(5)
|
|
yield {}
|
|
|
|
async def run():
|
|
gen = module._iterate_sync_in_thread(stuck, timeout=30)
|
|
task = asyncio.create_task(anext(gen))
|
|
await asyncio.sleep(0.05)
|
|
task.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await task
|
|
await gen.aclose()
|
|
assert not (set(threading.enumerate()) - baseline)
|
|
|
|
try:
|
|
asyncio.run(run())
|
|
finally:
|
|
release.set()
|
|
|
|
|
|
@pytest.mark.parametrize("name", ["dashscope", "tbox"])
|
|
def test_process_cancel_timeout_backpressure_and_reaping(name, tmp_path):
|
|
with load(name):
|
|
module = importlib.import_module("pkg.vendor_process")
|
|
script = tmp_path / "fixture.py"
|
|
script.write_text(
|
|
'import sys,json,time,os\np=json.loads(sys.stdin.readline())\nprint(json.dumps({"item":{"pid":os.getpid()}}),flush=True)\nif p["operation"]=="hang": time.sleep(60)\nelse:\n while True: print(json.dumps({"item":{"text":"x"*10000}}),flush=True)\n'
|
|
)
|
|
|
|
async def run():
|
|
for mode in ["hang", "fast"]:
|
|
for _ in range(3):
|
|
gen = module.vendor_stream({"operation": mode}, timeout=0.3, worker=script)
|
|
first = await anext(gen)
|
|
pid = first["pid"]
|
|
await asyncio.sleep(0.05)
|
|
await gen.aclose()
|
|
with pytest.raises(ProcessLookupError):
|
|
os.kill(pid, 0)
|
|
gen = module.vendor_stream({"operation": "hang"}, timeout=0.15, worker=script)
|
|
pid = (await anext(gen))["pid"]
|
|
with pytest.raises(TimeoutError):
|
|
await anext(gen)
|
|
with pytest.raises(ProcessLookupError):
|
|
os.kill(pid, 0)
|
|
gen = module.vendor_stream({"operation": "hang"}, timeout=30, worker=script)
|
|
pid = (await anext(gen))["pid"]
|
|
task = asyncio.create_task(anext(gen))
|
|
await asyncio.sleep(0.03)
|
|
task.cancel()
|
|
await asyncio.sleep(0)
|
|
task.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await task
|
|
with pytest.raises(ProcessLookupError):
|
|
os.kill(pid, 0)
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
@pytest.mark.parametrize("name", ["dashscope", "tbox"])
|
|
def test_vendor_deadline_never_cancels_consumer(name, tmp_path):
|
|
with load(name):
|
|
module = importlib.import_module("pkg.vendor_process")
|
|
script = tmp_path / "fixture.py"
|
|
script.write_text(
|
|
'import sys,json,time\np=json.loads(sys.stdin.readline())\nprint(json.dumps({"item":{}}),flush=True)\ntime.sleep(60)\n'
|
|
)
|
|
|
|
async def run():
|
|
gen = module.vendor_stream({}, timeout=0.1, worker=script)
|
|
await anext(gen)
|
|
try:
|
|
await asyncio.sleep(0.2)
|
|
with pytest.raises(TimeoutError):
|
|
await anext(gen)
|
|
finally:
|
|
await gen.aclose()
|
|
|
|
asyncio.run(run())
|