mirror of
https://github.com/rennf93/roboco.git
synced 2026-08-03 07:23:24 +02:00
Renn F
64c48356d0
Service-level tests now exercise provider, permissions, project, journal, messaging, work_session, metrics, kanban, extraction, learning, notification, dashboard, llm_routing, a2a, task, repository_base, audit, db_seed, branch_name, indexed_document, query_helpers, agent. API route tests cover provider, journal, project, sessions, dashboard, work_session, tasks, a2a, groups, notifications, agents, channels, messages, kanban, api_resources. Pure-function helpers covered: handlers, deps_helpers, middleware, middleware_docs, transcription, pr templates, agents_config, errors, logging, journal/notification/channel/a2a access, task_lifecycle, streaming, converters, crypto, schemas (common + websocket), events, permissions extras. pyproject ruff per-file-ignores extended for tests so PLR2004 (status code magic values), PLC0415 (lazy imports), PLR0913 (fixture params), ARG001 (unused fixture deps), SIM105, and E501 don't fight test idioms.
62 lines
1.7 KiB
Python
62 lines
1.7 KiB
Python
"""runtime.streaming coverage."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from roboco.runtime.streaming import (
|
|
get_reasoning_stream_callback,
|
|
set_reasoning_stream_callback,
|
|
stream_reasoning,
|
|
)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_callback():
|
|
"""Reset the global callback after each test."""
|
|
yield
|
|
set_reasoning_stream_callback(None)
|
|
|
|
|
|
def test_get_callback_returns_none_initially() -> None:
|
|
set_reasoning_stream_callback(None)
|
|
assert get_reasoning_stream_callback() is None
|
|
|
|
|
|
def test_set_and_get_callback() -> None:
|
|
async def cb(agent_id, chunk, metadata):
|
|
pass
|
|
|
|
set_reasoning_stream_callback(cb)
|
|
assert get_reasoning_stream_callback() is cb
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stream_reasoning_calls_callback() -> None:
|
|
received: list[tuple[str, str, dict]] = []
|
|
|
|
async def cb(agent_id: str, chunk: str, metadata: dict) -> None:
|
|
received.append((agent_id, chunk, metadata))
|
|
|
|
set_reasoning_stream_callback(cb)
|
|
await stream_reasoning("be-dev-1", "thinking...", {"step": 1})
|
|
assert received == [("be-dev-1", "thinking...", {"step": 1})]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stream_reasoning_no_callback_silent() -> None:
|
|
set_reasoning_stream_callback(None)
|
|
# No raise.
|
|
await stream_reasoning("be-dev-1", "chunk")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stream_reasoning_default_metadata_is_empty_dict() -> None:
|
|
received: list[tuple[str, str, dict]] = []
|
|
|
|
async def cb(agent_id: str, chunk: str, metadata: dict) -> None:
|
|
received.append((agent_id, chunk, metadata))
|
|
|
|
set_reasoning_stream_callback(cb)
|
|
await stream_reasoning("be-dev-1", "chunk")
|
|
assert received[0][2] == {}
|