rp/tests/test_agents.py

from pr.agents.agent_communication import (
    AgentCommunicationBus,
    AgentMessage,
    MessageType,
)
from pr.agents.agent_manager import AgentInstance, AgentManager
from pr.agents.agent_roles import AgentRole, get_agent_role, list_agent_roles


def test_get_agent_role():
    role = get_agent_role("coding")
    assert isinstance(role, AgentRole)
    assert role.name == "coding"


def test_list_agent_roles():
    roles = list_agent_roles()
    assert isinstance(roles, dict)
    assert len(roles) > 0
    assert "coding" in roles


def test_agent_role():
    role = AgentRole(
        name="test",
        description="test",
        system_prompt="test",
        allowed_tools=set(),
        specialization_areas=[],
    )
    assert role.name == "test"


def test_agent_instance():
    role = get_agent_role("coding")
    instance = AgentInstance(agent_id="test", role=role)
    assert instance.agent_id == "test"
    assert instance.role == role


def test_agent_manager_init():
    mgr = AgentManager(":memory:", None)
    assert mgr is not None


def test_agent_manager_create_agent():
    mgr = AgentManager(":memory:", None)
    agent = mgr.create_agent("coding", "test_agent")
    assert agent is not None


def test_agent_manager_get_agent():
    mgr = AgentManager(":memory:", None)
    mgr.create_agent("coding", "test_agent")
    agent = mgr.get_agent("test_agent")
    assert isinstance(agent, AgentInstance)


def test_agent_manager_remove_agent():
    mgr = AgentManager(":memory:", None)
    mgr.create_agent("coding", "test_agent")
    mgr.remove_agent("test_agent")
    agent = mgr.get_agent("test_agent")
    assert agent is None


def test_agent_manager_send_agent_message():
    mgr = AgentManager(":memory:", None)
    mgr.create_agent("coding", "a")
    mgr.create_agent("coding", "b")
    mgr.send_agent_message("a", "b", "test")
    assert True


def test_agent_manager_get_agent_messages():
    mgr = AgentManager(":memory:", None)
    mgr.create_agent("coding", "test")
    messages = mgr.get_agent_messages("test")
    assert isinstance(messages, list)


def test_agent_manager_get_session_summary():
    mgr = AgentManager(":memory:", None)
    summary = mgr.get_session_summary()
    assert isinstance(summary, str)


def test_agent_manager_collaborate_agents():
    mgr = AgentManager(":memory:", None)
    result = mgr.collaborate_agents("orchestrator", "task", ["coding", "research"])
    assert result is not None


def test_agent_manager_execute_agent_task():
    mgr = AgentManager(":memory:", None)
    mgr.create_agent("coding", "test")
    result = mgr.execute_agent_task("test", "task")
    assert result is not None


def test_agent_manager_clear_session():
    mgr = AgentManager(":memory:", None)
    mgr.clear_session()
    assert True


def test_agent_message():
    msg = AgentMessage(
        from_agent="a",
        to_agent="b",
        message_type=MessageType.REQUEST,
        content="test",
        metadata={},
        timestamp=1.0,
        message_id="id",
    )
    assert msg.from_agent == "a"


def test_agent_message_to_dict():
    msg = AgentMessage(
        from_agent="a",
        to_agent="b",
        message_type=MessageType.REQUEST,
        content="test",
        metadata={},
        timestamp=1.0,
        message_id="id",
    )
    d = msg.to_dict()
    assert isinstance(d, dict)


def test_agent_message_from_dict():
    d = {
        "from_agent": "a",
        "to_agent": "b",
        "message_type": "request",
        "content": "test",
        "metadata": {},
        "timestamp": 1.0,
        "message_id": "id",
    }
    msg = AgentMessage.from_dict(d)
    assert isinstance(msg, AgentMessage)


def test_agent_communication_bus_init():
    bus = AgentCommunicationBus(":memory:")
    assert bus is not None


def test_agent_communication_bus_send_message():
    bus = AgentCommunicationBus(":memory:")
    msg = AgentMessage(
        from_agent="a",
        to_agent="b",
        message_type=MessageType.REQUEST,
        content="test",
        metadata={},
        timestamp=1.0,
        message_id="id",
    )
    bus.send_message(msg)
    assert True


def test_agent_communication_bus_receive_messages():
    bus = AgentCommunicationBus(":memory:")
    msg = AgentMessage(
        from_agent="a",
        to_agent="b",
        message_type=MessageType.REQUEST,
        content="test",
        metadata={},
        timestamp=1.0,
        message_id="id",
    )
    bus.send_message(msg)
    messages = bus.receive_messages("b")
    assert len(messages) == 1


def test_agent_communication_bus_get_conversation_history():
    bus = AgentCommunicationBus(":memory:")
    msg = AgentMessage(
        from_agent="a",
        to_agent="b",
        message_type=MessageType.REQUEST,
        content="test",
        metadata={},
        timestamp=1.0,
        message_id="id",
    )
    bus.send_message(msg)
    history = bus.get_conversation_history("a", "b")
    assert len(history) == 1


def test_agent_communication_bus_mark_as_read():
    bus = AgentCommunicationBus(":memory:")
    msg = AgentMessage(
        from_agent="a",
        to_agent="b",
        message_type=MessageType.REQUEST,
        content="test",
        metadata={},
        timestamp=1.0,
        message_id="id",
    )
    bus.send_message(msg)
    bus.mark_as_read(msg.message_id)
    assert True