66 lines
2.2 KiB
Python
66 lines
2.2 KiB
Python
import json
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
|
|
from control_backend.agents.bdi import (
|
|
TextBeliefExtractorAgent,
|
|
)
|
|
from control_backend.core.agent_system import InternalMessage
|
|
|
|
|
|
@pytest.fixture
|
|
def agent():
|
|
agent = TextBeliefExtractorAgent("text_belief_agent")
|
|
agent.send = AsyncMock()
|
|
return agent
|
|
|
|
|
|
def make_msg(sender: str, body: str, thread: str | None = None) -> InternalMessage:
|
|
return InternalMessage(to="unused", sender=sender, body=body, thread=thread)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_message_ignores_other_agents(agent):
|
|
msg = make_msg("unknown", "some data", None)
|
|
|
|
await agent.handle_message(msg)
|
|
|
|
agent.send.assert_not_called() # noqa # `agent.send` has no such property, but we mock it.
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_message_from_transcriber(agent, mock_settings):
|
|
transcription = "hello world"
|
|
msg = make_msg(mock_settings.agent_settings.transcription_name, transcription, None)
|
|
|
|
await agent.handle_message(msg)
|
|
|
|
agent.send.assert_awaited_once() # noqa # `agent.send` has no such property, but we mock it.
|
|
sent: InternalMessage = agent.send.call_args.args[0] # noqa
|
|
assert sent.to == mock_settings.agent_settings.bdi_belief_collector_name
|
|
assert sent.thread == "beliefs"
|
|
parsed = json.loads(sent.body)
|
|
assert parsed == {"beliefs": {"user_said": [transcription]}, "type": "belief_extraction_text"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_process_transcription_demo(agent, mock_settings):
|
|
transcription = "this is a test"
|
|
|
|
await agent._process_transcription_demo(transcription)
|
|
|
|
agent.send.assert_awaited_once() # noqa # `agent.send` has no such property, but we mock it.
|
|
sent: InternalMessage = agent.send.call_args.args[0] # noqa
|
|
assert sent.to == mock_settings.agent_settings.bdi_belief_collector_name
|
|
assert sent.thread == "beliefs"
|
|
parsed = json.loads(sent.body)
|
|
assert parsed["beliefs"]["user_said"] == [transcription]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_setup_initializes_beliefs(agent):
|
|
"""Covers the setup method and ensures beliefs are initialized."""
|
|
await agent.setup()
|
|
assert agent.beliefs == {"mood": ["X"], "car": ["Y"]}
|