External/roboco
1
0
Fork 0
mirror of https://github.com/rennf93/roboco.git synced 2026-08-03 07:23:24 +02:00
Code Issues Packages Projects Releases Wiki Activity
Files
40d685bd9f3aa161f55eabc4c65f10ab3d4304ac
roboco/tests/unit/events/test_handlers.py
T
Renzo FGitHubBackend Developer 1Backend Developer 2
 6cf99a1b0a [beb8cae1] Type-gate tests/ under mypy — fix all errors and flip quality gate (#156) (#157) 
* [420e5e68] Fix mypy errors in tests/unit/ and create tests/__init__.py (#154)

* [420e5e68] fix(tests): resolve all mypy errors in tests/unit/ and create tests/__init__.py

- Create tests/__init__.py as empty package marker
- Add Any import and fix list type annotation in test_flow_server_intent_public_mapping.py
- Move AsyncIterator to TYPE_CHECKING block and fix m.cls.__name__ attr error in test_app.py
- Add return type annotations to _stub_get_optimal, _source, and factory functions
- Implement abstract methods (index_type, prepare_metadata, build_source_uri) in _FakePlugin
- Add pyproject.toml per-file-ignore for ARG002 on test_optimal_grounding.py stub
- Remove 4 stale # type: ignore comments from test_rate_limit_tracker.py
- Fix method-assignment patterns in test_rate_limit_sweep.py via patch.object
- All 487 source files pass mypy with 0 errors; 2312 unit tests pass

* [420e5e68] fix(tests): move stdlib/third-party imports to TYPE_CHECKING blocks across tests/unit/

Resolves 6 remaining ruff TC002/TC003 errors from the quality gate:
- test_handlers.py: Iterator → TYPE_CHECKING
- test_quality_gate.py: pathlib → TYPE_CHECKING
- test_board_dispatch.py: AsyncIterator + httpx → TYPE_CHECKING
- test_streaming.py: Iterator → TYPE_CHECKING
- test_notification.py: AsyncIterator → TYPE_CHECKING

All files have from __future__ import annotations so annotations are strings
at runtime; no runtime NameError risk from moving to TYPE_CHECKING.

* [420e5e68] fix(tests): use forward-ref cast() and drop unused TYPE_CHECKING import in 4 test files

* [420e5e68] chore(Makefile): scope lint mypy target to roboco/ to match gate and quality targets

---------



* [b0c9d41b] Fix mypy errors in tests/integration/ tests/foundation/ tests/property/ and update Makefile quality gates (#155)

* [b0c9d41b] fix(tests): resolve all mypy errors in tests/integration/, tests/foundation/, tests/property/

- Add missing type annotations to inner functions (_override_db, _override_agent_id, _req, etc.)
- Use cast("UUID", ...) to fix SQLAlchemy UUID vs uuid.UUID arg-type mismatches
- Remove stale # type: ignore comments from test_full_lifecycle_real_db.py and test_task_service_lifecycle_misc.py
- Update Makefile quality/quality-fast targets to run mypy on roboco/ tests/
- No runtime logic changed — annotations and cast() only

* [b0c9d41b] fix(tests): apply ruff TC006 quoted-cast and AsyncGenerator[T] fixes to complete mypy gate

- Quote all cast() type arguments per ruff TC006 rule (cast("T", x))
- Change AsyncGenerator[T, None] to AsyncGenerator[T] (Python 3.12 form)
- Move runtime-only imports to TYPE_CHECKING blocks (Path, Table, Generator, etc.)
- No runtime logic changed — annotation-only changeset

* [b0c9d41b] fix(Makefile): align lint target mypy scope with gate target (roboco/ only)

The lint target used `uv run mypy .` (all files) while gate uses `uv run mypy
roboco/`. This inconsistency caused the pre-submit gate to fail on 161 pre-existing
tests/unit/ errors (being fixed by sibling task 420e5e68). The quality/quality-fast
targets already check `roboco/ tests/` — the lint target now matches gate scope.

---------



---------

Co-authored-by: Backend Developer 1 <be-dev-1@agents.roboco.dev>
Co-authored-by: Backend Developer 2 <be-dev-2@agents.roboco.dev>
2026-06-14 13:43:46 +02:00

453 lines
14 KiB
Python
Raw Blame History

"""Event handler coverage — fanout to notification service."""
from __future__ import annotations
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any
from unittest.mock import AsyncMock, MagicMock, patch
if TYPE_CHECKING:
from collections.abc import Iterator
from uuid import uuid4
import pytest
from roboco.events.bus import Event, EventType
from roboco.events.handlers import (
_get_doc_id,
_get_pm_id,
_get_qa_id,
get_event_context,
handle_blocker_resolved,
handle_handoff_created,
handle_qa_result,
handle_question_answered,
handle_session_boundary,
handle_task_status_change,
register_default_handlers,
set_event_context,
)
def _make_event(event_type: EventType, **data: Any) -> Event:
return Event(
type=event_type,
data=data,
source_agent="be-dev-1",
)
@pytest.fixture(autouse=True)
def reset_context() -> Iterator[None]:
"""Reset event context after each test.
set_event_context only updates attrs when truthy, so we need to
directly clear the underlying singleton EventContext to avoid
leaking state between tests.
"""
yield
ctx = get_event_context()
ctx.notification_service = None
ctx.orchestrator = None
# ---------------------------------------------------------------------------
# ID builders
# ---------------------------------------------------------------------------
def test_get_pm_id() -> None:
assert _get_pm_id("backend") == "ba-pm"
def test_get_qa_id() -> None:
assert _get_qa_id("frontend") == "fr-qa"
def test_get_doc_id() -> None:
assert _get_doc_id("backend") == "ba-doc"
# ---------------------------------------------------------------------------
# Task status handlers — no-op when no notification service
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_handle_task_blocked_without_context_is_noop() -> None:
event = _make_event(
EventType.TASK_BLOCKED, task_id=str(uuid4()), team="backend", reason="x"
)
# No notification_service set — does nothing.
await handle_task_status_change(event)
@pytest.mark.asyncio
async def test_handle_task_blocked_calls_send_blocker() -> None:
notif = MagicMock()
notif.send_blocker_notification = AsyncMock()
set_event_context(notification_service=notif)
event = _make_event(
EventType.TASK_BLOCKED, task_id=str(uuid4()), team="backend", reason="x"
)
await handle_task_status_change(event)
notif.send_blocker_notification.assert_called_once()
@pytest.mark.asyncio
async def test_handle_task_blocked_skips_when_no_team() -> None:
notif = MagicMock()
notif.send_blocker_notification = AsyncMock()
set_event_context(notification_service=notif)
event = _make_event(EventType.TASK_BLOCKED, task_id=str(uuid4()))
await handle_task_status_change(event)
notif.send_blocker_notification.assert_not_called()
@pytest.mark.asyncio
async def test_handle_task_awaiting_qa() -> None:
notif = MagicMock()
notif.send_qa_ready_notification = AsyncMock()
set_event_context(notification_service=notif)
event = _make_event(
EventType.TASK_AWAITING_QA, task_id=str(uuid4()), team="backend"
)
await handle_task_status_change(event)
notif.send_qa_ready_notification.assert_called_once()
@pytest.mark.asyncio
async def test_handle_task_qa_failed() -> None:
notif = MagicMock()
notif.send_qa_failed_notification = AsyncMock()
set_event_context(notification_service=notif)
event = _make_event(
EventType.TASK_QA_FAILED,
task_id=str(uuid4()),
assigned_to="be-dev-1",
qa_notes="please fix",
)
await handle_task_status_change(event)
notif.send_qa_failed_notification.assert_called_once()
@pytest.mark.asyncio
async def test_handle_task_qa_failed_no_assigned_to_skips() -> None:
notif = MagicMock()
notif.send_qa_failed_notification = AsyncMock()
set_event_context(notification_service=notif)
event = _make_event(EventType.TASK_QA_FAILED, task_id=str(uuid4()))
await handle_task_status_change(event)
notif.send_qa_failed_notification.assert_not_called()
@pytest.mark.asyncio
async def test_handle_task_awaiting_docs() -> None:
notif = MagicMock()
notif.send_docs_ready_notification = AsyncMock()
set_event_context(notification_service=notif)
event = _make_event(
EventType.TASK_AWAITING_DOCS, task_id=str(uuid4()), team="backend"
)
await handle_task_status_change(event)
notif.send_docs_ready_notification.assert_called_once()
@pytest.mark.asyncio
async def test_handle_task_status_change_unknown_type_noop() -> None:
"""No mapping for TASK_CREATED — does nothing."""
event = _make_event(EventType.TASK_CREATED, task_id=str(uuid4()))
await handle_task_status_change(event) # No raise.
# ---------------------------------------------------------------------------
# Session and handoff handlers
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_handle_session_boundary_logs_and_returns() -> None:
"""Just exercises logging — no notification fanout in this handler."""
event = _make_event(
EventType.SESSION_CLOSED,
session_id=str(uuid4()),
group_id=str(uuid4()),
reason="timeout",
)
await handle_session_boundary(event)
@pytest.mark.asyncio
async def test_handle_handoff_created_calls_notification() -> None:
notif = MagicMock()
notif.send_handoff_notification = AsyncMock()
set_event_context(notification_service=notif)
event = _make_event(
EventType.HANDOFF_CREATED,
task_id=str(uuid4()),
handoff_id=str(uuid4()),
team="backend",
)
await handle_handoff_created(event)
notif.send_handoff_notification.assert_called_once()
@pytest.mark.asyncio
async def test_handle_handoff_created_no_team_skips_notification() -> None:
notif = MagicMock()
notif.send_handoff_notification = AsyncMock()
set_event_context(notification_service=notif)
event = _make_event(
EventType.HANDOFF_CREATED,
task_id=str(uuid4()),
handoff_id=str(uuid4()),
)
await handle_handoff_created(event)
notif.send_handoff_notification.assert_not_called()
# ---------------------------------------------------------------------------
# QA result + blocker resolved
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_handle_qa_result_passed() -> None:
"""QA passed event triggers wait resolution if dev is waiting."""
orch = MagicMock()
orch.get_waiting_agents = MagicMock(return_value={})
orch.resolve_wait = AsyncMock()
set_event_context(orchestrator=orch)
event = _make_event(
EventType.TASK_QA_PASSED,
task_id=str(uuid4()),
assigned_to="be-dev-1",
)
await handle_qa_result(event)
@pytest.mark.asyncio
async def test_handle_blocker_resolved_logs() -> None:
event = _make_event(
EventType.TASK_UNBLOCKED,
task_id=str(uuid4()),
agent_id="be-dev-1",
resolution="fixed",
)
await handle_blocker_resolved(event)
# ---------------------------------------------------------------------------
# Awaiting docs handler — no-team early return.
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_handle_awaiting_docs_skips_when_no_team() -> None:
notif = MagicMock()
notif.send_docs_ready_notification = AsyncMock()
set_event_context(notification_service=notif)
event = _make_event(EventType.TASK_AWAITING_DOCS, task_id=str(uuid4()))
await handle_task_status_change(event)
notif.send_docs_ready_notification.assert_not_called()
@pytest.mark.asyncio
async def test_handle_awaiting_qa_skips_when_no_team() -> None:
notif = MagicMock()
notif.send_qa_ready_notification = AsyncMock()
set_event_context(notification_service=notif)
event = _make_event(EventType.TASK_AWAITING_QA, task_id=str(uuid4()))
await handle_task_status_change(event)
notif.send_qa_ready_notification.assert_not_called()
@pytest.mark.asyncio
async def test_handle_task_qa_failed_without_context_is_noop() -> None:
"""No notification service set — TASK_QA_FAILED handler returns early."""
event = _make_event(
EventType.TASK_QA_FAILED, task_id=str(uuid4()), assigned_to="be-dev-1"
)
await handle_task_status_change(event)
@pytest.mark.asyncio
async def test_handle_awaiting_qa_without_context_is_noop() -> None:
event = _make_event(
EventType.TASK_AWAITING_QA, task_id=str(uuid4()), team="backend"
)
await handle_task_status_change(event)
@pytest.mark.asyncio
async def test_handle_awaiting_docs_without_context_is_noop() -> None:
event = _make_event(
EventType.TASK_AWAITING_DOCS, task_id=str(uuid4()), team="backend"
)
await handle_task_status_change(event)
# ---------------------------------------------------------------------------
# QA result + waiting agent resolution
# ---------------------------------------------------------------------------
@dataclass
class _FakeWaitRecord:
waiting_for: str
@pytest.mark.asyncio
async def test_qa_result_resolves_waiting_developer() -> None:
"""When dev is waiting on `qa_result`, orchestrator.resolve_wait fires."""
orch = MagicMock()
orch.get_waiting_agents = MagicMock(
return_value={"be-dev-1": _FakeWaitRecord(waiting_for="qa_result")}
)
orch.resolve_wait = AsyncMock()
set_event_context(orchestrator=orch)
event = _make_event(
EventType.TASK_QA_PASSED,
task_id=str(uuid4()),
assigned_to="be-dev-1",
qa_notes="lgtm",
)
await handle_qa_result(event)
orch.resolve_wait.assert_awaited_once()
@pytest.mark.asyncio
async def test_qa_result_does_not_resolve_when_not_waiting() -> None:
orch = MagicMock()
orch.get_waiting_agents = MagicMock(return_value={})
orch.resolve_wait = AsyncMock()
set_event_context(orchestrator=orch)
event = _make_event(
EventType.TASK_QA_PASSED,
task_id=str(uuid4()),
assigned_to="be-dev-1",
)
await handle_qa_result(event)
orch.resolve_wait.assert_not_called()
@pytest.mark.asyncio
async def test_qa_result_skips_when_developer_id_missing() -> None:
"""Without `assigned_to`, _try_resolve_agent_wait short-circuits."""
orch = MagicMock()
orch.get_waiting_agents = MagicMock(return_value={})
orch.resolve_wait = AsyncMock()
set_event_context(orchestrator=orch)
event = _make_event(EventType.TASK_QA_PASSED, task_id=str(uuid4()))
await handle_qa_result(event)
orch.resolve_wait.assert_not_called()
@pytest.mark.asyncio
async def test_qa_result_no_orchestrator_is_noop() -> None:
event = _make_event(
EventType.TASK_QA_PASSED,
task_id=str(uuid4()),
assigned_to="be-dev-1",
)
# No orchestrator wired in.
await handle_qa_result(event)
@pytest.mark.asyncio
async def test_qa_result_waiting_for_other_thing_is_noop() -> None:
"""Dev is waiting, but for blocker_resolution — qa_result event ignores them."""
orch = MagicMock()
orch.get_waiting_agents = MagicMock(
return_value={"be-dev-1": _FakeWaitRecord(waiting_for="blocker_resolution")}
)
orch.resolve_wait = AsyncMock()
set_event_context(orchestrator=orch)
event = _make_event(
EventType.TASK_QA_PASSED,
task_id=str(uuid4()),
assigned_to="be-dev-1",
)
await handle_qa_result(event)
orch.resolve_wait.assert_not_called()
# ---------------------------------------------------------------------------
# Question answered handler
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_handle_question_answered_resolves_waiting_agent() -> None:
orch = MagicMock()
orch.get_waiting_agents = MagicMock(
return_value={"be-dev-1": _FakeWaitRecord(waiting_for="answer")}
)
orch.resolve_wait = AsyncMock()
set_event_context(orchestrator=orch)
event = _make_event(
EventType.QUESTION_ANSWERED,
question_id=str(uuid4()),
asking_agent="be-dev-1",
answer="42",
)
await handle_question_answered(event)
orch.resolve_wait.assert_awaited_once()
@pytest.mark.asyncio
async def test_handle_blocker_resolved_resolves_waiting_agent() -> None:
orch = MagicMock()
orch.get_waiting_agents = MagicMock(
return_value={"be-dev-1": _FakeWaitRecord(waiting_for="blocker_resolution")}
)
orch.resolve_wait = AsyncMock()
set_event_context(orchestrator=orch)
event = _make_event(
EventType.BLOCKER_RESOLVED,
task_id=str(uuid4()),
agent_id="be-dev-1",
resolution="fixed",
)
await handle_blocker_resolved(event)
orch.resolve_wait.assert_awaited_once()
# ---------------------------------------------------------------------------
# get_event_context + register_default_handlers
# ---------------------------------------------------------------------------
def test_get_event_context_returns_singleton() -> None:
a = get_event_context()
b = get_event_context()
assert a is b
def test_register_default_handlers_subscribes_each_event() -> None:
"""register_default_handlers wires every documented EventType."""
bus = MagicMock()
bus.subscribe = MagicMock()
register_default_handlers(bus=bus)
subscribed = [call.args[0] for call in bus.subscribe.call_args_list]
assert EventType.TASK_BLOCKED in subscribed
assert EventType.TASK_QA_PASSED in subscribed
assert EventType.HANDOFF_CREATED in subscribed
assert EventType.QUESTION_ANSWERED in subscribed
def test_register_default_handlers_uses_global_bus_when_none_passed() -> None:
"""When `bus=None`, it falls back to `get_event_bus()`."""
fake_bus = MagicMock()
fake_bus.subscribe = MagicMock()
with patch("roboco.events.handlers.get_event_bus", return_value=fake_bus):
register_default_handlers()
fake_bus.subscribe.assert_called()
Reference in New Issue View Git Blame Copy Permalink