refactor: testing
Redid testing structure, added tests and changed some tests. ref: N25B-301
This commit is contained in:
@@ -1,211 +0,0 @@
|
||||
import json
|
||||
import logging
|
||||
from unittest.mock import AsyncMock, MagicMock, call
|
||||
|
||||
import pytest
|
||||
from control_backend.agents.bdi.bdi_core_agent.behaviours.belief_setter_behaviour import (
|
||||
BeliefSetterBehaviour,
|
||||
)
|
||||
|
||||
# Define a constant for the collector agent name to use in tests
|
||||
COLLECTOR_AGENT_NAME = "belief_collector_agent"
|
||||
COLLECTOR_AGENT_JID = f"{COLLECTOR_AGENT_NAME}@test"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_agent(mocker):
|
||||
"""Fixture to create a mock BDIAgent."""
|
||||
agent = MagicMock()
|
||||
agent.bdi = MagicMock()
|
||||
agent.jid = "bdi_agent@test"
|
||||
return agent
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def belief_setter_behaviour(mock_agent, mocker):
|
||||
"""Fixture to create an instance of BeliefSetterBehaviour with a mocked agent."""
|
||||
# Patch the settings to use a predictable agent name
|
||||
mocker.patch(
|
||||
"control_backend.agents.bdi.bdi_core_agent."
|
||||
"behaviours.belief_setter_behaviour.settings.agent_settings.bdi_belief_collector_name",
|
||||
COLLECTOR_AGENT_NAME,
|
||||
)
|
||||
|
||||
setter = BeliefSetterBehaviour()
|
||||
setter.agent = mock_agent
|
||||
# Mock the receive method, we will control its return value in each test
|
||||
setter.receive = AsyncMock()
|
||||
return setter
|
||||
|
||||
|
||||
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_message_received(belief_setter_behaviour, mocker):
|
||||
"""
|
||||
Test that when a message is received, _process_message is called.
|
||||
"""
|
||||
# Arrange
|
||||
msg = MagicMock()
|
||||
belief_setter_behaviour.receive.return_value = msg
|
||||
mocker.patch.object(belief_setter_behaviour, "_process_message")
|
||||
|
||||
# Act
|
||||
await belief_setter_behaviour.run()
|
||||
|
||||
# Assert
|
||||
belief_setter_behaviour._process_message.assert_called_once_with(msg)
|
||||
|
||||
|
||||
def test_process_message_from_bdi_belief_collector_agent(belief_setter_behaviour, mocker):
|
||||
"""
|
||||
Test processing a message from the correct belief collector agent.
|
||||
"""
|
||||
# Arrange
|
||||
msg = create_mock_message(sender_node=COLLECTOR_AGENT_NAME, body="", thread="")
|
||||
mock_process_belief = mocker.patch.object(belief_setter_behaviour, "_process_belief_message")
|
||||
|
||||
# Act
|
||||
belief_setter_behaviour._process_message(msg)
|
||||
|
||||
# Assert
|
||||
mock_process_belief.assert_called_once_with(msg)
|
||||
|
||||
|
||||
def test_process_message_from_other_agent(belief_setter_behaviour, mocker):
|
||||
"""
|
||||
Test that messages from other agents are ignored.
|
||||
"""
|
||||
# Arrange
|
||||
msg = create_mock_message(sender_node="other_agent", body="", thread="")
|
||||
mock_process_belief = mocker.patch.object(belief_setter_behaviour, "_process_belief_message")
|
||||
|
||||
# Act
|
||||
belief_setter_behaviour._process_message(msg)
|
||||
|
||||
# Assert
|
||||
mock_process_belief.assert_not_called()
|
||||
|
||||
|
||||
def test_process_belief_message_valid_json(belief_setter_behaviour, mocker):
|
||||
"""
|
||||
Test processing a valid belief message with correct thread and JSON body.
|
||||
"""
|
||||
# Arrange
|
||||
beliefs_payload = {"is_hot": ["kitchen"], "is_clean": ["kitchen", "bathroom"]}
|
||||
msg = create_mock_message(
|
||||
sender_node=COLLECTOR_AGENT_JID, body=json.dumps(beliefs_payload), thread="beliefs"
|
||||
)
|
||||
mock_set_beliefs = mocker.patch.object(belief_setter_behaviour, "_set_beliefs")
|
||||
|
||||
# Act
|
||||
belief_setter_behaviour._process_belief_message(msg)
|
||||
|
||||
# Assert
|
||||
mock_set_beliefs.assert_called_once_with(beliefs_payload)
|
||||
|
||||
|
||||
def test_process_belief_message_invalid_json(belief_setter_behaviour, mocker, caplog):
|
||||
"""
|
||||
Test that a message with invalid JSON is handled gracefully and an error is logged.
|
||||
"""
|
||||
# Arrange
|
||||
msg = create_mock_message(
|
||||
sender_node=COLLECTOR_AGENT_JID, body="this is not a json string", thread="beliefs"
|
||||
)
|
||||
mock_set_beliefs = mocker.patch.object(belief_setter_behaviour, "_set_beliefs")
|
||||
|
||||
# Act
|
||||
belief_setter_behaviour._process_belief_message(msg)
|
||||
|
||||
# Assert
|
||||
mock_set_beliefs.assert_not_called()
|
||||
|
||||
|
||||
def test_process_belief_message_wrong_thread(belief_setter_behaviour, mocker):
|
||||
"""
|
||||
Test that a message with an incorrect thread is ignored.
|
||||
"""
|
||||
# Arrange
|
||||
msg = create_mock_message(
|
||||
sender_node=COLLECTOR_AGENT_JID, body='{"some": "data"}', thread="not_beliefs"
|
||||
)
|
||||
mock_set_beliefs = mocker.patch.object(belief_setter_behaviour, "_set_beliefs")
|
||||
|
||||
# Act
|
||||
belief_setter_behaviour._process_belief_message(msg)
|
||||
|
||||
# Assert
|
||||
mock_set_beliefs.assert_not_called()
|
||||
|
||||
|
||||
def test_process_belief_message_empty_body(belief_setter_behaviour, mocker):
|
||||
"""
|
||||
Test that a message with an empty body is ignored.
|
||||
"""
|
||||
# Arrange
|
||||
msg = create_mock_message(sender_node=COLLECTOR_AGENT_JID, body="", thread="beliefs")
|
||||
mock_set_beliefs = mocker.patch.object(belief_setter_behaviour, "_set_beliefs")
|
||||
|
||||
# Act
|
||||
belief_setter_behaviour._process_belief_message(msg)
|
||||
|
||||
# Assert
|
||||
mock_set_beliefs.assert_not_called()
|
||||
|
||||
|
||||
def test_set_beliefs_success(belief_setter_behaviour, mock_agent, caplog):
|
||||
"""
|
||||
Test that beliefs are correctly set on the agent's BDI.
|
||||
"""
|
||||
# Arrange
|
||||
beliefs_to_set = {
|
||||
"is_hot": ["kitchen"],
|
||||
"door_opened": ["front_door", "back_door"],
|
||||
}
|
||||
|
||||
# Act
|
||||
with caplog.at_level(logging.INFO):
|
||||
belief_setter_behaviour._set_beliefs(beliefs_to_set)
|
||||
|
||||
# Assert
|
||||
expected_calls = [
|
||||
call("is_hot", "kitchen"),
|
||||
call("door_opened", "front_door", "back_door"),
|
||||
]
|
||||
mock_agent.bdi.set_belief.assert_has_calls(expected_calls, any_order=True)
|
||||
assert mock_agent.bdi.set_belief.call_count == 2
|
||||
|
||||
|
||||
# def test_responded_unset(belief_setter_behaviour, mock_agent):
|
||||
# # Arrange
|
||||
# new_beliefs = {"user_said": ["message"]}
|
||||
#
|
||||
# # Act
|
||||
# belief_setter_behaviour._set_beliefs(new_beliefs)
|
||||
#
|
||||
# # Assert
|
||||
# mock_agent.bdi.set_belief.assert_has_calls([call("user_said", "message")])
|
||||
# mock_agent.bdi.remove_belief.assert_has_calls([call("responded")])
|
||||
|
||||
# def test_set_beliefs_bdi_not_initialized(belief_setter_behaviour, mock_agent, caplog):
|
||||
# """
|
||||
# Test that a warning is logged if the agent's BDI is not initialized.
|
||||
# """
|
||||
# # Arrange
|
||||
# mock_agent.bdi = None # Simulate BDI not being ready
|
||||
# beliefs_to_set = {"is_hot": ["kitchen"]}
|
||||
#
|
||||
# # Act
|
||||
# with caplog.at_level(logging.WARNING):
|
||||
# belief_setter_behaviour._set_beliefs(beliefs_to_set)
|
||||
#
|
||||
# # Assert
|
||||
# assert "Cannot set beliefs, since agent's BDI is not yet initialized." in caplog.text
|
||||
103
test/unit/agents/bdi/test_bdi_core_agent.py
Normal file
103
test/unit/agents/bdi/test_bdi_core_agent.py
Normal file
@@ -0,0 +1,103 @@
|
||||
from unittest.mock import AsyncMock, MagicMock, mock_open, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from control_backend.agents.bdi.bdi_core_agent.bdi_core_agent import BDICoreAgent
|
||||
from control_backend.core.agent_system import InternalMessage
|
||||
from control_backend.core.config import settings
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_agentspeak_env():
|
||||
with patch("agentspeak.runtime.Environment") as mock_env:
|
||||
yield mock_env
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def agent():
|
||||
agent = BDICoreAgent("bdi_agent", "dummy.asl")
|
||||
agent.send = AsyncMock()
|
||||
agent.bdi_agent = MagicMock()
|
||||
return agent
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_setup_loads_asl(mock_agentspeak_env, agent):
|
||||
# Mock file opening
|
||||
with patch("builtins.open", mock_open(read_data="+initial_goal.")):
|
||||
await agent.setup()
|
||||
|
||||
# Check if environment tried to build agent
|
||||
mock_agentspeak_env.return_value.build_agent.assert_called()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_setup_no_asl(mock_agentspeak_env, agent):
|
||||
with patch("builtins.open", side_effect=FileNotFoundError):
|
||||
await agent.setup()
|
||||
|
||||
mock_agentspeak_env.return_value.build_agent.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_belief_collector_message(agent):
|
||||
"""Test that incoming beliefs are added to the BDI agent"""
|
||||
# Simulate message from belief collector
|
||||
import json
|
||||
|
||||
beliefs = {"user_said": ["Hello"]}
|
||||
msg = InternalMessage(
|
||||
to="bdi_agent",
|
||||
sender=settings.agent_settings.bdi_belief_collector_name,
|
||||
body=json.dumps(beliefs),
|
||||
thread="beliefs",
|
||||
)
|
||||
|
||||
await agent.handle_message(msg)
|
||||
|
||||
# Expect bdi_agent.call to be triggered to add belief
|
||||
assert agent.bdi_agent.call.called
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_llm_response(agent):
|
||||
"""Test that LLM responses are forwarded to the Robot Speech Agent"""
|
||||
msg = InternalMessage(
|
||||
to="bdi_agent", sender=settings.agent_settings.llm_name, body="This is the LLM reply"
|
||||
)
|
||||
|
||||
await agent.handle_message(msg)
|
||||
|
||||
# Verify forward
|
||||
assert agent.send.called
|
||||
sent_msg = agent.send.call_args[0][0]
|
||||
assert sent_msg.to == settings.agent_settings.robot_speech_name
|
||||
assert "This is the LLM reply" in sent_msg.body
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_custom_actions(agent):
|
||||
agent._send_to_llm = MagicMock(side_effect=agent.send) # Mock specific method
|
||||
|
||||
# Initialize actions manually since we didn't call setup with real file
|
||||
agent._add_custom_actions()
|
||||
|
||||
# Find the action
|
||||
action_fn = None
|
||||
for (functor, _), fn in agent.actions.actions.items():
|
||||
if functor == ".reply":
|
||||
action_fn = fn
|
||||
break
|
||||
|
||||
assert action_fn is not None
|
||||
|
||||
# Invoke action
|
||||
mock_term = MagicMock()
|
||||
mock_term.args = ["Hello"]
|
||||
mock_intention = MagicMock()
|
||||
|
||||
# Run generator
|
||||
gen = action_fn(agent, mock_term, mock_intention)
|
||||
next(gen) # Execute
|
||||
|
||||
agent._send_to_llm.assert_called_with("Hello")
|
||||
@@ -7,16 +7,6 @@ from control_backend.agents.bdi.text_belief_extractor_agent.text_belief_extracto
|
||||
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
|
||||
@@ -40,29 +30,29 @@ async def test_handle_message_ignores_other_agents(agent):
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_message_from_transcriber(agent):
|
||||
async def test_handle_message_from_transcriber(agent, mock_settings):
|
||||
transcription = "hello world"
|
||||
msg = make_msg(settings.agent_settings.transcription_name, transcription, None)
|
||||
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 == settings.agent_settings.bdi_belief_collector_name
|
||||
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):
|
||||
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 == settings.agent_settings.bdi_belief_collector_name
|
||||
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]
|
||||
Reference in New Issue
Block a user