test: make some BDI tests work again
ref: N25B-301
This commit is contained in:
Twirre Meulenbelt
@@ -1,190 +0,0 @@
|
||||
import json
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from control_backend.agents.bdi.text_belief_extractor_agent.behaviours.text_belief_extractor_behaviour import ( # noqa: E501, We can't shorten this import.
|
||||
TextBeliefExtractorBehaviour,
|
||||
)
|
||||
from spade.message import Message
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_settings():
|
||||
"""
|
||||
Mocks the settings object that the behaviour imports.
|
||||
We patch it at the source where it's imported by the module under test.
|
||||
"""
|
||||
# Create a mock object that mimics the nested structure
|
||||
settings_mock = MagicMock()
|
||||
settings_mock.agent_settings.transcription_name = "transcriber"
|
||||
settings_mock.agent_settings.bdi_belief_collector_name = "collector"
|
||||
settings_mock.agent_settings.host = "fake.host"
|
||||
|
||||
# Use patch to replace the settings object during the test
|
||||
# Adjust 'control_backend.behaviours.belief_from_text.settings' to where
|
||||
# your behaviour file imports it from.
|
||||
with patch(
|
||||
"control_backend.agents.bdi.text_belief_extractor_agent.behaviours"
|
||||
".text_belief_extractor_behaviour.settings",
|
||||
settings_mock,
|
||||
):
|
||||
yield settings_mock
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def behavior(mock_settings):
|
||||
"""
|
||||
Creates an instance of the BDITextBeliefBehaviour behaviour and mocks its
|
||||
agent, logger, send, and receive methods.
|
||||
"""
|
||||
b = TextBeliefExtractorBehaviour()
|
||||
|
||||
b.agent = MagicMock()
|
||||
b.send = AsyncMock()
|
||||
b.receive = AsyncMock()
|
||||
|
||||
return b
|
||||
|
||||
|
||||
def create_mock_message(sender_node: str, body: str, thread: str) -> MagicMock:
|
||||
"""Helper function to create a configured mock message."""
|
||||
msg = MagicMock()
|
||||
msg.sender.node = sender_node # MagicMock automatically creates nested mocks
|
||||
msg.body = body
|
||||
msg.thread = thread
|
||||
return msg
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_run_no_message(behavior):
|
||||
"""
|
||||
Tests the run() method when no message is received.
|
||||
"""
|
||||
# Arrange: Configure receive to return None
|
||||
behavior.receive.return_value = None
|
||||
|
||||
# Act: Run the behavior
|
||||
await behavior.run()
|
||||
|
||||
# Assert
|
||||
# 1. Check that receive was called
|
||||
behavior.receive.assert_called_once()
|
||||
# 2. Check that no message was sent
|
||||
behavior.send.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_run_message_from_other_agent(behavior):
|
||||
"""
|
||||
Tests the run() method when a message is received from an
|
||||
unknown agent (not the transcriber).
|
||||
"""
|
||||
# Arrange: Create a mock message from an unknown sender
|
||||
mock_msg = create_mock_message("unknown", "some data", None)
|
||||
behavior.receive.return_value = mock_msg
|
||||
behavior._process_transcription_demo = MagicMock()
|
||||
|
||||
# Act
|
||||
await behavior.run()
|
||||
|
||||
# Assert
|
||||
# 1. Check that receive was called
|
||||
behavior.receive.assert_called_once()
|
||||
# 2. Check that _process_transcription_demo was not sent
|
||||
behavior._process_transcription_demo.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_run_message_from_transcriber_demo(behavior, mock_settings, monkeypatch):
|
||||
"""
|
||||
Tests the main success path: receiving a message from the
|
||||
transcription agent, which triggers _process_transcription_demo.
|
||||
"""
|
||||
# Arrange: Create a mock message from the transcriber
|
||||
transcription_text = "hello world"
|
||||
mock_msg = create_mock_message(
|
||||
mock_settings.agent_settings.transcription_name, transcription_text, None
|
||||
)
|
||||
behavior.receive.return_value = mock_msg
|
||||
|
||||
# Act
|
||||
await behavior.run()
|
||||
|
||||
# Assert
|
||||
# 1. Check that receive was called
|
||||
behavior.receive.assert_called_once()
|
||||
|
||||
# 2. Check that send was called *once*
|
||||
behavior.send.assert_called_once()
|
||||
|
||||
# 3. Deeply inspect the message that was sent
|
||||
sent_msg: Message = behavior.send.call_args[0][0]
|
||||
|
||||
assert (
|
||||
sent_msg.to
|
||||
== mock_settings.agent_settings.bdi_belief_collector_name
|
||||
+ "@"
|
||||
+ mock_settings.agent_settings.host
|
||||
)
|
||||
|
||||
# Check thread
|
||||
assert sent_msg.thread == "beliefs"
|
||||
|
||||
# Parse the received JSON string back into a dict
|
||||
expected_dict = {
|
||||
"beliefs": {"user_said": [transcription_text]},
|
||||
"type": "belief_extraction_text",
|
||||
}
|
||||
sent_dict = json.loads(sent_msg.body)
|
||||
|
||||
# Assert that the dictionaries are equal
|
||||
assert sent_dict == expected_dict
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_process_transcription_success(behavior, mock_settings):
|
||||
"""
|
||||
Tests the (currently unused) _process_transcription method's
|
||||
success path, using its hardcoded mock response.
|
||||
"""
|
||||
# Arrange
|
||||
test_text = "I am feeling happy"
|
||||
# This is the hardcoded response inside the method
|
||||
expected_response_body = '{"mood": [["happy"]]}'
|
||||
|
||||
# Act
|
||||
await behavior._process_transcription(test_text)
|
||||
|
||||
# Assert
|
||||
# 1. Check that a message was sent
|
||||
behavior.send.assert_called_once()
|
||||
|
||||
# 2. Inspect the sent message
|
||||
sent_msg: Message = behavior.send.call_args[0][0]
|
||||
expected_to = (
|
||||
mock_settings.agent_settings.bdi_belief_collector_name
|
||||
+ "@"
|
||||
+ mock_settings.agent_settings.host
|
||||
)
|
||||
assert str(sent_msg.to) == expected_to
|
||||
assert sent_msg.thread == "beliefs"
|
||||
assert sent_msg.body == expected_response_body
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_process_transcription_json_decode_error(behavior, mock_settings):
|
||||
"""
|
||||
Tests the _process_transcription method's error handling
|
||||
when the (mocked) response is invalid JSON.
|
||||
We do this by patching json.loads to raise an error.
|
||||
"""
|
||||
# Arrange
|
||||
test_text = "I am feeling happy"
|
||||
# Patch json.loads to raise an error when called
|
||||
with patch("json.loads", side_effect=json.JSONDecodeError("Mock error", "", 0)):
|
||||
# Act
|
||||
await behavior._process_transcription(test_text)
|
||||
|
||||
# Assert
|
||||
# 1. Check that NO message was sent
|
||||
behavior.send.assert_not_called()
|
||||
@@ -0,0 +1,68 @@
|
||||
import json
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
import pytest
|
||||
|
||||
from control_backend.agents.bdi.text_belief_extractor_agent.text_belief_extractor_agent import (
|
||||
TextBeliefExtractorAgent,
|
||||
)
|
||||
from control_backend.core.agent_system import InternalMessage
|
||||
from control_backend.core.config import settings
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def patch_settings(monkeypatch):
|
||||
monkeypatch.setattr(settings.agent_settings, "transcription_name", "transcriber", raising=False)
|
||||
monkeypatch.setattr(
|
||||
settings.agent_settings, "bdi_belief_collector_name", "collector", raising=False
|
||||
)
|
||||
monkeypatch.setattr(settings.agent_settings, "host", "fake.host", raising=False)
|
||||
|
||||
|
||||
@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):
|
||||
transcription = "hello world"
|
||||
msg = make_msg(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 == 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):
|
||||
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 == settings.agent_settings.bdi_belief_collector_name
|
||||
assert sent.thread == "beliefs"
|
||||
parsed = json.loads(sent.body)
|
||||
assert parsed["beliefs"]["user_said"] == [transcription]
|
||||
Reference in New Issue
Block a user