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.
74 lines
1.6 KiB
Python
74 lines
1.6 KiB
Python
"""api.schemas.websocket coverage."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from uuid import uuid4
|
|
|
|
from roboco.api.schemas.websocket import (
|
|
NewMessageBroadcast,
|
|
WSAgentStream,
|
|
WSMessage,
|
|
WSMessageDelete,
|
|
WSMessageEdit,
|
|
WSMessageNew,
|
|
WSNotification,
|
|
WSSessionClosed,
|
|
)
|
|
|
|
|
|
def test_new_message_broadcast() -> None:
|
|
bcast = NewMessageBroadcast(
|
|
channel_id=uuid4(),
|
|
session_id=uuid4(),
|
|
message_id=uuid4(),
|
|
agent_id=uuid4(),
|
|
content="hello",
|
|
message_type="dialogue",
|
|
)
|
|
assert bcast.content == "hello"
|
|
|
|
|
|
def test_ws_message_base() -> None:
|
|
msg = WSMessage(type="custom")
|
|
assert msg.type == "custom"
|
|
|
|
|
|
def test_ws_message_new() -> None:
|
|
msg = WSMessageNew(
|
|
message_id=uuid4(),
|
|
agent_id=uuid4(),
|
|
content="hi",
|
|
message_type="dialogue",
|
|
)
|
|
assert msg.type == "message.new"
|
|
|
|
|
|
def test_ws_message_edit() -> None:
|
|
msg = WSMessageEdit(message_id=uuid4(), content="edited")
|
|
assert msg.type == "message.edit"
|
|
|
|
|
|
def test_ws_message_delete() -> None:
|
|
msg = WSMessageDelete(message_id=uuid4())
|
|
assert msg.type == "message.delete"
|
|
|
|
|
|
def test_ws_agent_stream() -> None:
|
|
msg = WSAgentStream(agent_id=uuid4(), chunk="thinking...")
|
|
assert msg.type == "agent.stream"
|
|
|
|
|
|
def test_ws_session_closed() -> None:
|
|
msg = WSSessionClosed(session_id=uuid4(), reason="timeout")
|
|
assert msg.type == "session.closed"
|
|
|
|
|
|
def test_ws_notification() -> None:
|
|
msg = WSNotification(
|
|
notification_id=uuid4(),
|
|
notification_type="MENTION",
|
|
subject="hi",
|
|
priority="normal",
|
|
)
|
|
assert msg.type == "notification"
|