mirror of
https://github.com/rennf93/roboco.git
synced 2026-08-03 07:23:24 +02:00
318 lines
11 KiB
Python
318 lines
11 KiB
Python
"""Coverage for misc model methods, factories, and helper functions.
|
|
|
|
Targets the long-tail of small-percent gaps in roboco.models.* — convenience
|
|
properties, factory functions, lookup helpers, and __post_init__ branches.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
from uuid import uuid4
|
|
|
|
from roboco.models import AgentRole, Team
|
|
from roboco.models.a2a import (
|
|
A2AConversation,
|
|
A2AConversationStatus,
|
|
A2ATaskState,
|
|
a2a_state_to_task_status,
|
|
task_status_to_a2a_state,
|
|
)
|
|
from roboco.models.agents import AgentConfig
|
|
from roboco.models.base import ModelProvider
|
|
from roboco.models.channel import (
|
|
create_announcements_channel,
|
|
create_cell_channel,
|
|
create_cross_cell_channel,
|
|
)
|
|
from roboco.models.handoff import HandoffParams, create_handoff
|
|
from roboco.models.llm_catalog import (
|
|
MODEL_CATALOG,
|
|
_build_anthropic_entries,
|
|
provider_type_for_model,
|
|
)
|
|
from roboco.models.permissions import (
|
|
AgentContext,
|
|
PermissionLevel,
|
|
_build_role_levels,
|
|
)
|
|
from roboco.models.runtime import (
|
|
AgentInstance,
|
|
OrchestratorAgentConfig,
|
|
OrchestratorAgentState,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# AgentConfig convenience properties (roboco/models/agents.py 63, 68, 73, 78)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_agent_config_convenience_properties() -> None:
|
|
cfg = AgentConfig(
|
|
name="Test",
|
|
slug="test",
|
|
role=AgentRole.DEVELOPER,
|
|
team=Team.BACKEND,
|
|
system_prompt="hello",
|
|
)
|
|
# provider, model, temperature, max_tokens come from model_config_data
|
|
assert cfg.provider == cfg.model_config_data.provider
|
|
assert cfg.model == cfg.model_config_data.name
|
|
assert cfg.temperature == cfg.model_config_data.temperature
|
|
assert cfg.max_tokens == cfg.model_config_data.max_tokens
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# AgentInstance.__post_init__ — explicit None id branch (runtime.py 76-77)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_agent_instance_post_init_assigns_uuid_when_falsy() -> None:
|
|
# Forcing an empty UUID(int=0) is falsy → the post_init triggers re-assign.
|
|
inst = AgentInstance.__new__(AgentInstance)
|
|
# Fill required dataclass fields explicitly so __post_init__ runs cleanly.
|
|
inst.id = None # type: ignore[assignment]
|
|
inst.agent_id = "be-dev-1"
|
|
inst.state = OrchestratorAgentState.OFFLINE
|
|
inst.container_id = None
|
|
inst.config = None
|
|
inst.started_at = None
|
|
inst.last_activity = None
|
|
inst.current_task_id = None
|
|
inst.error_count = 0
|
|
inst.waiting_for = None
|
|
inst.waiting_context = {}
|
|
inst.__post_init__()
|
|
assert inst.id is not None
|
|
|
|
|
|
def test_agent_instance_default_factory_assigns_id() -> None:
|
|
inst = AgentInstance(agent_id="be-dev-1")
|
|
assert inst.id is not None
|
|
|
|
|
|
def test_orchestrator_agent_config_defaults() -> None:
|
|
|
|
cfg = OrchestratorAgentConfig(
|
|
agent_id="be-dev-1",
|
|
blueprint_path=Path("/tmp/blueprint"),
|
|
)
|
|
assert cfg.model == "sonnet"
|
|
assert cfg.provider_type == "anthropic"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# llm_catalog: catalog skip when MODEL_MAP missing entry (line 56)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_build_anthropic_entries_skips_missing_models() -> None:
|
|
# Patch MODEL_MAP to be missing one entry → loop hits the `continue` branch.
|
|
with patch(
|
|
"roboco.models.llm_catalog.MODEL_MAP",
|
|
{"opus": "claude-opus-4-6"}, # sonnet+haiku missing
|
|
):
|
|
entries = _build_anthropic_entries()
|
|
# Only one entry survives — opus.
|
|
assert len(entries) == 1
|
|
assert entries[0].model_name == "opus"
|
|
|
|
|
|
def test_provider_type_for_model_known() -> None:
|
|
# Pick the first catalog entry deterministically.
|
|
sample = MODEL_CATALOG[0]
|
|
assert provider_type_for_model(sample.model_name) == sample.provider_type
|
|
|
|
|
|
def test_provider_type_for_model_unknown_returns_none() -> None:
|
|
assert provider_type_for_model("nonexistent-model-x") is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# permissions._build_role_levels — exception branch (lines 35-36)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_build_role_levels_skips_invalid_role() -> None:
|
|
"""Bad role string should be silently swallowed in the except branch."""
|
|
with patch(
|
|
"roboco.models.permissions.ROLE_PERMISSION_LEVELS",
|
|
{"not_a_role": "CEO", "ceo": "CEO"},
|
|
):
|
|
result = _build_role_levels()
|
|
# 'not_a_role' silently dropped; ceo retained.
|
|
assert AgentRole.CEO in result
|
|
assert result[AgentRole.CEO] == PermissionLevel.CEO
|
|
|
|
|
|
def test_build_role_levels_skips_invalid_level() -> None:
|
|
with patch(
|
|
"roboco.models.permissions.ROLE_PERMISSION_LEVELS",
|
|
{"ceo": "NOT_A_LEVEL"},
|
|
):
|
|
result = _build_role_levels()
|
|
assert result == {}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Channel factories — create_cell_channel, create_cross_cell_channel,
|
|
# create_announcements_channel (channel.py 98, 116-117, 135)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_create_cell_channel() -> None:
|
|
members = [uuid4(), uuid4()]
|
|
auditor_id = uuid4()
|
|
ch = create_cell_channel("backend", members, auditor_id)
|
|
assert ch.name == "#backend-cell"
|
|
assert ch.slug == "backend-cell"
|
|
assert ch.members == members
|
|
assert ch.silent_observers == [auditor_id]
|
|
|
|
|
|
def test_create_cross_cell_channel() -> None:
|
|
members = [uuid4(), uuid4()]
|
|
main_pm = uuid4()
|
|
auditor = uuid4()
|
|
ch = create_cross_cell_channel("dev-all", members, main_pm, auditor)
|
|
assert ch.name == "#dev-all"
|
|
# main_pm joins the member list.
|
|
assert main_pm in ch.members
|
|
assert auditor in ch.silent_observers
|
|
|
|
|
|
def test_create_announcements_channel() -> None:
|
|
agents = [uuid4() for _ in range(3)]
|
|
board = [uuid4(), uuid4()]
|
|
main_pm = uuid4()
|
|
auditor = uuid4()
|
|
ch = create_announcements_channel(agents, board, main_pm, auditor)
|
|
assert ch.slug == "announcements"
|
|
# Writers = board + main_pm.
|
|
assert main_pm in ch.writers
|
|
for b in board:
|
|
assert b in ch.writers
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# A2A helpers — A2AConversation methods + a2a_state_to_task_status fallback
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_a2a_conversation_other_agent() -> None:
|
|
conv = A2AConversation(agent_a="be-dev-1", agent_b="fe-dev-1")
|
|
assert conv.other_agent("be-dev-1") == "fe-dev-1"
|
|
assert conv.other_agent("fe-dev-1") == "be-dev-1"
|
|
|
|
|
|
def test_a2a_conversation_my_unread() -> None:
|
|
_UNREAD_A = 2
|
|
_UNREAD_B = 5
|
|
conv = A2AConversation(
|
|
agent_a="be-dev-1",
|
|
agent_b="fe-dev-1",
|
|
unread_by_a=_UNREAD_A,
|
|
unread_by_b=_UNREAD_B,
|
|
)
|
|
assert conv.my_unread("be-dev-1") == _UNREAD_A
|
|
assert conv.my_unread("fe-dev-1") == _UNREAD_B
|
|
|
|
|
|
def test_a2a_conversation_is_participant() -> None:
|
|
conv = A2AConversation(agent_a="be-dev-1", agent_b="fe-dev-1")
|
|
assert conv.is_participant("be-dev-1") is True
|
|
assert conv.is_participant("fe-dev-1") is True
|
|
assert conv.is_participant("ux-dev-1") is False
|
|
|
|
|
|
def test_a2a_state_to_task_status_unknown_returns_pending() -> None:
|
|
# Default fallback for any unknown state is "pending".
|
|
class _FakeState:
|
|
pass
|
|
|
|
fake = _FakeState()
|
|
assert a2a_state_to_task_status(fake) == "pending" # type: ignore[arg-type]
|
|
|
|
|
|
def test_a2a_state_to_task_status_known() -> None:
|
|
assert a2a_state_to_task_status(A2ATaskState.SUBMITTED) == "pending"
|
|
assert a2a_state_to_task_status(A2ATaskState.WORKING) == "in_progress"
|
|
assert a2a_state_to_task_status(A2ATaskState.COMPLETED) == "completed"
|
|
assert a2a_state_to_task_status(A2ATaskState.FAILED) == "cancelled"
|
|
assert a2a_state_to_task_status(A2ATaskState.INPUT_REQUIRED) == "blocked"
|
|
|
|
|
|
def test_task_status_to_a2a_state_known() -> None:
|
|
assert task_status_to_a2a_state("pending") == A2ATaskState.SUBMITTED
|
|
assert task_status_to_a2a_state("in_progress") == A2ATaskState.WORKING
|
|
assert task_status_to_a2a_state("completed") == A2ATaskState.COMPLETED
|
|
assert task_status_to_a2a_state("cancelled") == A2ATaskState.CANCELLED
|
|
assert task_status_to_a2a_state("blocked") == A2ATaskState.INPUT_REQUIRED
|
|
|
|
|
|
def test_task_status_to_a2a_state_unknown_returns_working() -> None:
|
|
assert task_status_to_a2a_state("garbage_status") == A2ATaskState.WORKING
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# AgentContext.level — line 79
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_agent_context_level_returns_role_level() -> None:
|
|
ctx = AgentContext(agent_id=uuid4(), role=AgentRole.CEO)
|
|
assert ctx.level == PermissionLevel.CEO
|
|
|
|
|
|
def test_agent_context_level_falls_back_when_role_missing() -> None:
|
|
"""Edge: build an AgentContext for a role removed from ROLE_LEVELS."""
|
|
ctx = AgentContext(agent_id=uuid4(), role=AgentRole.DEVELOPER)
|
|
# Default fallback returns CELL_MEMBER even when not present.
|
|
assert ctx.level in PermissionLevel
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Conversation status enum + closed/paused round-trip
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_a2a_conversation_status_values() -> None:
|
|
assert A2AConversationStatus.ACTIVE == "active"
|
|
assert A2AConversationStatus.PAUSED == "paused"
|
|
assert A2AConversationStatus.CLOSED == "closed"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Handoff factory (models/handoff.py 202-208)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_create_handoff_includes_changelog_required_doc() -> None:
|
|
task_id = uuid4()
|
|
handoff = create_handoff(
|
|
HandoffParams(
|
|
task_id=task_id,
|
|
summary="Built feature X",
|
|
commits=[{"sha": "abc123", "message": "init"}],
|
|
dev_notes_location="/notes/x",
|
|
new_functionality=["new login"],
|
|
modified_behavior=["redirect"],
|
|
breaking_changes=["remove old endpoint"],
|
|
)
|
|
)
|
|
assert handoff.task_id == task_id
|
|
assert handoff.summary == "Built feature X"
|
|
# required_docs always includes a changelog item.
|
|
doc_types = [d.doc_type for d in handoff.required_docs]
|
|
assert "changelog" in doc_types
|
|
assert handoff.new_functionality == ["new login"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Sanity: ModelProvider import is reachable
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_model_provider_enum_has_anthropic() -> None:
|
|
assert ModelProvider.ANTHROPIC == "anthropic"
|