mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-05-30 06:41:51 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 10
hermes-agent/tests/gateway/test_api_server_runs.py
Teknium e2fd462ebe
ci(tests): add pytest-timeout 60s hard cap to break suite-teardown deadlock (#28861) 
* ci(tests): add pytest-timeout 60s hard cap to break suite-teardown deadlock

The full pytest suite reliably hangs at ~96% on origin/main, blowing through
the 20-minute GHA job timeout on every CI push since yesterday. Individual
tests complete in <30s — the deadlock builds up at session teardown after
all tests run, when leaked threads and atexit handlers from thousands of
tests interact and one of them lands in a futex-wait that never resolves.

This PR is a stopgap that unblocks CI immediately + speeds up several slow
tests we found while diagnosing.

Changes
- pyproject.toml: add pytest-timeout==2.4.0 to dev deps; bake
  --timeout=60 --timeout-method=thread into the default addopts.
- scripts/run_tests.sh: re-add --timeout flags directly because the script
  wipes pyproject addopts with -o 'addopts='.
- .github/workflows/tests.yml: explicit --timeout/--timeout-method on the
  CI pytest invocation for clarity.
- gateway/run.py: in _run_agent, if the stream consumer was never created
  (e.g. non-streaming agent or test stub), cancel the stream_task
  immediately instead of waiting out the 5s wait_for timeout. ~5s saved
  per non-streaming gateway test run.
- tests/run_agent/conftest.py: extend _fast_retry_backoff to patch
  agent.conversation_loop.jittered_backoff alongside run_agent.jittered_backoff.
  The retry loop was extracted into agent.conversation_loop which holds its
  own import — patching the run_agent reference alone left tests burning
  real wall-clock backoff seconds.
- tests/run_agent/test_anthropic_error_handling.py
  tests/run_agent/test_run_agent.py (TestRetryExhaustion)
  tests/run_agent/test_fallback_model.py: same conversation_loop fix for
  per-test fixtures (defensive — the conftest covers them too).
- tests/gateway/test_gateway_inactivity_timeout.py: trim run_duration
  10.0 → 2.0 / 5.0 → 2.0 on three tests that wait the full SlowFakeAgent
  duration. Adjusted thresholds proportionally.
- tests/gateway/test_api_server_runs.py: test_stop_interrupt_exception_does_not_crash
  trips the interrupted event in addition to raising, so the slow_run
  thread unblocks at teardown instead of waiting 10s.
- tests/hermes_cli/test_update_gateway_restart.py: also patch
  time.monotonic in the autouse fixture. _wait_for_service_active loops
  on a wall-clock deadline; with sleep no-op'd the loop spun on real
  monotonic until 10s real-time per restart attempt (20s+ per test).
- tests/tools/test_zombie_process_cleanup.py: cut runner._restart_drain_timeout
  5.0 → 0.1 in test_gateway_stop_calls_close.

Suite still hangs at 96% on full no-timeout runs; with these changes CI
runs through to a real pass/fail signal.

* chore(lock): regenerate uv.lock after adding pytest-timeout

* ci: drop pytest-timeout 60 → 30s + bump GHA job 20 → 30 min

Prior commit's timeout=60 was too generous — CI test job still hit the
20-min wall-clock cap with the suite hung at 96% (orphan agent-browser
subprocesses blocking pytest session teardown). The local timeout=20
run completed in 6:17, so 30s is conservative enough to let real tests
finish but aggressive enough to short-circuit deadlocks. Also bump GHA
job timeout to 30 min as a safety margin.

* test: delete 11 pre-existing failing tests + revert monotonic patch

The previous PR commit landed pytest-timeout=30s and the suite now
completes in 18:14 instead of hanging at 96%, but 11 pre-existing tests
fail with real assertions. Per Teknium: nuke them.

Deleted (no replacements):
- tests/gateway/test_restart_resume_pending.py::test_clean_drain_does_not_mark_resume_pending
- tests/gateway/test_restart_resume_pending.py::test_drain_timeout_only_marks_still_running_sessions
- tests/hermes_cli/test_gateway_service.py::TestGatewaySystemServiceRouting::test_gateway_install_passes_system_flags
- tests/hermes_cli/test_gateway_wsl.py::TestGatewayCommandWSLMessages::test_install_wsl_with_systemd_warns
- tests/hermes_cli/test_update_gateway_restart.py::TestCmdUpdateLaunchdRestart::test_update_detects_launchd_and_skips_manual_restart_message
- tests/hermes_cli/test_update_gateway_restart.py::TestCmdUpdateLaunchdRestart::test_update_restarts_profile_manual_gateways
- tests/tools/test_file_operations.py::TestGitBaselineCheck::* (6 tests, entire class — _check_git_baseline helper doesn't exist)

Also reverted my time.monotonic autouse-fixture hack in
test_update_gateway_restart.py — it was causing worker crashes in CI by
poisoning later tests in the same xdist worker. The two slow tests in
that file (~24s and ~20s) will go back to taking real time but should
still finish under the 30s pytest-timeout.

* test: delete more pre-existing CI failures

After previous push 3 more tests failed on CI; cull them all.

Removed:
- tests/hermes_cli/test_update_gateway_restart.py::TestCmdUpdateLaunchdRestart::test_update_without_launchd_shows_manual_restart
- tests/hermes_cli/test_update_gateway_restart.py::TestCmdUpdateLaunchdRestart::test_update_profile_manual_gateway_falls_back_to_sigterm
- tests/hermes_cli/test_update_gateway_restart.py::TestCmdUpdateResetFailedBeforeRestart::test_reset_failed_also_runs_before_retry_restart
- tests/hermes_cli/test_update_gateway_restart.py::TestCmdUpdateResetFailedBeforeRestart::test_final_failure_message_tells_user_to_reset_failed
- tests/run_agent/test_tool_call_args_sanitizer.py::test_marker_message_inserted_when_missing

The 4 update_gateway_restart tests trigger `_wait_for_service_active`
polling on a real wall-clock deadline that occasionally exceeds the 30s
pytest-timeout cap and crashes xdist workers. The marker test has a
pre-existing assertion mismatch.

* test: nuke entire TestCmdUpdateLaunchdRestart class

After surgical deletes of 4 tests this class keeps producing new
worker-crashing tests. The pattern is consistent: any test in this
class that triggers cmd_update's _wait_for_service_active polling
spins on real wall-clock time and trips pytest-timeout's thread
method, crashing the xdist worker.

Just delete the whole class (285 lines, ~10 tests). These exercise
macOS-only launchd behavior that's better tested on a real macOS
runner than in linux xdist.

* test: stub the 2 fallback_model tests that crash xdist workers on CI

* test: delete test_anthropic_error_handling.py + test_fallback_model.py entirely

These two files exercise the agent retry/fallback code paths and
consistently crash xdist workers under pytest-timeout's thread method.
Whack-a-mole-stubbing individual tests just surfaces the next ones.
Nuke both files.

* test: delete tests/hermes_cli/test_update_gateway_restart.py entirely

This file's cmd_update integration tests consistently crash xdist
workers under pytest-timeout's thread method. Surgical deletes just
surface the next set. Removing the whole file.

* ci(tests): switch pytest-timeout method thread → signal

Thread-method has been crashing xdist workers when it interrupts code
that's not interruption-safe (retry loops, threading.Event waits, etc).
Signal method uses SIGALRM which is interpreter-level and cleanly raises
a Failed: Timeout exception in test code. Should stop the worker crash
cascade — failures will surface as proper Timeout markers we can
diagnose individually.
2026-05-19 17:27:24 -07:00

531 lines
21 KiB
Python
Raw Blame History

"""Tests for /v1/runs endpoints: start, status, events, and stop.
Covers:
- POST /v1/runs — start a run (202)
- GET /v1/runs/{run_id} — poll run status
- GET /v1/runs/{run_id}/events — SSE event stream
- POST /v1/runs/{run_id}/stop — interrupt a running agent
- Auth, error handling, and cleanup
"""
import asyncio
import json
import threading
import time as _time
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from aiohttp import web
from aiohttp.test_utils import TestClient, TestServer
from gateway.config import PlatformConfig
from gateway.platforms.api_server import (
APIServerAdapter,
cors_middleware,
security_headers_middleware,
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_adapter(api_key: str = "") -> APIServerAdapter:
"""Create an adapter with optional API key."""
extra = {}
if api_key:
extra["key"] = api_key
config = PlatformConfig(enabled=True, extra=extra)
adapter = APIServerAdapter(config)
return adapter
def _create_runs_app(adapter: APIServerAdapter) -> web.Application:
"""Create an aiohttp app with /v1/runs routes registered."""
mws = [mw for mw in (cors_middleware, security_headers_middleware) if mw is not None]
app = web.Application(middlewares=mws)
app["api_server_adapter"] = adapter
app.router.add_post("/v1/runs", adapter._handle_runs)
app.router.add_get("/v1/runs/{run_id}", adapter._handle_get_run)
app.router.add_get("/v1/runs/{run_id}/events", adapter._handle_run_events)
app.router.add_post("/v1/runs/{run_id}/approval", adapter._handle_run_approval)
app.router.add_post("/v1/runs/{run_id}/stop", adapter._handle_stop_run)
return app
def _make_slow_agent(**kwargs):
"""Create a mock agent that blocks in run_conversation until interrupted.
Returns (mock_agent, agent_ready_event, interrupt_event) where
agent_ready_event is set once run_conversation starts, and
interrupt_event is set when interrupt() is called.
"""
ready = threading.Event()
interrupted = threading.Event()
mock_agent = MagicMock()
def _do_interrupt(message=None):
interrupted.set()
mock_agent.interrupt = MagicMock(side_effect=_do_interrupt)
def _slow_run(user_message=None, conversation_history=None, task_id=None):
ready.set()
# Block until interrupt() is called
interrupted.wait(timeout=10)
return {"final_response": "interrupted"}
mock_agent.run_conversation.side_effect = _slow_run
mock_agent.session_prompt_tokens = 0
mock_agent.session_completion_tokens = 0
mock_agent.session_total_tokens = 0
return mock_agent, ready, interrupted
@pytest.fixture
def adapter():
return _make_adapter()
@pytest.fixture
def auth_adapter():
return _make_adapter(api_key="sk-secret")
# ---------------------------------------------------------------------------
# POST /v1/runs — start a run
# ---------------------------------------------------------------------------
class TestStartRun:
@pytest.mark.asyncio
async def test_start_returns_202(self, adapter):
app = _create_runs_app(adapter)
async with TestClient(TestServer(app)) as cli:
with patch.object(adapter, "_create_agent") as mock_create:
mock_agent = MagicMock()
mock_agent.run_conversation.return_value = {"final_response": "done"}
mock_agent.session_prompt_tokens = 10
mock_agent.session_completion_tokens = 5
mock_agent.session_total_tokens = 15
mock_create.return_value = mock_agent
resp = await cli.post("/v1/runs", json={"input": "hello"})
assert resp.status == 202
data = await resp.json()
assert data["status"] == "started"
assert data["run_id"].startswith("run_")
status_resp = await cli.get(f"/v1/runs/{data['run_id']}")
assert status_resp.status == 200
status = await status_resp.json()
assert status["run_id"] == data["run_id"]
assert status["status"] in {"queued", "running", "completed"}
assert status["object"] == "hermes.run"
@pytest.mark.asyncio
async def test_start_invalid_json_returns_400(self, adapter):
app = _create_runs_app(adapter)
async with TestClient(TestServer(app)) as cli:
resp = await cli.post(
"/v1/runs",
data="not json",
headers={"Content-Type": "application/json"},
)
assert resp.status == 400
@pytest.mark.asyncio
async def test_start_missing_input_returns_400(self, adapter):
app = _create_runs_app(adapter)
async with TestClient(TestServer(app)) as cli:
resp = await cli.post("/v1/runs", json={"model": "test"})
assert resp.status == 400
data = await resp.json()
assert "input" in data["error"]["message"]
@pytest.mark.asyncio
async def test_start_empty_input_returns_400(self, adapter):
app = _create_runs_app(adapter)
async with TestClient(TestServer(app)) as cli:
resp = await cli.post("/v1/runs", json={"input": ""})
assert resp.status == 400
@pytest.mark.asyncio
async def test_start_invalid_history_does_not_allocate_run(self, adapter):
app = _create_runs_app(adapter)
async with TestClient(TestServer(app)) as cli:
resp = await cli.post(
"/v1/runs",
json={"input": "hello", "conversation_history": {"role": "user"}},
)
assert resp.status == 400
assert adapter._run_streams == {}
assert adapter._run_statuses == {}
@pytest.mark.asyncio
async def test_start_requires_auth(self, auth_adapter):
app = _create_runs_app(auth_adapter)
async with TestClient(TestServer(app)) as cli:
resp = await cli.post("/v1/runs", json={"input": "hello"})
assert resp.status == 401
@pytest.mark.asyncio
async def test_start_with_valid_auth(self, auth_adapter):
app = _create_runs_app(auth_adapter)
async with TestClient(TestServer(app)) as cli:
with patch.object(auth_adapter, "_create_agent") as mock_create:
mock_agent = MagicMock()
mock_agent.run_conversation.return_value = {"final_response": "ok"}
mock_agent.session_prompt_tokens = 0
mock_agent.session_completion_tokens = 0
mock_agent.session_total_tokens = 0
mock_create.return_value = mock_agent
resp = await cli.post(
"/v1/runs",
json={"input": "hello"},
headers={"Authorization": "Bearer sk-secret"},
)
assert resp.status == 202
# ---------------------------------------------------------------------------
# GET /v1/runs/{run_id} — poll run status
# ---------------------------------------------------------------------------
class TestRunStatus:
@pytest.mark.asyncio
async def test_status_completed_run_includes_output_and_usage(self, adapter):
app = _create_runs_app(adapter)
async with TestClient(TestServer(app)) as cli:
with patch.object(adapter, "_create_agent") as mock_create:
mock_agent = MagicMock()
mock_agent.run_conversation.return_value = {"final_response": "done"}
mock_agent.session_prompt_tokens = 4
mock_agent.session_completion_tokens = 2
mock_agent.session_total_tokens = 6
mock_create.return_value = mock_agent
resp = await cli.post("/v1/runs", json={"input": "hello"})
data = await resp.json()
run_id = data["run_id"]
for _ in range(20):
status_resp = await cli.get(f"/v1/runs/{run_id}")
assert status_resp.status == 200
status = await status_resp.json()
if status["status"] == "completed":
break
await asyncio.sleep(0.05)
assert status["status"] == "completed"
assert status["output"] == "done"
assert status["usage"]["total_tokens"] == 6
assert status["last_event"] == "run.completed"
@pytest.mark.asyncio
async def test_status_reflects_explicit_session_id(self, adapter):
app = _create_runs_app(adapter)
async with TestClient(TestServer(app)) as cli:
with patch.object(adapter, "_create_agent") as mock_create:
mock_agent = MagicMock()
mock_agent.run_conversation.return_value = {"final_response": "done"}
mock_agent.session_prompt_tokens = 0
mock_agent.session_completion_tokens = 0
mock_agent.session_total_tokens = 0
mock_create.return_value = mock_agent
resp = await cli.post(
"/v1/runs",
json={"input": "hello", "session_id": "space-session"},
)
data = await resp.json()
run_id = data["run_id"]
for _ in range(20):
status_resp = await cli.get(f"/v1/runs/{run_id}")
status = await status_resp.json()
if status["status"] == "completed":
break
await asyncio.sleep(0.05)
mock_agent.run_conversation.assert_called_once()
assert mock_agent.run_conversation.call_args.kwargs["task_id"] == "space-session"
assert status["session_id"] == "space-session"
@pytest.mark.asyncio
async def test_status_not_found_returns_404(self, adapter):
app = _create_runs_app(adapter)
async with TestClient(TestServer(app)) as cli:
resp = await cli.get("/v1/runs/run_nonexistent")
assert resp.status == 404
@pytest.mark.asyncio
async def test_status_requires_auth(self, auth_adapter):
app = _create_runs_app(auth_adapter)
async with TestClient(TestServer(app)) as cli:
resp = await cli.get("/v1/runs/run_any")
assert resp.status == 401
# ---------------------------------------------------------------------------
# GET /v1/runs/{run_id}/events — SSE event stream
# ---------------------------------------------------------------------------
class TestRunEvents:
@pytest.mark.asyncio
async def test_events_stream_returns_completed(self, adapter):
"""Events stream should receive run.completed when agent finishes."""
app = _create_runs_app(adapter)
async with TestClient(TestServer(app)) as cli:
with patch.object(adapter, "_create_agent") as mock_create:
mock_agent = MagicMock()
mock_agent.run_conversation.return_value = {"final_response": "Hello!"}
mock_agent.session_prompt_tokens = 10
mock_agent.session_completion_tokens = 5
mock_agent.session_total_tokens = 15
mock_create.return_value = mock_agent
# Start run
resp = await cli.post("/v1/runs", json={"input": "hello"})
assert resp.status == 202
data = await resp.json()
run_id = data["run_id"]
# Subscribe to events
events_resp = await cli.get(f"/v1/runs/{run_id}/events")
assert events_resp.status == 200
body = await events_resp.text()
# Should contain run.completed
assert "run.completed" in body
assert "Hello!" in body
@pytest.mark.asyncio
async def test_approval_response_without_pending_returns_409(self, adapter):
app = _create_runs_app(adapter)
async with TestClient(TestServer(app)) as cli:
with patch.object(adapter, "_create_agent") as mock_create:
mock_agent = MagicMock()
mock_agent.run_conversation.return_value = {"final_response": "done"}
mock_agent.session_prompt_tokens = 0
mock_agent.session_completion_tokens = 0
mock_agent.session_total_tokens = 0
mock_create.return_value = mock_agent
resp = await cli.post("/v1/runs", json={"input": "hello"})
data = await resp.json()
run_id = data["run_id"]
approval_resp = await cli.post(
f"/v1/runs/{run_id}/approval",
json={"choice": "once"},
)
assert approval_resp.status == 409
approval_data = await approval_resp.json()
assert approval_data["error"]["code"] in {
"approval_not_active",
"approval_not_pending",
}
@pytest.mark.asyncio
async def test_approval_string_false_does_not_resolve_all(self, adapter):
"""Quoted false must not fan out approval resolution across the queue."""
app = _create_runs_app(adapter)
run_id = "run_bool_parse"
adapter._run_statuses[run_id] = {"run_id": run_id, "status": "running"}
adapter._run_approval_sessions[run_id] = "session-123"
async with TestClient(TestServer(app)) as cli:
with patch("tools.approval.resolve_gateway_approval", return_value=1) as mock_resolve:
approval_resp = await cli.post(
f"/v1/runs/{run_id}/approval",
json={"choice": "once", "all": "false"},
)
assert approval_resp.status == 200
mock_resolve.assert_called_once_with(
"session-123",
"once",
resolve_all=False,
)
@pytest.mark.asyncio
async def test_events_not_found_returns_404(self, adapter):
app = _create_runs_app(adapter)
async with TestClient(TestServer(app)) as cli:
resp = await cli.get("/v1/runs/run_nonexistent/events")
assert resp.status == 404
@pytest.mark.asyncio
async def test_events_requires_auth(self, auth_adapter):
app = _create_runs_app(auth_adapter)
async with TestClient(TestServer(app)) as cli:
resp = await cli.get("/v1/runs/run_any/events")
assert resp.status == 401
# ---------------------------------------------------------------------------
# POST /v1/runs/{run_id}/stop — interrupt a running agent
# ---------------------------------------------------------------------------
class TestStopRun:
@pytest.mark.asyncio
async def test_stop_running_agent(self, adapter):
"""Stop should interrupt the agent and cancel the task."""
app = _create_runs_app(adapter)
async with TestClient(TestServer(app)) as cli:
with patch.object(adapter, "_create_agent") as mock_create:
mock_agent, agent_ready, _ = _make_slow_agent()
mock_create.return_value = mock_agent
# Start run
resp = await cli.post("/v1/runs", json={"input": "hello"})
assert resp.status == 202
data = await resp.json()
run_id = data["run_id"]
# Wait for agent to start running in the thread
agent_ready.wait(timeout=3.0)
await asyncio.sleep(0.1)
# Verify agent ref is stored
assert run_id in adapter._active_run_agents
# Stop the run
stop_resp = await cli.post(f"/v1/runs/{run_id}/stop")
assert stop_resp.status == 200
stop_data = await stop_resp.json()
assert stop_data["run_id"] == run_id
assert stop_data["status"] == "stopping"
# Agent interrupt should have been called
mock_agent.interrupt.assert_called_once_with("Stop requested via API")
status_resp = await cli.get(f"/v1/runs/{run_id}")
assert status_resp.status == 200
status_data = await status_resp.json()
assert status_data["status"] in {"stopping", "cancelled"}
# Refs should be cleaned up
await asyncio.sleep(0.5)
assert run_id not in adapter._active_run_agents
assert run_id not in adapter._active_run_tasks
@pytest.mark.asyncio
async def test_stop_nonexistent_run_returns_404(self, adapter):
app = _create_runs_app(adapter)
async with TestClient(TestServer(app)) as cli:
resp = await cli.post("/v1/runs/run_nonexistent/stop")
assert resp.status == 404
@pytest.mark.asyncio
async def test_stop_requires_auth(self, auth_adapter):
app = _create_runs_app(auth_adapter)
async with TestClient(TestServer(app)) as cli:
resp = await cli.post("/v1/runs/run_any/stop")
assert resp.status == 401
@pytest.mark.asyncio
async def test_stop_already_completed_run_returns_404(self, adapter):
"""Stopping a run that already finished should return 404 (refs cleaned up)."""
app = _create_runs_app(adapter)
async with TestClient(TestServer(app)) as cli:
with patch.object(adapter, "_create_agent") as mock_create:
mock_agent = MagicMock()
mock_agent.run_conversation.return_value = {"final_response": "done"}
mock_agent.session_prompt_tokens = 0
mock_agent.session_completion_tokens = 0
mock_agent.session_total_tokens = 0
mock_create.return_value = mock_agent
# Start and wait for completion
resp = await cli.post("/v1/runs", json={"input": "hello"})
assert resp.status == 202
data = await resp.json()
run_id = data["run_id"]
await asyncio.sleep(0.3)
# Run should be done, refs cleaned up
assert run_id not in adapter._active_run_agents
# Stop should return 404
stop_resp = await cli.post(f"/v1/runs/{run_id}/stop")
assert stop_resp.status == 404
@pytest.mark.asyncio
async def test_stop_interrupt_exception_does_not_crash(self, adapter):
"""If agent.interrupt() raises, stop should still succeed."""
app = _create_runs_app(adapter)
async with TestClient(TestServer(app)) as cli:
with patch.object(adapter, "_create_agent") as mock_create:
mock_agent, agent_ready, interrupted = _make_slow_agent()
# Override the interrupt side_effect to raise. Still trip
# ``interrupted`` so the slow_run thread unblocks at teardown
# — without this the agent thread blocks the full 10s
# timeout and the test teardown waits the same amount.
def _raising_interrupt(message=None):
interrupted.set()
raise RuntimeError("interrupt failed")
mock_agent.interrupt = MagicMock(side_effect=_raising_interrupt)
mock_create.return_value = mock_agent
resp = await cli.post("/v1/runs", json={"input": "hello"})
assert resp.status == 202
data = await resp.json()
run_id = data["run_id"]
agent_ready.wait(timeout=3.0)
await asyncio.sleep(0.1)
stop_resp = await cli.post(f"/v1/runs/{run_id}/stop")
assert stop_resp.status == 200
stop_data = await stop_resp.json()
assert stop_data["status"] == "stopping"
@pytest.mark.asyncio
async def test_stop_sends_sentinel_to_events_stream(self, adapter):
"""After stop, the events stream should close."""
app = _create_runs_app(adapter)
async with TestClient(TestServer(app)) as cli:
with patch.object(adapter, "_create_agent") as mock_create:
mock_agent, agent_ready, _ = _make_slow_agent()
mock_create.return_value = mock_agent
# Start run
resp = await cli.post("/v1/runs", json={"input": "hello"})
assert resp.status == 202
data = await resp.json()
run_id = data["run_id"]
agent_ready.wait(timeout=3.0)
await asyncio.sleep(0.1)
# Subscribe to events in background
events_task = asyncio.ensure_future(
cli.get(f"/v1/runs/{run_id}/events")
)
await asyncio.sleep(0.1)
# Stop the run
stop_resp = await cli.post(f"/v1/runs/{run_id}/stop")
assert stop_resp.status == 200
# Events stream should close
events_resp = await asyncio.wait_for(events_task, timeout=5.0)
assert events_resp.status == 200
body = await events_resp.text()
# Stream should have received run.failed and closed
assert "run.failed" in body or "stream closed" in body
Reference in a new issue View git blame Copy permalink