pepperplus/pepperplus-cb
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix: unit tests fixes and ruff formating

Browse Source 
N25B-205
This commit is contained in:
Björn Otgaar
2025-10-28 11:31:05 +01:00
parent 52faa59184
commit 47a87d0b4a
11 changed files with 307 additions and 209 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
test/integration/agents/test_ri_commands_agent.py
View File

@@ -6,16 +6,20 @@ from unittest.mock import AsyncMock, MagicMock, patch
from control_backend.agents.ri_command_agent import RICommandAgent
from control_backend.schemas.ri_message import SpeechCommand
@pytest.mark.asyncio
async def test_setup_bind(monkeypatch):
"""Test setup with bind=True"""
fake_socket = MagicMock()
monkeypatch.setattr("control_backend.agents.ri_command_agent.context.socket",
lambda _: fake_socket)
monkeypatch.setattr(
"control_backend.agents.ri_command_agent.context.socket", lambda _: fake_socket
)
agent = RICommandAgent("test@server", "password", address="tcp://localhost:5555", bind=True)
monkeypatch.setattr("control_backend.agents.ri_command_agent.settings",
MagicMock(zmq_settings=MagicMock(internal_comm_address="tcp://internal:1234")))
monkeypatch.setattr(
"control_backend.agents.ri_command_agent.settings",
MagicMock(zmq_settings=MagicMock(internal_comm_address="tcp://internal:1234")),
)
await agent.setup()
@@ -28,29 +32,35 @@ async def test_setup_bind(monkeypatch):
# Ensure behaviour attached
assert any(isinstance(b, agent.SendCommandsBehaviour) for b in agent.behaviours)
@pytest.mark.asyncio
async def test_setup_connect(monkeypatch):
"""Test setup with bind=False"""
fake_socket = MagicMock()
monkeypatch.setattr("control_backend.agents.ri_command_agent.context.socket",
lambda _: fake_socket)
monkeypatch.setattr(
"control_backend.agents.ri_command_agent.context.socket", lambda _: fake_socket
)
agent = RICommandAgent("test@server", "password", address="tcp://localhost:5555", bind=False)
monkeypatch.setattr("control_backend.agents.ri_command_agent.settings",
MagicMock(zmq_settings=MagicMock(internal_comm_address="tcp://internal:1234")))
monkeypatch.setattr(
"control_backend.agents.ri_command_agent.settings",
MagicMock(zmq_settings=MagicMock(internal_comm_address="tcp://internal:1234")),
)
await agent.setup()
# Ensure PUB socket connected
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.recv_multipart = AsyncMock(
return_value=(b"command", json.dumps(message_dict).encode("utf-8"))
)
fake_socket.send_json = AsyncMock()
agent = RICommandAgent("test@server", "password")
@@ -60,14 +70,15 @@ async def test_send_commands_behaviour_valid_message():
behaviour = agent.SendCommandsBehaviour()
behaviour.agent = agent
with patch('control_backend.agents.ri_command_agent.SpeechCommand') as MockSpeechCommand:
with patch("control_backend.agents.ri_command_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)
fake_socket.send_json.assert_awaited_with(mock_message.model_dump())
@pytest.mark.asyncio
async def test_send_commands_behaviour_invalid_message(caplog):