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.
44 lines
1.2 KiB
Python
44 lines
1.2 KiB
Python
"""models.events Event class coverage."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from uuid import uuid4
|
|
|
|
from roboco.events.bus import Event, EventType
|
|
|
|
|
|
def test_event_default_id_and_timestamp() -> None:
|
|
e = Event(type=EventType.TASK_CREATED, data={})
|
|
assert e.id is not None
|
|
assert e.timestamp is not None
|
|
|
|
|
|
def test_event_to_json_round_trip() -> None:
|
|
original = Event(
|
|
type=EventType.TASK_BLOCKED,
|
|
data={"reason": "x"},
|
|
source_agent="be-dev-1",
|
|
correlation_id="abc",
|
|
)
|
|
json_str = original.to_json()
|
|
restored = Event.from_json(json_str)
|
|
assert restored.id == original.id
|
|
assert restored.type == original.type
|
|
assert restored.data == original.data
|
|
assert restored.source_agent == original.source_agent
|
|
assert restored.correlation_id == original.correlation_id
|
|
|
|
|
|
def test_event_to_json_no_optional_fields() -> None:
|
|
e = Event(type=EventType.TASK_CREATED, data={})
|
|
json_str = e.to_json()
|
|
restored = Event.from_json(json_str)
|
|
assert restored.source_agent is None
|
|
assert restored.correlation_id is None
|
|
|
|
|
|
def test_event_with_explicit_id() -> None:
|
|
eid = uuid4()
|
|
e = Event(type=EventType.TASK_BLOCKED, data={}, id=eid)
|
|
assert e.id == eid
|