test: Add test for capturing output from multiple asyncio tasks in parallel

This commit is contained in:
Alex Davies 2025-01-08 11:36:00 -04:00
parent 1208621d5e
commit 7aa5115165

View File

@ -16,6 +16,33 @@ def test_sync_capture_line_by_line():
assert "[stdout] Line 2" in output
assert "[stderr] Error!" in output
@pytest.mark.asyncio
async def test_async_capture_parallel_tasks():
output = []
def callback(data):
output.append(data)
async def task1():
async with CaptureOutput(callback).async_context():
print("Task 1 - Line 1")
print("Task 1 - Line 2")
print("Task 1 - Error!", file=sys.stderr)
async def task2():
async with CaptureOutput(callback).async_context():
print("Task 2 - Line 1")
print("Task 2 - Line 2")
print("Task 2 - Error!", file=sys.stderr)
await asyncio.gather(task1(), task2())
assert "[stdout] Task 1 - Line 1" in output
assert "[stdout] Task 1 - Line 2" in output
assert "[stderr] Task 1 - Error!" in output
assert "[stdout] Task 2 - Line 1" in output
assert "[stdout] Task 2 - Line 2" in output
assert "[stderr] Task 2 - Error!" in output
@pytest.mark.asyncio
async def test_async_capture_line_by_line():
output = []