efe49c219c
Expanding abbreviations to remove ambiguity, simplifying agent names to reduce repetition. ref: N25B-257
99 lines
3.2 KiB
Python
99 lines
3.2 KiB
Python
import json
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
import zmq
|
|
|
|
from control_backend.agents.actuation.robot_speech_agent import RobotSpeechAgent
|
|
|
|
|
|
@pytest.fixture
|
|
def zmq_context(mocker):
|
|
mock_context = mocker.patch(
|
|
"control_backend.agents.actuation.robot_speech_agent.zmq.Context.instance"
|
|
)
|
|
mock_context.return_value = MagicMock()
|
|
return mock_context
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_setup_bind(zmq_context, mocker):
|
|
"""Test setup with bind=True"""
|
|
fake_socket = zmq_context.return_value.socket.return_value
|
|
|
|
agent = RobotSpeechAgent("test@server", "password", address="tcp://localhost:5555", bind=True)
|
|
settings = mocker.patch("control_backend.agents.actuation.robot_speech_agent.settings")
|
|
settings.zmq_settings.internal_sub_address = "tcp://internal:1234"
|
|
|
|
await agent.setup()
|
|
|
|
fake_socket.bind.assert_any_call("tcp://localhost:5555")
|
|
fake_socket.connect.assert_any_call("tcp://internal:1234")
|
|
fake_socket.setsockopt.assert_any_call(zmq.SUBSCRIBE, b"command")
|
|
|
|
# Ensure behaviour attached
|
|
assert any(isinstance(b, agent.SendZMQCommandsBehaviour) for b in agent.behaviours)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_setup_connect(zmq_context, mocker):
|
|
"""Test setup with bind=False"""
|
|
fake_socket = zmq_context.return_value.socket.return_value
|
|
|
|
agent = RobotSpeechAgent("test@server", "password", address="tcp://localhost:5555", bind=False)
|
|
settings = mocker.patch("control_backend.agents.actuation.robot_speech_agent.settings")
|
|
settings.zmq_settings.internal_sub_address = "tcp://internal:1234"
|
|
|
|
await agent.setup()
|
|
|
|
fake_socket.connect.assert_any_call("tcp://localhost:5555")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_commands_behaviour_valid_message():
|
|
"""Test behaviour with valid JSON message"""
|
|
fake_socket = AsyncMock()
|
|
message_dict = {"message": "hello"}
|
|
fake_socket.recv_multipart = AsyncMock(
|
|
return_value=(b"command", json.dumps(message_dict).encode("utf-8"))
|
|
)
|
|
fake_socket.send_json = AsyncMock()
|
|
|
|
agent = RobotSpeechAgent("test@server", "password")
|
|
agent.subsocket = fake_socket
|
|
agent.pubsocket = fake_socket
|
|
|
|
behaviour = agent.SendZMQCommandsBehaviour()
|
|
behaviour.agent = agent
|
|
|
|
with patch(
|
|
"control_backend.agents.actuation.robot_speech_agent.SpeechCommand"
|
|
) as MockSpeechCommand:
|
|
mock_message = MagicMock()
|
|
MockSpeechCommand.model_validate.return_value = mock_message
|
|
|
|
await behaviour.run()
|
|
|
|
fake_socket.recv_multipart.assert_awaited()
|
|
fake_socket.send_json.assert_awaited_with(mock_message.model_dump())
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_commands_behaviour_invalid_message():
|
|
"""Test behaviour with invalid JSON message triggers error logging"""
|
|
fake_socket = AsyncMock()
|
|
fake_socket.recv_multipart = AsyncMock(return_value=(b"command", b"{invalid_json}"))
|
|
fake_socket.send_json = AsyncMock()
|
|
|
|
agent = RobotSpeechAgent("test@server", "password")
|
|
agent.subsocket = fake_socket
|
|
agent.pubsocket = fake_socket
|
|
|
|
behaviour = agent.SendZMQCommandsBehaviour()
|
|
behaviour.agent = agent
|
|
|
|
await behaviour.run()
|
|
|
|
fake_socket.recv_multipart.assert_awaited()
|
|
fake_socket.send_json.assert_not_awaited()
|