An unofficial terminal client for Tangled, optimized for humans and agents. tgcli.wisp.place
cli atproto go tangled
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107"""Bounded log collection and stream cleanup tests."""
import asyncio
import pytest
from tests.conftest import FakeTG
@pytest.mark.parametrize("ending", ["\n", ""])async def test_complete_stream(fake_tg: FakeTG, log_line: str, ending: str) -> None: fake_tg.configure(stdout=log_line.rstrip("\n") + ending) result = await fake_tg.runner().collect_logs( [], max_events=10, max_bytes=4096, timeout_seconds=2 ) assert result.stop_reason == "complete" assert len(result.events) == 1 assert result.events[0].data is not None assert result.events[0].data.content == "hello" assert result.error is None
@pytest.mark.parametrize("limit", ["events", "bytes"])async def test_output_limits_keep_prior_events( fake_tg: FakeTG, log_line: str, limit: str) -> None: fake_tg.configure(stdout=log_line * 3, stall=True, spawn_child=True) result = await asyncio.wait_for( fake_tg.runner().collect_logs( [], max_events=1 if limit == "events" else 10, max_bytes=len(log_line.encode()) if limit == "bytes" else 4096, timeout_seconds=2, ), 5, ) assert result.stop_reason == ("event_limit" if limit == "events" else "byte_limit") assert len(result.events) == 1 await fake_tg.assert_stopped()
@pytest.mark.parametrize("server_timeout, collection_timeout", [(1, 10), (10, 1)])async def test_timeout_keeps_events( fake_tg: FakeTG, log_line: str, server_timeout: int, collection_timeout: int) -> None: fake_tg.configure(stdout=log_line, stall=True, spawn_child=True) result = await asyncio.wait_for( fake_tg.runner(timeout_seconds=server_timeout).collect_logs( [], max_events=10, max_bytes=4096, timeout_seconds=collection_timeout ), 5, ) assert result.stop_reason == "timeout" assert len(result.events) == 1 await fake_tg.assert_stopped()
@pytest.mark.parametrize("failure", ["invalid_json", "invalid_model", "exit"])async def test_stream_errors_keep_events( fake_tg: FakeTG, log_line: str, failure: str) -> None: suffix = {"invalid_json": "not json\n", "invalid_model": "{}\n", "exit": ""}[ failure ] fake_tg.configure( stdout=log_line + suffix, stderr="stream failed" if failure == "exit" else "", exit_code=1 if failure == "exit" else 0, ) result = await fake_tg.runner().collect_logs( [], max_events=10, max_bytes=4096, timeout_seconds=2 ) assert result.stop_reason == "error" assert len(result.events) == 1 assert result.error
async def test_oversized_line(fake_tg: FakeTG) -> None: fake_tg.configure(stdout="x" * (2 << 20) + "\n", stall=True, spawn_child=True) result = await asyncio.wait_for( fake_tg.runner().collect_logs( [], max_events=10, max_bytes=1048576, timeout_seconds=2 ), 5, ) assert result.stop_reason == "byte_limit" assert not result.events await fake_tg.assert_stopped()
async def test_cancellation_stops_stream_tree(fake_tg: FakeTG, log_line: str) -> None: fake_tg.configure(stdout=log_line, stall=True, spawn_child=True) task = asyncio.create_task( fake_tg.runner().collect_logs( [], max_events=10, max_bytes=4096, timeout_seconds=10 ) ) try: await fake_tg.wait_ready() task.cancel() with pytest.raises(asyncio.CancelledError): await asyncio.wait_for(task, 5) await fake_tg.assert_stopped() finally: task.cancel() await asyncio.gather(task, return_exceptions=True)