Gestures in the CB. #36

Merged
9828273 merged 16 commits from feat/cb2ri-gestures into dev 2025-12-16 09:24:26 +00:00
13 changed files with 1085 additions and 43 deletions

View File

@@ -1 +1,2 @@
from .robot_gesture_agent import RobotGestureAgent as RobotGestureAgent
from .robot_speech_agent import RobotSpeechAgent as RobotSpeechAgent from .robot_speech_agent import RobotSpeechAgent as RobotSpeechAgent

View File

@@ -0,0 +1,162 @@
import json
import zmq
import zmq.asyncio as azmq
from control_backend.agents import BaseAgent
from control_backend.core.agent_system import InternalMessage
from control_backend.core.config import settings
from control_backend.schemas.ri_message import GestureCommand, RIEndpoint
class RobotGestureAgent(BaseAgent):
"""
This agent acts as a bridge between the control backend and the Robot Interface (RI).
It receives gesture commands from other agents or from the UI,
and forwards them to the robot via a ZMQ PUB socket.
:ivar subsocket: ZMQ SUB socket for receiving external commands (e.g., from UI).
:ivar pubsocket: ZMQ PUB socket for sending commands to the Robot Interface.
:ivar address: Address to bind/connect the PUB socket.
:ivar bind: Whether to bind or connect the PUB socket.
:ivar gesture_data: A list of strings for available gestures
"""
subsocket: azmq.Socket
repsocket: azmq.Socket
pubsocket: azmq.Socket
address = ""
bind = False
gesture_data = []
def __init__(
self,
name: str,
address=settings.zmq_settings.ri_command_address,
bind=False,
gesture_data=None,
):
self.gesture_data = gesture_data or []
super().__init__(name)
self.address = address
self.bind = bind
async def setup(self):
"""
Initialize the agent.
1. Sets up the PUB socket to talk to the robot.
2. Sets up the SUB socket to listen for "command" topics (from UI/External).
3. Starts the loop for handling ZMQ commands.
"""
self.logger.info("Setting up %s", self.name)
context = azmq.Context.instance()
# To the robot
self.pubsocket = context.socket(zmq.PUB)
if self.bind:
self.pubsocket.bind(self.address)
else:
self.pubsocket.connect(self.address)
# Receive internal topics regarding commands
self.subsocket = context.socket(zmq.SUB)
self.subsocket.connect(settings.zmq_settings.internal_sub_address)
self.subsocket.setsockopt(zmq.SUBSCRIBE, b"command")
self.subsocket.setsockopt(zmq.SUBSCRIBE, b"send_gestures")
# REP socket for replying to gesture requests
self.repsocket = context.socket(zmq.REP)
self.repsocket.bind(settings.zmq_settings.internal_gesture_rep_adress)
self.add_behavior(self._zmq_command_loop())
self.add_behavior(self._fetch_gestures_loop())
self.logger.info("Finished setting up %s", self.name)
async def stop(self):
if self.subsocket:
self.subsocket.close()
if self.pubsocket:
self.pubsocket.close()
await super().stop()
async def handle_message(self, msg: InternalMessage):
"""
Handle commands received from other internal Python agents.
Validates the message as a :class:`GestureCommand` and forwards it to the robot.
:param msg: The internal message containing the command.
"""
try:
gesture_command = GestureCommand.model_validate_json(msg.body)
if gesture_command.endpoint == RIEndpoint.GESTURE_TAG:
if gesture_command.data not in self.gesture_data:
self.logger.warning(
"Received gesture tag '%s' which is not in available tags. Early returning",
gesture_command.data,
)
return
await self.pubsocket.send_json(gesture_command.model_dump())
except Exception:
self.logger.exception("Error processing internal message.")
async def _zmq_command_loop(self):
"""
Loop to handle commands received via ZMQ (e.g., from the UI).
Listens on the 'command' topic, validates the JSON and forwards it to the robot.
"""
while self._running:
try:
topic, body = await self.subsocket.recv_multipart()
# Don't process send_gestures here
if topic != b"command":
continue
body = json.loads(body)
gesture_command = GestureCommand.model_validate(body)
if gesture_command.endpoint == RIEndpoint.GESTURE_TAG:
if gesture_command.data not in self.gesture_data:
self.logger.warning(
"Received gesture tag '%s' which is not in available tags.\
Early returning",
gesture_command.data,
)
continue
await self.pubsocket.send_json(gesture_command.model_dump())
except Exception:
self.logger.exception("Error processing ZMQ message.")
async def _fetch_gestures_loop(self):
"""
Loop to handle fetching gestures received via ZMQ (e.g., from the UI).
Listens on the 'send_gestures' topic, and returns a list on the get_gestures topic.
"""
while self._running:
try:
# Get a request
body = await self.repsocket.recv()
# Figure out amount, if specified
try:
body = json.loads(body)
except json.JSONDecodeError:
body = None
amount = None
if isinstance(body, int):
amount = body
# Fetch tags from gesture data and respond
tags = self.gesture_data[:amount] if amount else self.gesture_data
response = json.dumps({"tags": tags}).encode()
await self.repsocket.send(response)
except Exception:
self.logger.exception("Error fetching gesture tags.")

View File

@@ -29,7 +29,7 @@ class RobotSpeechAgent(BaseAgent):
def __init__( def __init__(
self, self,
name: str, name: str,
address=settings.zmq_settings.ri_command_address, address: str,
bind=False, bind=False,
): ):
super().__init__(name) super().__init__(name)

View File

@@ -6,6 +6,7 @@ import zmq.asyncio as azmq
from zmq.asyncio import Context from zmq.asyncio import Context
from control_backend.agents import BaseAgent from control_backend.agents import BaseAgent
from control_backend.agents.actuation.robot_gesture_agent import RobotGestureAgent
from control_backend.core.config import settings from control_backend.core.config import settings
from ..actuation.robot_speech_agent import RobotSpeechAgent from ..actuation.robot_speech_agent import RobotSpeechAgent
@@ -180,12 +181,21 @@ class RICommunicationAgent(BaseAgent):
else: else:
self._req_socket.bind(addr) self._req_socket.bind(addr)
case "actuation": case "actuation":
ri_commands_agent = RobotSpeechAgent( gesture_data = port_data.get("gestures", [])
robot_speech_agent = RobotSpeechAgent(
settings.agent_settings.robot_speech_name, settings.agent_settings.robot_speech_name,
address=addr, address=addr,
bind=bind, bind=bind,
) )
await ri_commands_agent.start() robot_gesture_agent = RobotGestureAgent(
settings.agent_settings.robot_gesture_name,
address=addr,
bind=bind,
gesture_data=gesture_data,
)
await robot_speech_agent.start()
await asyncio.sleep(0.1) # Small delay
await robot_gesture_agent.start()
case "audio": case "audio":
vad_agent = VADAgent(audio_in_address=addr, audio_in_bind=bind) vad_agent = VADAgent(audio_in_address=addr, audio_in_bind=bind)
await vad_agent.start() await vad_agent.start()

View File

@@ -8,15 +8,15 @@ from fastapi.responses import StreamingResponse
from zmq.asyncio import Context, Socket from zmq.asyncio import Context, Socket
from control_backend.core.config import settings from control_backend.core.config import settings
from control_backend.schemas.ri_message import SpeechCommand from control_backend.schemas.ri_message import GestureCommand, SpeechCommand
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
router = APIRouter() router = APIRouter()
@router.post("/command", status_code=202) @router.post("/command/speech", status_code=202)
async def receive_command(command: SpeechCommand, request: Request): async def receive_command_speech(command: SpeechCommand, request: Request):
""" """
Send a direct speech command to the robot. Send a direct speech command to the robot.
@@ -27,14 +27,32 @@ async def receive_command(command: SpeechCommand, request: Request):
:param command: The speech command payload. :param command: The speech command payload.
:param request: The FastAPI request object. :param request: The FastAPI request object.
""" """
# Validate and retrieve data.
SpeechCommand.model_validate(command)
topic = b"command" topic = b"command"
pub_socket: Socket = request.app.state.endpoints_pub_socket pub_socket: Socket = request.app.state.endpoints_pub_socket
await pub_socket.send_multipart([topic, command.model_dump_json().encode()]) await pub_socket.send_multipart([topic, command.model_dump_json().encode()])
return {"status": "Command received"} return {"status": "Speech command received"}
@router.post("/command/gesture", status_code=202)
async def receive_command_gesture(command: GestureCommand, request: Request):
"""
Send a direct gesture command to the robot.
Publishes the command to the internal 'command' topic. The
:class:`~control_backend.agents.actuation.robot_speech_agent.RobotGestureAgent`
will forward this to the robot.
:param command: The speech command payload.
:param request: The FastAPI request object.
"""
topic = b"command"
pub_socket: Socket = request.app.state.endpoints_pub_socket
await pub_socket.send_multipart([topic, command.model_dump_json().encode()])
return {"status": "Gesture command received"}
@router.get("/ping_check") @router.get("/ping_check")
@@ -45,6 +63,41 @@ async def ping(request: Request):
pass pass
@router.get("/commands/gesture/tags")
async def get_available_gesture_tags(request: Request, count=0):
"""
Endpoint to retrieve the available gesture tags for the robot.
:param request: The FastAPI request object.
:return: A list of available gesture tags.
"""
req_socket = Context.instance().socket(zmq.REQ)
req_socket.connect(settings.zmq_settings.internal_gesture_rep_adress)
# Check to see if we've got any count given in the query parameter
amount = count or None
timeout = 5 # seconds
await req_socket.send(f"{amount}".encode() if amount else b"None")
try:
body = await asyncio.wait_for(req_socket.recv(), timeout=timeout)
except TimeoutError:
body = '{"tags": []}'
logger.debug("Got timeout error fetching gestures.")
# Handle empty response and JSON decode errors
available_tags = []
if body:
try:
available_tags = json.loads(body).get("tags", [])
except json.JSONDecodeError as e:
logger.error(f"Failed to parse gesture tags JSON: {e}, body: {body}")
# Return empty list on JSON error
available_tags = []
return {"available_gesture_tags": available_tags}
@router.get("/ping_stream") @router.get("/ping_stream")
async def ping_stream(request: Request): async def ping_stream(request: Request):
""" """

View File

@@ -17,6 +17,7 @@ class ZMQSettings(BaseModel):
internal_sub_address: str = "tcp://localhost:5561" internal_sub_address: str = "tcp://localhost:5561"
ri_command_address: str = "tcp://localhost:0000" ri_command_address: str = "tcp://localhost:0000"
ri_communication_address: str = "tcp://*:5555" ri_communication_address: str = "tcp://*:5555"
internal_gesture_rep_adress: str = "tcp://localhost:7788"
class AgentSettings(BaseModel): class AgentSettings(BaseModel):
@@ -46,6 +47,7 @@ class AgentSettings(BaseModel):
transcription_name: str = "transcription_agent" transcription_name: str = "transcription_agent"
ri_communication_name: str = "ri_communication_agent" ri_communication_name: str = "ri_communication_agent"
robot_speech_name: str = "robot_speech_agent" robot_speech_name: str = "robot_speech_agent"
robot_gesture_name: str = "robot_gesture_agent"
class BehaviourSettings(BaseModel): class BehaviourSettings(BaseModel):

View File

@@ -1,7 +1,7 @@
from enum import Enum from enum import Enum
from typing import Any from typing import Any, Literal
from pydantic import BaseModel from pydantic import BaseModel, model_validator
class RIEndpoint(str, Enum): class RIEndpoint(str, Enum):
@@ -10,6 +10,8 @@ class RIEndpoint(str, Enum):
""" """
SPEECH = "actuate/speech" SPEECH = "actuate/speech"
GESTURE_SINGLE = "actuate/gesture/single"
GESTURE_TAG = "actuate/gesture/tag"
PING = "ping" PING = "ping"
NEGOTIATE_PORTS = "negotiate/ports" NEGOTIATE_PORTS = "negotiate/ports"
@@ -36,3 +38,27 @@ class SpeechCommand(RIMessage):
endpoint: RIEndpoint = RIEndpoint(RIEndpoint.SPEECH) endpoint: RIEndpoint = RIEndpoint(RIEndpoint.SPEECH)
data: str data: str
class GestureCommand(RIMessage):
"""
A specific command to make the robot do a gesture.
:ivar endpoint: Should be ``RIEndpoint.GESTURE_SINGLE`` or ``RIEndpoint.GESTURE_TAG``.
:ivar data: The id of the gesture to be executed.
"""
endpoint: Literal[ # pyright: ignore[reportIncompatibleVariableOverride] - We validate this stricter rule ourselves
RIEndpoint.GESTURE_SINGLE, RIEndpoint.GESTURE_TAG
]
data: str
@model_validator(mode="after")
def check_endpoint(self):
allowed = {
RIEndpoint.GESTURE_SINGLE,
RIEndpoint.GESTURE_TAG,
}
if self.endpoint not in allowed:
raise ValueError("endpoint must be GESTURE_SINGLE or GESTURE_TAG")
return self

View File

@@ -0,0 +1,444 @@
import json
from unittest.mock import AsyncMock, MagicMock
import pytest
import zmq
from control_backend.agents.actuation.robot_gesture_agent import RobotGestureAgent
from control_backend.core.agent_system import InternalMessage
from control_backend.schemas.ri_message import RIEndpoint
@pytest.fixture
def zmq_context(mocker):
"""Mock the ZMQ context."""
mock_context = mocker.patch(
"control_backend.agents.actuation.robot_gesture_agent.azmq.Context.instance"
)
mock_context.return_value = MagicMock()
return mock_context
@pytest.mark.asyncio
async def test_setup_bind(zmq_context, mocker):
"""Setup binds and subscribes to internal commands."""
fake_socket = zmq_context.return_value.socket.return_value
agent = RobotGestureAgent("robot_gesture", address="tcp://localhost:5556", bind=True)
settings = mocker.patch("control_backend.agents.actuation.robot_gesture_agent.settings")
settings.zmq_settings.internal_sub_address = "tcp://internal:1234"
agent.add_behavior = MagicMock()
await agent.setup()
# Check PUB socket binding
fake_socket.bind.assert_any_call("tcp://localhost:5556")
# Check REP socket binding
fake_socket.bind.assert_called()
# Check SUB socket connection and subscriptions
fake_socket.connect.assert_any_call("tcp://internal:1234")
fake_socket.setsockopt.assert_any_call(zmq.SUBSCRIBE, b"command")
fake_socket.setsockopt.assert_any_call(zmq.SUBSCRIBE, b"send_gestures")
# Check behavior was added (twice: once for command loop, once for fetch gestures loop)
assert agent.add_behavior.call_count == 2
@pytest.mark.asyncio
async def test_setup_connect(zmq_context, mocker):
"""Setup connects when bind=False."""
fake_socket = zmq_context.return_value.socket.return_value
agent = RobotGestureAgent("robot_gesture", address="tcp://localhost:5556", bind=False)
settings = mocker.patch("control_backend.agents.actuation.robot_gesture_agent.settings")
settings.zmq_settings.internal_sub_address = "tcp://internal:1234"
agent.add_behavior = MagicMock()
await agent.setup()
# Check PUB socket connection (not binding)
fake_socket.connect.assert_any_call("tcp://localhost:5556")
fake_socket.connect.assert_any_call("tcp://internal:1234")
# Check REP socket binding (always binds)
fake_socket.bind.assert_called()
# Check behavior was added (twice)
assert agent.add_behavior.call_count == 2
@pytest.mark.asyncio
async def test_handle_message_sends_valid_gesture_command():
"""Internal message with valid gesture tag is forwarded to robot pub socket."""
pubsocket = AsyncMock()
agent = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"])
agent.pubsocket = pubsocket
payload = {
"endpoint": RIEndpoint.GESTURE_TAG,
"data": "hello", # "hello" is in gesture_data
}
msg = InternalMessage(to="robot", sender="tester", body=json.dumps(payload))
await agent.handle_message(msg)
pubsocket.send_json.assert_awaited_once()
@pytest.mark.asyncio
async def test_handle_message_sends_non_gesture_command():
"""Internal message with non-gesture endpoint is not forwarded by this agent."""
pubsocket = AsyncMock()
agent = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"])
agent.pubsocket = pubsocket
payload = {"endpoint": "some_other_endpoint", "data": "invalid_tag_not_in_list"}
msg = InternalMessage(to="robot", sender="tester", body=json.dumps(payload))
await agent.handle_message(msg)
# Non-gesture endpoints should not be forwarded by this agent
pubsocket.send_json.assert_not_awaited()
@pytest.mark.asyncio
async def test_handle_message_rejects_invalid_gesture_tag():
"""Internal message with invalid gesture tag is not forwarded."""
pubsocket = AsyncMock()
agent = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"])
agent.pubsocket = pubsocket
# Use a tag that's not in gesture_data
payload = {"endpoint": RIEndpoint.GESTURE_TAG, "data": "invalid_tag_not_in_list"}
msg = InternalMessage(to="robot", sender="tester", body=json.dumps(payload))
await agent.handle_message(msg)
pubsocket.send_json.assert_not_awaited()
@pytest.mark.asyncio
async def test_handle_message_invalid_payload():
"""Invalid payload is caught and does not send."""
pubsocket = AsyncMock()
agent = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"])
agent.pubsocket = pubsocket
msg = InternalMessage(to="robot", sender="tester", body=json.dumps({"bad": "data"}))
await agent.handle_message(msg)
pubsocket.send_json.assert_not_awaited()
@pytest.mark.asyncio
async def test_zmq_command_loop_valid_gesture_payload():
"""UI command with valid gesture tag is read from SUB and published."""
command = {"endpoint": RIEndpoint.GESTURE_TAG, "data": "hello"}
fake_socket = AsyncMock()
async def recv_once():
# stop after first iteration
agent._running = False
return (b"command", json.dumps(command).encode("utf-8"))
fake_socket.recv_multipart = recv_once
fake_socket.send_json = AsyncMock()
agent = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"])
agent.subsocket = fake_socket
agent.pubsocket = fake_socket
agent._running = True
await agent._zmq_command_loop()
fake_socket.send_json.assert_awaited_once()
@pytest.mark.asyncio
async def test_zmq_command_loop_valid_non_gesture_payload():
"""UI command with non-gesture endpoint is not forwarded by this agent."""
command = {"endpoint": "some_other_endpoint", "data": "anything"}
fake_socket = AsyncMock()
async def recv_once():
agent._running = False
return (b"command", json.dumps(command).encode("utf-8"))
fake_socket.recv_multipart = recv_once
fake_socket.send_json = AsyncMock()
agent = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"])
agent.subsocket = fake_socket
agent.pubsocket = fake_socket
agent._running = True
await agent._zmq_command_loop()
fake_socket.send_json.assert_not_awaited()
@pytest.mark.asyncio
async def test_zmq_command_loop_invalid_gesture_tag():
"""UI command with invalid gesture tag is not forwarded."""
command = {"endpoint": RIEndpoint.GESTURE_TAG, "data": "invalid_tag_not_in_list"}
fake_socket = AsyncMock()
async def recv_once():
agent._running = False
return (b"command", json.dumps(command).encode("utf-8"))
fake_socket.recv_multipart = recv_once
fake_socket.send_json = AsyncMock()
agent = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"])
agent.subsocket = fake_socket
agent.pubsocket = fake_socket
agent._running = True
await agent._zmq_command_loop()
fake_socket.send_json.assert_not_awaited()
@pytest.mark.asyncio
async def test_zmq_command_loop_invalid_json():
"""Invalid JSON is ignored without sending."""
fake_socket = AsyncMock()
async def recv_once():
agent._running = False
return (b"command", b"{not_json}")
fake_socket.recv_multipart = recv_once
fake_socket.send_json = AsyncMock()
agent = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"])
agent.subsocket = fake_socket
agent.pubsocket = fake_socket
agent._running = True
await agent._zmq_command_loop()
fake_socket.send_json.assert_not_awaited()
@pytest.mark.asyncio
async def test_zmq_command_loop_ignores_send_gestures_topic():
"""send_gestures topic is ignored in command loop."""
fake_socket = AsyncMock()
async def recv_once():
agent._running = False
return (b"send_gestures", b"{}")
fake_socket.recv_multipart = recv_once
fake_socket.send_json = AsyncMock()
agent = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"])
agent.subsocket = fake_socket
agent.pubsocket = fake_socket
agent._running = True
await agent._zmq_command_loop()
fake_socket.send_json.assert_not_awaited()
@pytest.mark.asyncio
async def test_fetch_gestures_loop_without_amount():
"""Fetch gestures request without amount returns all tags."""
fake_repsocket = AsyncMock()
async def recv_once():
agent._running = False
return b"{}" # Empty JSON request
fake_repsocket.recv = recv_once
fake_repsocket.send = AsyncMock()
agent = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no", "wave", "point"])
agent.repsocket = fake_repsocket
agent._running = True
await agent._fetch_gestures_loop()
fake_repsocket.send.assert_awaited_once()
# Check the response contains all tags
args, kwargs = fake_repsocket.send.call_args
response = json.loads(args[0])
assert "tags" in response
assert response["tags"] == ["hello", "yes", "no", "wave", "point"]
@pytest.mark.asyncio
async def test_fetch_gestures_loop_with_amount():
"""Fetch gestures request with amount returns limited tags."""
fake_repsocket = AsyncMock()
amount = 3
async def recv_once():
agent._running = False
return json.dumps(amount).encode()
fake_repsocket.recv = recv_once
fake_repsocket.send = AsyncMock()
agent = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no", "wave", "point"])
agent.repsocket = fake_repsocket
agent._running = True
await agent._fetch_gestures_loop()
fake_repsocket.send.assert_awaited_once()
args, kwargs = fake_repsocket.send.call_args
response = json.loads(args[0])
assert "tags" in response
assert len(response["tags"]) == amount
assert response["tags"] == ["hello", "yes", "no"]
@pytest.mark.asyncio
async def test_fetch_gestures_loop_with_integer_request():
"""Fetch gestures request with integer amount."""
fake_repsocket = AsyncMock()
amount = 2
async def recv_once():
agent._running = False
return json.dumps(amount).encode()
fake_repsocket.recv = recv_once
fake_repsocket.send = AsyncMock()
agent = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"])
agent.repsocket = fake_repsocket
agent._running = True
await agent._fetch_gestures_loop()
fake_repsocket.send.assert_awaited_once()
args, kwargs = fake_repsocket.send.call_args
response = json.loads(args[0])
assert response["tags"] == ["hello", "yes"]
@pytest.mark.asyncio
async def test_fetch_gestures_loop_with_invalid_json():
"""Invalid JSON request returns all tags."""
fake_repsocket = AsyncMock()
async def recv_once():
agent._running = False
return b"not_json"
fake_repsocket.recv = recv_once
fake_repsocket.send = AsyncMock()
agent = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"])
agent.repsocket = fake_repsocket
agent._running = True
await agent._fetch_gestures_loop()
fake_repsocket.send.assert_awaited_once()
args, kwargs = fake_repsocket.send.call_args
response = json.loads(args[0])
assert response["tags"] == ["hello", "yes", "no"]
@pytest.mark.asyncio
async def test_fetch_gestures_loop_with_non_integer_json():
"""Non-integer JSON request returns all tags."""
fake_repsocket = AsyncMock()
async def recv_once():
agent._running = False
return json.dumps({"not": "an_integer"}).encode()
fake_repsocket.recv = recv_once
fake_repsocket.send = AsyncMock()
agent = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"])
agent.repsocket = fake_repsocket
agent._running = True
await agent._fetch_gestures_loop()
fake_repsocket.send.assert_awaited_once()
args, kwargs = fake_repsocket.send.call_args
response = json.loads(args[0])
assert response["tags"] == ["hello", "yes", "no"]
def test_gesture_data_attribute():
"""Test that gesture_data returns the expected list."""
gesture_data = ["hello", "yes", "no", "wave"]
agent = RobotGestureAgent("robot_gesture", gesture_data=gesture_data)
assert agent.gesture_data == gesture_data
assert isinstance(agent.gesture_data, list)
assert len(agent.gesture_data) == 4
assert "hello" in agent.gesture_data
assert "yes" in agent.gesture_data
assert "no" in agent.gesture_data
assert "invalid_tag_not_in_list" not in agent.gesture_data
@pytest.mark.asyncio
async def test_stop_closes_sockets():
"""Stop method closes all sockets."""
pubsocket = MagicMock()
subsocket = MagicMock()
repsocket = MagicMock()
agent = RobotGestureAgent("robot_gesture")
agent.pubsocket = pubsocket
agent.subsocket = subsocket
agent.repsocket = repsocket
await agent.stop()
pubsocket.close.assert_called_once()
subsocket.close.assert_called_once()
# Note: repsocket is not closed in stop() method, but you might want to add it
# repsocket.close.assert_called_once()
@pytest.mark.asyncio
async def test_initialization_with_custom_gesture_data():
"""Agent can be initialized with custom gesture data."""
custom_gestures = ["custom1", "custom2", "custom3"]
agent = RobotGestureAgent("robot_gesture", gesture_data=custom_gestures)
assert agent.gesture_data == custom_gestures
@pytest.mark.asyncio
async def test_fetch_gestures_loop_handles_exception():
"""Exception in fetch gestures loop is caught and logged."""
fake_repsocket = AsyncMock()
async def recv_once():
agent._running = False
raise Exception("Test exception")
fake_repsocket.recv = recv_once
fake_repsocket.send = AsyncMock()
agent = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"])
agent.repsocket = fake_repsocket
agent.logger = MagicMock()
agent._running = True
# Should not raise exception
await agent._fetch_gestures_loop()
# Exception should be logged
agent.logger.exception.assert_called_once()

View File

@@ -8,6 +8,11 @@ from control_backend.agents.actuation.robot_speech_agent import RobotSpeechAgent
from control_backend.core.agent_system import InternalMessage from control_backend.core.agent_system import InternalMessage
def mock_speech_agent():
agent = RobotSpeechAgent("robot_speech", address="tcp://localhost:5555", bind=False)
return agent
@pytest.fixture @pytest.fixture
def zmq_context(mocker): def zmq_context(mocker):
mock_context = mocker.patch( mock_context = mocker.patch(
@@ -56,7 +61,7 @@ async def test_setup_connect(zmq_context, mocker):
async def test_handle_message_sends_command(): async def test_handle_message_sends_command():
"""Internal message is forwarded to robot pub socket as JSON.""" """Internal message is forwarded to robot pub socket as JSON."""
pubsocket = AsyncMock() pubsocket = AsyncMock()
agent = RobotSpeechAgent("robot_speech") agent = mock_speech_agent()
agent.pubsocket = pubsocket agent.pubsocket = pubsocket
payload = {"endpoint": "actuate/speech", "data": "hello"} payload = {"endpoint": "actuate/speech", "data": "hello"}
@@ -80,7 +85,7 @@ async def test_zmq_command_loop_valid_payload(zmq_context):
fake_socket.recv_multipart = recv_once fake_socket.recv_multipart = recv_once
fake_socket.send_json = AsyncMock() fake_socket.send_json = AsyncMock()
agent = RobotSpeechAgent("robot_speech") agent = mock_speech_agent()
agent.subsocket = fake_socket agent.subsocket = fake_socket
agent.pubsocket = fake_socket agent.pubsocket = fake_socket
agent._running = True agent._running = True
@@ -101,7 +106,7 @@ async def test_zmq_command_loop_invalid_json():
fake_socket.recv_multipart = recv_once fake_socket.recv_multipart = recv_once
fake_socket.send_json = AsyncMock() fake_socket.send_json = AsyncMock()
agent = RobotSpeechAgent("robot_speech") agent = mock_speech_agent()
agent.subsocket = fake_socket agent.subsocket = fake_socket
agent.pubsocket = fake_socket agent.pubsocket = fake_socket
agent._running = True agent._running = True
@@ -115,7 +120,7 @@ async def test_zmq_command_loop_invalid_json():
async def test_handle_message_invalid_payload(): async def test_handle_message_invalid_payload():
"""Invalid payload is caught and does not send.""" """Invalid payload is caught and does not send."""
pubsocket = AsyncMock() pubsocket = AsyncMock()
agent = RobotSpeechAgent("robot_speech") agent = mock_speech_agent()
agent.pubsocket = pubsocket agent.pubsocket = pubsocket
msg = InternalMessage(to="robot", sender="tester", body=json.dumps({"bad": "data"})) msg = InternalMessage(to="robot", sender="tester", body=json.dumps({"bad": "data"}))
@@ -129,7 +134,7 @@ async def test_handle_message_invalid_payload():
async def test_stop_closes_sockets(): async def test_stop_closes_sockets():
pubsocket = MagicMock() pubsocket = MagicMock()
subsocket = MagicMock() subsocket = MagicMock()
agent = RobotSpeechAgent("robot_speech") agent = mock_speech_agent()
agent.pubsocket = pubsocket agent.pubsocket = pubsocket
agent.subsocket = subsocket agent.subsocket = subsocket

View File

@@ -10,6 +10,10 @@ def speech_agent_path():
return "control_backend.agents.communication.ri_communication_agent.RobotSpeechAgent" return "control_backend.agents.communication.ri_communication_agent.RobotSpeechAgent"
def gesture_agent_path():
return "control_backend.agents.communication.ri_communication_agent.RobotGestureAgent"
@pytest.fixture @pytest.fixture
def zmq_context(mocker): def zmq_context(mocker):
mock_context = mocker.patch( mock_context = mocker.patch(
@@ -22,7 +26,7 @@ def zmq_context(mocker):
def negotiation_message( def negotiation_message(
actuation_port: int = 5556, actuation_port: int = 5556,
bind_main: bool = False, bind_main: bool = False,
bind_actuation: bool = True, bind_actuation: bool = False,
main_port: int = 5555, main_port: int = 5555,
): ):
return { return {
@@ -41,9 +45,12 @@ async def test_setup_success_connects_and_starts_robot(zmq_context):
fake_socket.recv_json = AsyncMock(return_value=negotiation_message()) fake_socket.recv_json = AsyncMock(return_value=negotiation_message())
fake_socket.send_multipart = AsyncMock() fake_socket.send_multipart = AsyncMock()
with patch(speech_agent_path(), autospec=True) as MockRobot: with (
robot_instance = MockRobot.return_value patch(speech_agent_path(), autospec=True) as MockSpeech,
robot_instance.start = AsyncMock() patch(gesture_agent_path(), autospec=True) as MockGesture,
):
MockSpeech.return_value.start = AsyncMock()
MockGesture.return_value.start = AsyncMock()
agent = RICommunicationAgent("ri_comm", address="tcp://localhost:5555", bind=False) agent = RICommunicationAgent("ri_comm", address="tcp://localhost:5555", bind=False)
agent.add_behavior = MagicMock() agent.add_behavior = MagicMock()
@@ -52,9 +59,17 @@ async def test_setup_success_connects_and_starts_robot(zmq_context):
fake_socket.connect.assert_any_call("tcp://localhost:5555") fake_socket.connect.assert_any_call("tcp://localhost:5555")
fake_socket.send_json.assert_any_call({"endpoint": "negotiate/ports", "data": {}}) fake_socket.send_json.assert_any_call({"endpoint": "negotiate/ports", "data": {}})
robot_instance.start.assert_awaited_once() MockSpeech.return_value.start.assert_awaited_once()
MockRobot.assert_called_once_with(ANY, address="tcp://*:5556", bind=True) MockGesture.return_value.start.assert_awaited_once()
MockSpeech.assert_called_once_with(ANY, address="tcp://localhost:5556", bind=False)
MockGesture.assert_called_once_with(
ANY,
address="tcp://localhost:5556",
bind=False,
gesture_data=[],
)
agent.add_behavior.assert_called_once() agent.add_behavior.assert_called_once()
assert agent.connected is True assert agent.connected is True
@@ -69,10 +84,13 @@ async def test_setup_binds_when_requested(zmq_context):
agent.add_behavior = MagicMock() agent.add_behavior = MagicMock()
with patch(speech_agent_path(), autospec=True) as MockRobot: with (
MockRobot.return_value.start = AsyncMock() patch(speech_agent_path(), autospec=True) as MockSpeech,
patch(gesture_agent_path(), autospec=True) as MockGesture,
):
MockSpeech.return_value.start = AsyncMock()
MockGesture.return_value.start = AsyncMock()
await agent.setup() await agent.setup()
fake_socket.bind.assert_any_call("tcp://localhost:5555") fake_socket.bind.assert_any_call("tcp://localhost:5555")
agent.add_behavior.assert_called_once() agent.add_behavior.assert_called_once()
@@ -88,7 +106,6 @@ async def test_negotiate_invalid_endpoint_retries(zmq_context):
agent._req_socket = fake_socket agent._req_socket = fake_socket
success = await agent._negotiate_connection(max_retries=1) success = await agent._negotiate_connection(max_retries=1)
assert success is False assert success is False
@@ -112,8 +129,12 @@ async def test_handle_negotiation_response_updates_req_socket(zmq_context):
fake_socket = zmq_context.return_value.socket.return_value fake_socket = zmq_context.return_value.socket.return_value
agent = RICommunicationAgent("ri_comm", address="tcp://localhost:5555", bind=False) agent = RICommunicationAgent("ri_comm", address="tcp://localhost:5555", bind=False)
agent._req_socket = fake_socket agent._req_socket = fake_socket
with patch(speech_agent_path(), autospec=True) as MockRobot: with (
MockRobot.return_value.start = AsyncMock() patch(speech_agent_path(), autospec=True) as MockSpeech,
patch(gesture_agent_path(), autospec=True) as MockGesture,
):
MockSpeech.return_value.start = AsyncMock()
MockGesture.return_value.start = AsyncMock()
await agent._handle_negotiation_response( await agent._handle_negotiation_response(
negotiation_message( negotiation_message(
main_port=6000, main_port=6000,
@@ -135,7 +156,6 @@ async def test_handle_disconnection_publishes_and_reconnects():
agent._negotiate_connection = AsyncMock(return_value=True) agent._negotiate_connection = AsyncMock(return_value=True)
await agent._handle_disconnection() await agent._handle_disconnection()
pub_socket.send_multipart.assert_awaited() pub_socket.send_multipart.assert_awaited()
assert agent.connected is True assert agent.connected is True
@@ -192,7 +212,7 @@ async def test_setup_warns_on_failed_negotiate(zmq_context, mocker):
fake_socket.recv_json = AsyncMock() fake_socket.recv_json = AsyncMock()
agent = RICommunicationAgent("ri_comm") agent = RICommunicationAgent("ri_comm")
async def swallow(coro): def swallow(coro):
coro.close() coro.close()
agent.add_behavior = swallow agent.add_behavior = swallow

View File

@@ -1,12 +1,14 @@
# tests/test_robot_endpoints.py
import json import json
from unittest.mock import AsyncMock, MagicMock from unittest.mock import AsyncMock, MagicMock, patch
import pytest import pytest
import zmq.asyncio
from fastapi import FastAPI from fastapi import FastAPI
from fastapi.testclient import TestClient from fastapi.testclient import TestClient
from control_backend.api.v1.endpoints import robot from control_backend.api.v1.endpoints import robot
from control_backend.schemas.ri_message import SpeechCommand from control_backend.schemas.ri_message import GestureCommand, SpeechCommand
@pytest.fixture @pytest.fixture
@@ -26,7 +28,27 @@ def client(app):
return TestClient(app) return TestClient(app)
def test_receive_command_success(client): @pytest.fixture
def mock_zmq_context():
"""Mock the ZMQ context used by the endpoint module."""
with patch("control_backend.api.v1.endpoints.robot.Context.instance") as mock_context:
context_instance = MagicMock()
mock_context.return_value = context_instance
yield context_instance
@pytest.fixture
def mock_sockets(mock_zmq_context):
"""Optional helper if you want both a sub and req/push socket available."""
mock_sub_socket = AsyncMock(spec=zmq.asyncio.Socket)
mock_req_socket = AsyncMock(spec=zmq.asyncio.Socket)
mock_zmq_context.socket.return_value = mock_sub_socket
return {"sub": mock_sub_socket, "req": mock_req_socket}
def test_receive_speech_command_success(client):
""" """
Test for successful reception of a command. Ensures the status code is 202 and the response body Test for successful reception of a command. Ensures the status code is 202 and the response body
is correct. It also verifies that the ZeroMQ socket's send_multipart method is called with the is correct. It also verifies that the ZeroMQ socket's send_multipart method is called with the
@@ -40,11 +62,11 @@ def test_receive_command_success(client):
speech_command = SpeechCommand(**command_data) speech_command = SpeechCommand(**command_data)
# Act # Act
response = client.post("/command", json=command_data) response = client.post("/command/speech", json=command_data)
# Assert # Assert
assert response.status_code == 202 assert response.status_code == 202
assert response.json() == {"status": "Command received"} assert response.json() == {"status": "Speech command received"}
# Verify that the ZMQ socket was used correctly # Verify that the ZMQ socket was used correctly
mock_pub_socket.send_multipart.assert_awaited_once_with( mock_pub_socket.send_multipart.assert_awaited_once_with(
@@ -52,13 +74,48 @@ def test_receive_command_success(client):
) )
def test_receive_command_invalid_payload(client): def test_receive_gesture_command_success(client):
"""
Test for successful reception of a command that is a gesture command.
Ensures the status code is 202 and the response body is correct.
"""
# Arrange
mock_pub_socket = AsyncMock()
client.app.state.endpoints_pub_socket = mock_pub_socket
command_data = {"endpoint": "actuate/gesture/tag", "data": "happy"}
gesture_command = GestureCommand(**command_data)
# Act
response = client.post("/command/gesture", json=command_data)
# Assert
assert response.status_code == 202
assert response.json() == {"status": "Gesture command received"}
# Verify that the ZMQ socket was used correctly
mock_pub_socket.send_multipart.assert_awaited_once_with(
[b"command", gesture_command.model_dump_json().encode()]
)
def test_receive_speech_command_invalid_payload(client):
""" """
Test invalid data handling (schema validation). Test invalid data handling (schema validation).
""" """
# Missing required field(s) # Missing required field(s)
bad_payload = {"invalid": "data"} bad_payload = {"invalid": "data"}
response = client.post("/command", json=bad_payload) response = client.post("/command/speech", json=bad_payload)
assert response.status_code == 422 # validation error
def test_receive_gesture_command_invalid_payload(client):
"""
Test invalid data handling (schema validation).
"""
# Missing required field(s)
bad_payload = {"invalid": "data"}
response = client.post("/command/gesture", json=bad_payload)
assert response.status_code == 422 # validation error assert response.status_code == 422 # validation error
@@ -69,6 +126,9 @@ def test_ping_check_returns_none(client):
assert response.json() is None assert response.json() is None
# ----------------------------
# ping_stream tests (unchanged behavior)
# ----------------------------
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_ping_stream_yields_ping_event(monkeypatch): async def test_ping_stream_yields_ping_event(monkeypatch):
"""Test that ping_stream yields a proper SSE message when a ping is received.""" """Test that ping_stream yields a proper SSE message when a ping is received."""
@@ -81,6 +141,11 @@ async def test_ping_stream_yields_ping_event(monkeypatch):
mock_context.socket.return_value = mock_sub_socket mock_context.socket.return_value = mock_sub_socket
monkeypatch.setattr(robot.Context, "instance", lambda: mock_context) monkeypatch.setattr(robot.Context, "instance", lambda: mock_context)
# patch settings address used by ping_stream
mock_settings = MagicMock()
mock_settings.zmq_settings.internal_sub_address = "tcp://localhost:5555"
monkeypatch.setattr(robot, "settings", mock_settings)
mock_request = AsyncMock() mock_request = AsyncMock()
mock_request.is_disconnected = AsyncMock(side_effect=[False, True]) mock_request.is_disconnected = AsyncMock(side_effect=[False, True])
@@ -94,7 +159,7 @@ async def test_ping_stream_yields_ping_event(monkeypatch):
with pytest.raises(StopAsyncIteration): with pytest.raises(StopAsyncIteration):
await anext(generator) await anext(generator)
mock_sub_socket.connect.assert_called_once() mock_sub_socket.connect.assert_called_once_with("tcp://localhost:5555")
mock_sub_socket.setsockopt.assert_called_once_with(robot.zmq.SUBSCRIBE, b"ping") mock_sub_socket.setsockopt.assert_called_once_with(robot.zmq.SUBSCRIBE, b"ping")
mock_sub_socket.recv_multipart.assert_awaited() mock_sub_socket.recv_multipart.assert_awaited()
@@ -111,6 +176,10 @@ async def test_ping_stream_handles_timeout(monkeypatch):
mock_context.socket.return_value = mock_sub_socket mock_context.socket.return_value = mock_sub_socket
monkeypatch.setattr(robot.Context, "instance", lambda: mock_context) monkeypatch.setattr(robot.Context, "instance", lambda: mock_context)
mock_settings = MagicMock()
mock_settings.zmq_settings.internal_sub_address = "tcp://localhost:5555"
monkeypatch.setattr(robot, "settings", mock_settings)
mock_request = AsyncMock() mock_request = AsyncMock()
mock_request.is_disconnected = AsyncMock(return_value=True) mock_request.is_disconnected = AsyncMock(return_value=True)
@@ -120,7 +189,7 @@ async def test_ping_stream_handles_timeout(monkeypatch):
with pytest.raises(StopAsyncIteration): with pytest.raises(StopAsyncIteration):
await anext(generator) await anext(generator)
mock_sub_socket.connect.assert_called_once() mock_sub_socket.connect.assert_called_once_with("tcp://localhost:5555")
mock_sub_socket.setsockopt.assert_called_once_with(robot.zmq.SUBSCRIBE, b"ping") mock_sub_socket.setsockopt.assert_called_once_with(robot.zmq.SUBSCRIBE, b"ping")
mock_sub_socket.recv_multipart.assert_awaited() mock_sub_socket.recv_multipart.assert_awaited()
@@ -139,6 +208,10 @@ async def test_ping_stream_yields_json_values(monkeypatch):
mock_context.socket.return_value = mock_sub_socket mock_context.socket.return_value = mock_sub_socket
monkeypatch.setattr(robot.Context, "instance", lambda: mock_context) monkeypatch.setattr(robot.Context, "instance", lambda: mock_context)
mock_settings = MagicMock()
mock_settings.zmq_settings.internal_sub_address = "tcp://localhost:5555"
monkeypatch.setattr(robot, "settings", mock_settings)
mock_request = AsyncMock() mock_request = AsyncMock()
mock_request.is_disconnected = AsyncMock(side_effect=[False, True]) mock_request.is_disconnected = AsyncMock(side_effect=[False, True])
@@ -151,6 +224,192 @@ async def test_ping_stream_yields_json_values(monkeypatch):
assert "connected" in event_text assert "connected" in event_text
assert "true" in event_text assert "true" in event_text
mock_sub_socket.connect.assert_called_once() mock_sub_socket.connect.assert_called_once_with("tcp://localhost:5555")
mock_sub_socket.setsockopt.assert_called_once_with(robot.zmq.SUBSCRIBE, b"ping") mock_sub_socket.setsockopt.assert_called_once_with(robot.zmq.SUBSCRIBE, b"ping")
mock_sub_socket.recv_multipart.assert_awaited() mock_sub_socket.recv_multipart.assert_awaited()
# ----------------------------
# Updated get_available_gesture_tags tests (REQ socket on tcp://localhost:7788)
# ----------------------------
@pytest.mark.asyncio
async def test_get_available_gesture_tags_success(client, monkeypatch):
"""
Test successful retrieval of available gesture tags using a REQ socket.
"""
# Arrange
mock_req_socket = AsyncMock(spec=zmq.asyncio.Socket)
mock_req_socket.connect = MagicMock()
mock_req_socket.send = AsyncMock()
response_data = {"tags": ["wave", "nod", "point", "dance"]}
mock_req_socket.recv = AsyncMock(return_value=json.dumps(response_data).encode())
mock_context = MagicMock()
mock_context.socket.return_value = mock_req_socket
monkeypatch.setattr(robot.Context, "instance", lambda: mock_context)
# Replace logger methods to avoid noisy logs in tests
monkeypatch.setattr(robot.logger, "debug", MagicMock())
monkeypatch.setattr(robot.logger, "error", MagicMock())
# Act
response = client.get("/commands/gesture/tags")
# Assert
assert response.status_code == 200
assert response.json() == {"available_gesture_tags": ["wave", "nod", "point", "dance"]}
# Verify ZeroMQ REQ interactions
mock_req_socket.connect.assert_called_once_with("tcp://localhost:7788")
mock_req_socket.send.assert_awaited_once_with(b"None")
mock_req_socket.recv.assert_awaited_once()
@pytest.mark.asyncio
async def test_get_available_gesture_tags_with_amount(client, monkeypatch):
"""
The endpoint currently ignores the 'amount' TODO, so behavior is the same as 'success'.
This test asserts that the endpoint still sends b"None" and returns the tags.
"""
# Arrange
mock_req_socket = AsyncMock(spec=zmq.asyncio.Socket)
mock_req_socket.connect = MagicMock()
mock_req_socket.send = AsyncMock()
response_data = {"tags": ["wave", "nod"]}
mock_req_socket.recv = AsyncMock(return_value=json.dumps(response_data).encode())
mock_context = MagicMock()
mock_context.socket.return_value = mock_req_socket
monkeypatch.setattr(robot.Context, "instance", lambda: mock_context)
monkeypatch.setattr(robot.logger, "debug", MagicMock())
monkeypatch.setattr(robot.logger, "error", MagicMock())
# Act
response = client.get("/commands/gesture/tags")
# Assert
assert response.status_code == 200
assert response.json() == {"available_gesture_tags": ["wave", "nod"]}
mock_req_socket.connect.assert_called_once_with("tcp://localhost:7788")
mock_req_socket.send.assert_awaited_once_with(b"None")
@pytest.mark.asyncio
async def test_get_available_gesture_tags_timeout(client, monkeypatch):
"""
Test timeout scenario when fetching gesture tags. Endpoint should handle TimeoutError
and return an empty list while logging the timeout.
"""
# Arrange
mock_req_socket = AsyncMock(spec=zmq.asyncio.Socket)
mock_req_socket.connect = MagicMock()
mock_req_socket.send = AsyncMock()
mock_req_socket.recv = AsyncMock(side_effect=TimeoutError)
mock_context = MagicMock()
mock_context.socket.return_value = mock_req_socket
monkeypatch.setattr(robot.Context, "instance", lambda: mock_context)
# Patch logger.debug so we can assert it was called with the expected message
mock_debug = MagicMock()
monkeypatch.setattr(robot.logger, "debug", mock_debug)
monkeypatch.setattr(robot.logger, "error", MagicMock())
# Act
response = client.get("/commands/gesture/tags")
# Assert
assert response.status_code == 200
assert response.json() == {"available_gesture_tags": []}
# Verify the timeout was logged using the exact string from the endpoint code
mock_debug.assert_called_once_with("Got timeout error fetching gestures.")
mock_req_socket.connect.assert_called_once_with("tcp://localhost:7788")
mock_req_socket.send.assert_awaited_once_with(b"None")
mock_req_socket.recv.assert_awaited_once()
@pytest.mark.asyncio
async def test_get_available_gesture_tags_empty_response(client, monkeypatch):
"""
Test scenario when response contains an empty 'tags' list.
"""
# Arrange
mock_req_socket = AsyncMock(spec=zmq.asyncio.Socket)
mock_req_socket.connect = MagicMock()
mock_req_socket.send = AsyncMock()
response_data = {"tags": []}
mock_req_socket.recv = AsyncMock(return_value=json.dumps(response_data).encode())
mock_context = MagicMock()
mock_context.socket.return_value = mock_req_socket
monkeypatch.setattr(robot.Context, "instance", lambda: mock_context)
monkeypatch.setattr(robot.logger, "debug", MagicMock())
monkeypatch.setattr(robot.logger, "error", MagicMock())
# Act
response = client.get("/commands/gesture/tags")
# Assert
assert response.status_code == 200
assert response.json() == {"available_gesture_tags": []}
@pytest.mark.asyncio
async def test_get_available_gesture_tags_missing_tags_key(client, monkeypatch):
"""
Test scenario when response JSON doesn't contain 'tags' key.
"""
# Arrange
mock_req_socket = AsyncMock(spec=zmq.asyncio.Socket)
mock_req_socket.connect = MagicMock()
mock_req_socket.send = AsyncMock()
response_data = {"some_other_key": "value"}
mock_req_socket.recv = AsyncMock(return_value=json.dumps(response_data).encode())
mock_context = MagicMock()
mock_context.socket.return_value = mock_req_socket
monkeypatch.setattr(robot.Context, "instance", lambda: mock_context)
monkeypatch.setattr(robot.logger, "debug", MagicMock())
monkeypatch.setattr(robot.logger, "error", MagicMock())
# Act
response = client.get("/commands/gesture/tags")
# Assert
assert response.status_code == 200
assert response.json() == {"available_gesture_tags": []}
@pytest.mark.asyncio
async def test_get_available_gesture_tags_invalid_json(client, monkeypatch):
"""
Test scenario when response contains invalid JSON. Endpoint should log the error
and return an empty list.
"""
# Arrange
mock_req_socket = AsyncMock(spec=zmq.asyncio.Socket)
mock_req_socket.connect = MagicMock()
mock_req_socket.send = AsyncMock()
mock_req_socket.recv = AsyncMock(return_value=b"invalid json")
mock_context = MagicMock()
mock_context.socket.return_value = mock_req_socket
monkeypatch.setattr(robot.Context, "instance", lambda: mock_context)
mock_error = MagicMock()
monkeypatch.setattr(robot.logger, "error", mock_error)
monkeypatch.setattr(robot.logger, "debug", MagicMock())
# Act
response = client.get("/commands/gesture/tags")
# Assert - invalid JSON should lead to empty list and error log invocation
assert response.status_code == 200
assert response.json() == {"available_gesture_tags": []}
assert mock_error.call_count == 1

View File

@@ -1,26 +1,88 @@
import pytest import pytest
from pydantic import ValidationError from pydantic import ValidationError
from control_backend.schemas.ri_message import RIEndpoint, RIMessage, SpeechCommand from control_backend.schemas.ri_message import GestureCommand, RIEndpoint, RIMessage, SpeechCommand
def valid_command_1(): def valid_command_1():
return SpeechCommand(data="Hallo?") return SpeechCommand(data="Hallo?")
def valid_command_2():
return GestureCommand(endpoint=RIEndpoint.GESTURE_TAG, data="happy")
def valid_command_3():
return GestureCommand(endpoint=RIEndpoint.GESTURE_SINGLE, data="happy_1")
def invalid_command_1(): def invalid_command_1():
return RIMessage(endpoint=RIEndpoint.PING, data="Hello again.") return RIMessage(endpoint=RIEndpoint.PING, data="Hello again.")
def invalid_command_2():
return RIMessage(endpoint=RIEndpoint.PING, data="Hey!")
def invalid_command_3():
return RIMessage(endpoint=RIEndpoint.GESTURE_SINGLE, data={1, 2, 3})
def invalid_command_4():
test: RIMessage = GestureCommand(endpoint=RIEndpoint.GESTURE_SINGLE, data="asdsad")
def change_endpoint(msg: RIMessage):
msg.endpoint = RIEndpoint.PING
change_endpoint(test)
return test
def test_valid_speech_command_1(): def test_valid_speech_command_1():
command = valid_command_1() command = valid_command_1()
RIMessage.model_validate(command) RIMessage.model_validate(command)
SpeechCommand.model_validate(command) SpeechCommand.model_validate(command)
def test_valid_gesture_command_1():
command = valid_command_2()
RIMessage.model_validate(command)
GestureCommand.model_validate(command)
def test_valid_gesture_command_2():
command = valid_command_3()
RIMessage.model_validate(command)
GestureCommand.model_validate(command)
def test_invalid_speech_command_1(): def test_invalid_speech_command_1():
command = invalid_command_1() command = invalid_command_1()
RIMessage.model_validate(command) RIMessage.model_validate(command)
with pytest.raises(ValidationError): with pytest.raises(ValidationError):
SpeechCommand.model_validate(command) SpeechCommand.model_validate(command)
def test_invalid_gesture_command_1():
command = invalid_command_2()
RIMessage.model_validate(command)
with pytest.raises(ValidationError):
GestureCommand.model_validate(command)
def test_invalid_gesture_command_2():
command = invalid_command_3()
RIMessage.model_validate(command)
with pytest.raises(ValidationError):
GestureCommand.model_validate(command)
def test_invalid_gesture_command_3():
command = invalid_command_4()
RIMessage.model_validate(command)
with pytest.raises(ValidationError):
GestureCommand.model_validate(command)

View File

@@ -53,8 +53,6 @@ async def test_lifespan_agent_start_exception():
Ensures exceptions are logged properly and re-raised. Ensures exceptions are logged properly and re-raised.
""" """
with ( with (
patch("control_backend.main.VADAgent.start", new_callable=AsyncMock),
patch("control_backend.main.VADAgent.reset_stream", new_callable=AsyncMock),
patch( patch(
"control_backend.main.RICommunicationAgent.start", new_callable=AsyncMock "control_backend.main.RICommunicationAgent.start", new_callable=AsyncMock
) as ri_start, ) as ri_start,