diff --git a/.gitlab/merge_request_templates/default.md b/.gitlab/merge_request_templates/default.md new file mode 100644 index 0000000..7a76ac5 --- /dev/null +++ b/.gitlab/merge_request_templates/default.md @@ -0,0 +1,9 @@ +%{first_multiline_commit_description} + +To verify: + +- [ ] Style checks pass +- [ ] Pipeline (tests) pass +- [ ] Documentation is up to date +- [ ] Tests are up to date (new code is covered) +- [ ] ... diff --git a/.logging_config.yaml b/.logging_config.yaml index a244917..4af5d56 100644 --- a/.logging_config.yaml +++ b/.logging_config.yaml @@ -3,6 +3,7 @@ version: 1 custom_levels: OBSERVATION: 25 ACTION: 26 + LLM: 9 formatters: # Console output @@ -26,7 +27,7 @@ handlers: stream: ext://sys.stdout ui: class: zmq.log.handlers.PUBHandler - level: DEBUG + level: LLM formatter: json_experiment # Level of external libraries @@ -36,5 +37,5 @@ root: loggers: control_backend: - level: DEBUG + level: LLM handlers: [ui] diff --git a/src/control_backend/agents/actuation/__init__.py b/src/control_backend/agents/actuation/__init__.py index e745333..8ff7e7f 100644 --- a/src/control_backend/agents/actuation/__init__.py +++ b/src/control_backend/agents/actuation/__init__.py @@ -1 +1,2 @@ +from .robot_gesture_agent import RobotGestureAgent as RobotGestureAgent from .robot_speech_agent import RobotSpeechAgent as RobotSpeechAgent diff --git a/src/control_backend/agents/actuation/robot_gesture_agent.py b/src/control_backend/agents/actuation/robot_gesture_agent.py new file mode 100644 index 0000000..4f5dd79 --- /dev/null +++ b/src/control_backend/agents/actuation/robot_gesture_agent.py @@ -0,0 +1,171 @@ +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 = [] + single_gesture_data = [] + + def __init__( + self, + name: str, + address=settings.zmq_settings.ri_command_address, + bind=False, + gesture_data=None, + single_gesture_data=None, + ): + self.gesture_data = gesture_data or [] + self.single_gesture_data = single_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 + elif gesture_command.endpoint == RIEndpoint.GESTURE_SINGLE: + if gesture_command.data not in self.single_gesture_data: + self.logger.warning( + "Received gesture '%s' which is not in available gestures. 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.") diff --git a/src/control_backend/agents/communication/ri_communication_agent.py b/src/control_backend/agents/communication/ri_communication_agent.py index 4a043e9..5c6ca77 100644 --- a/src/control_backend/agents/communication/ri_communication_agent.py +++ b/src/control_backend/agents/communication/ri_communication_agent.py @@ -6,6 +6,7 @@ import zmq.asyncio as azmq from zmq.asyncio import Context from control_backend.agents import BaseAgent +from control_backend.agents.actuation.robot_gesture_agent import RobotGestureAgent from control_backend.core.config import settings from ..actuation.robot_speech_agent import RobotSpeechAgent @@ -180,12 +181,23 @@ class RICommunicationAgent(BaseAgent): else: self._req_socket.bind(addr) case "actuation": - ri_commands_agent = RobotSpeechAgent( + gesture_data = port_data.get("gestures", []) + single_gesture_data = port_data.get("single_gestures", []) + robot_speech_agent = RobotSpeechAgent( settings.agent_settings.robot_speech_name, address=addr, 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, + single_gesture_data=single_gesture_data, + ) + await robot_speech_agent.start() + await asyncio.sleep(0.1) # Small delay + await robot_gesture_agent.start() case "audio": vad_agent = VADAgent(audio_in_address=addr, audio_in_bind=bind) await vad_agent.start() diff --git a/src/control_backend/agents/llm/llm_agent.py b/src/control_backend/agents/llm/llm_agent.py index 0263b30..55099e2 100644 --- a/src/control_backend/agents/llm/llm_agent.py +++ b/src/control_backend/agents/llm/llm_agent.py @@ -125,7 +125,7 @@ class LLMAgent(BaseAgent): full_message += token current_chunk += token - self.logger.info( + self.logger.llm( "Received token: %s", full_message, extra={"reference": message_id}, # Used in the UI to update old logs diff --git a/src/control_backend/agents/user_interrupt/__init__.py b/src/control_backend/agents/user_interrupt/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/control_backend/agents/user_interrupt/user_interrupt_agent.py b/src/control_backend/agents/user_interrupt/user_interrupt_agent.py new file mode 100644 index 0000000..b2efc41 --- /dev/null +++ b/src/control_backend/agents/user_interrupt/user_interrupt_agent.py @@ -0,0 +1,146 @@ +import json + +import zmq +from zmq.asyncio import Context + +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, SpeechCommand + + +class UserInterruptAgent(BaseAgent): + """ + User Interrupt Agent. + + This agent receives button_pressed events from the external HTTP API + (via ZMQ) and uses the associated context to trigger one of the following actions: + + - Send a prioritized message to the `RobotSpeechAgent` + - Send a prioritized gesture to the `RobotGestureAgent` + - Send a belief override to the `BDIProgramManager`in order to activate a + trigger/conditional norm or complete a goal. + + Prioritized actions clear the current RI queue before inserting the new item, + ensuring they are executed immediately after Pepper's current action has been fulfilled. + + :ivar sub_socket: The ZMQ SUB socket used to receive user intterupts. + """ + + def __init__(self, **kwargs): + super().__init__(**kwargs) + self.sub_socket = None + + async def _receive_button_event(self): + """ + The behaviour of the UserInterruptAgent. + Continuous loop that receives button_pressed events from the button_pressed HTTP endpoint. + These events contain a type and a context. + + These are the different types and contexts: + - type: "speech", context: string that the robot has to say. + - type: "gesture", context: single gesture name that the robot has to perform. + - type: "override", context: belief_id that overrides the goal/trigger/conditional norm. + """ + while True: + topic, body = await self.sub_socket.recv_multipart() + + try: + event_data = json.loads(body) + event_type = event_data.get("type") # e.g., "speech", "gesture" + event_context = event_data.get("context") # e.g., "Hello, I am Pepper!" + except json.JSONDecodeError: + self.logger.error("Received invalid JSON payload on topic %s", topic) + continue + + if event_type == "speech": + await self._send_to_speech_agent(event_context) + self.logger.info( + "Forwarded button press (speech) with context '%s' to RobotSpeechAgent.", + event_context, + ) + elif event_type == "gesture": + await self._send_to_gesture_agent(event_context) + self.logger.info( + "Forwarded button press (gesture) with context '%s' to RobotGestureAgent.", + event_context, + ) + elif event_type == "override": + await self._send_to_program_manager(event_context) + self.logger.info( + "Forwarded button press (override) with context '%s' to BDIProgramManager.", + event_context, + ) + else: + self.logger.warning( + "Received button press with unknown type '%s' (context: '%s').", + event_type, + event_context, + ) + + async def _send_to_speech_agent(self, text_to_say: str): + """ + method to send prioritized speech command to RobotSpeechAgent. + + :param text_to_say: The string that the robot has to say. + """ + cmd = SpeechCommand(data=text_to_say, is_priority=True) + out_msg = InternalMessage( + to=settings.agent_settings.robot_speech_name, + sender=self.name, + body=cmd.model_dump_json(), + ) + await self.send(out_msg) + + async def _send_to_gesture_agent(self, single_gesture_name: str): + """ + method to send prioritized gesture command to RobotGestureAgent. + + :param single_gesture_name: The gesture tag that the robot has to perform. + """ + # the endpoint is set to always be GESTURE_SINGLE for user interrupts + cmd = GestureCommand( + endpoint=RIEndpoint.GESTURE_SINGLE, data=single_gesture_name, is_priority=True + ) + out_msg = InternalMessage( + to=settings.agent_settings.robot_gesture_name, + sender=self.name, + body=cmd.model_dump_json(), + ) + await self.send(out_msg) + + async def _send_to_program_manager(self, belief_id: str): + """ + Send a button_override belief to the BDIProgramManager. + + :param belief_id: The belief_id that overrides the goal/trigger/conditional norm. + this id can belong to a basic belief or an inferred belief. + See also: https://utrechtuniversity.youtrack.cloud/articles/N25B-A-27/UI-components + """ + data = {"belief": belief_id} + message = InternalMessage( + to=settings.agent_settings.bdi_program_manager_name, + sender=self.name, + body=json.dumps(data), + thread="belief_override_id", + ) + await self.send(message) + self.logger.info( + "Sent button_override belief with id '%s' to Program manager.", + belief_id, + ) + + async def setup(self): + """ + Initialize the agent. + + Connects the internal ZMQ SUB socket and subscribes to the 'button_pressed' topic. + Starts the background behavior to receive the user interrupts. + """ + context = Context.instance() + + self.sub_socket = context.socket(zmq.SUB) + self.sub_socket.connect(settings.zmq_settings.internal_sub_address) + self.sub_socket.subscribe("button_pressed") + + self.add_behavior(self._receive_button_event()) diff --git a/src/control_backend/api/v1/endpoints/button_pressed.py b/src/control_backend/api/v1/endpoints/button_pressed.py new file mode 100644 index 0000000..5a94a53 --- /dev/null +++ b/src/control_backend/api/v1/endpoints/button_pressed.py @@ -0,0 +1,31 @@ +import logging + +from fastapi import APIRouter, Request + +from control_backend.schemas.events import ButtonPressedEvent + +logger = logging.getLogger(__name__) +router = APIRouter() + + +@router.post("/button_pressed", status_code=202) +async def receive_button_event(event: ButtonPressedEvent, request: Request): + """ + Endpoint to handle external button press events. + + Validates the event payload and publishes it to the internal 'button_pressed' topic. + Subscribers (in this case user_interrupt_agent) will pick this up to trigger + specific behaviors or state changes. + + :param event: The parsed ButtonPressedEvent object. + :param request: The FastAPI request object. + """ + logger.debug("Received button event: %s | %s", event.type, event.context) + + topic = b"button_pressed" + body = event.model_dump_json().encode() + + pub_socket = request.app.state.endpoints_pub_socket + await pub_socket.send_multipart([topic, body]) + + return {"status": "Event received"} diff --git a/src/control_backend/api/v1/endpoints/robot.py b/src/control_backend/api/v1/endpoints/robot.py index ae0fe66..afbf1ac 100644 --- a/src/control_backend/api/v1/endpoints/robot.py +++ b/src/control_backend/api/v1/endpoints/robot.py @@ -8,15 +8,15 @@ from fastapi.responses import StreamingResponse from zmq.asyncio import Context, Socket 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__) router = APIRouter() -@router.post("/command", status_code=202) -async def receive_command(command: SpeechCommand, request: Request): +@router.post("/command/speech", status_code=202) +async def receive_command_speech(command: SpeechCommand, request: Request): """ 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 request: The FastAPI request object. """ - # Validate and retrieve data. - SpeechCommand.model_validate(command) 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": "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") @@ -45,6 +63,41 @@ async def ping(request: Request): 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") async def ping_stream(request: Request): """ diff --git a/src/control_backend/api/v1/router.py b/src/control_backend/api/v1/router.py index ce5a70b..ebba0db 100644 --- a/src/control_backend/api/v1/router.py +++ b/src/control_backend/api/v1/router.py @@ -1,6 +1,6 @@ from fastapi.routing import APIRouter -from control_backend.api.v1.endpoints import logs, message, program, robot, sse +from control_backend.api.v1.endpoints import button_pressed, logs, message, program, robot, sse api_router = APIRouter() @@ -13,3 +13,5 @@ api_router.include_router(robot.router, prefix="/robot", tags=["Pings", "Command api_router.include_router(logs.router, tags=["Logs"]) api_router.include_router(program.router, tags=["Program"]) + +api_router.include_router(button_pressed.router, tags=["Button Pressed Events"]) diff --git a/src/control_backend/core/config.py b/src/control_backend/core/config.py index 35acf96..c4a4db7 100644 --- a/src/control_backend/core/config.py +++ b/src/control_backend/core/config.py @@ -26,6 +26,7 @@ class ZMQSettings(BaseModel): internal_pub_address: str = "tcp://localhost:5560" internal_sub_address: str = "tcp://localhost:5561" ri_communication_address: str = "tcp://*:5555" + internal_gesture_rep_adress: str = "tcp://localhost:7788" vad_pub_address: str = "inproc://vad_stream" @@ -58,6 +59,8 @@ class AgentSettings(BaseModel): transcription_name: str = "transcription_agent" ri_communication_name: str = "ri_communication_agent" robot_speech_name: str = "robot_speech_agent" + robot_gesture_name: str = "robot_gesture_agent" + user_interrupt_name: str = "user_interrupt_agent" class BehaviourSettings(BaseModel): diff --git a/src/control_backend/logging/setup_logging.py b/src/control_backend/logging/setup_logging.py index fe40738..05ae85a 100644 --- a/src/control_backend/logging/setup_logging.py +++ b/src/control_backend/logging/setup_logging.py @@ -4,6 +4,7 @@ import os import yaml import zmq +from zmq.log.handlers import PUBHandler from control_backend.core.config import settings @@ -51,15 +52,27 @@ def setup_logging(path: str = ".logging_config.yaml") -> None: logging.warning(f"Could not load logging configuration: {e}") config = {} - if "custom_levels" in config: - for level_name, level_num in config["custom_levels"].items(): - add_logging_level(level_name, level_num) + custom_levels = config.get("custom_levels", {}) or {} + for level_name, level_num in custom_levels.items(): + add_logging_level(level_name, level_num) if config.get("handlers") is not None and config.get("handlers").get("ui"): pub_socket = zmq.Context.instance().socket(zmq.PUB) pub_socket.connect(settings.zmq_settings.internal_pub_address) config["handlers"]["ui"]["interface_or_socket"] = pub_socket + logging.config.dictConfig(config) + # Patch ZMQ PUBHandler to know about custom levels + if custom_levels: + for logger_name in ("control_backend",): + logger = logging.getLogger(logger_name) + for handler in logger.handlers: + if isinstance(handler, PUBHandler): + # Use the INFO formatter as the default template + default_fmt = handler.formatters[logging.INFO] + for level_num in custom_levels.values(): + handler.setFormatter(default_fmt, level=level_num) + else: logging.warning("Logging config file not found. Using default logging configuration.") diff --git a/src/control_backend/main.py b/src/control_backend/main.py index 2c8b766..3509cbc 100644 --- a/src/control_backend/main.py +++ b/src/control_backend/main.py @@ -39,6 +39,9 @@ from control_backend.agents.communication import RICommunicationAgent # LLM Agents from control_backend.agents.llm import LLMAgent +# User Interrupt Agent +from control_backend.agents.user_interrupt.user_interrupt_agent import UserInterruptAgent + # Other backend imports from control_backend.api.v1.router import api_router from control_backend.core.config import settings @@ -138,6 +141,12 @@ async def lifespan(app: FastAPI): "name": settings.agent_settings.bdi_program_manager_name, }, ), + "UserInterruptAgent": ( + UserInterruptAgent, + { + "name": settings.agent_settings.user_interrupt_name, + }, + ), } agents = [] diff --git a/src/control_backend/schemas/events.py b/src/control_backend/schemas/events.py new file mode 100644 index 0000000..46967f7 --- /dev/null +++ b/src/control_backend/schemas/events.py @@ -0,0 +1,6 @@ +from pydantic import BaseModel + + +class ButtonPressedEvent(BaseModel): + type: str + context: str diff --git a/src/control_backend/schemas/ri_message.py b/src/control_backend/schemas/ri_message.py index 3f0e5d2..a48dec6 100644 --- a/src/control_backend/schemas/ri_message.py +++ b/src/control_backend/schemas/ri_message.py @@ -1,7 +1,7 @@ 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): @@ -10,6 +10,8 @@ class RIEndpoint(str, Enum): """ SPEECH = "actuate/speech" + GESTURE_SINGLE = "actuate/gesture/single" + GESTURE_TAG = "actuate/gesture/tag" PING = "ping" NEGOTIATE_PORTS = "negotiate/ports" @@ -36,3 +38,29 @@ class SpeechCommand(RIMessage): endpoint: RIEndpoint = RIEndpoint(RIEndpoint.SPEECH) data: str + is_priority: bool = False + + +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 + is_priority: bool = False + + @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 diff --git a/test/unit/agents/actuation/test_robot_gesture_agent.py b/test/unit/agents/actuation/test_robot_gesture_agent.py new file mode 100644 index 0000000..c68f052 --- /dev/null +++ b/test/unit/agents/actuation/test_robot_gesture_agent.py @@ -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() diff --git a/test/unit/agents/actuation/test_robot_speech_agent.py b/test/unit/agents/actuation/test_robot_speech_agent.py index 567a5a1..d95f66a 100644 --- a/test/unit/agents/actuation/test_robot_speech_agent.py +++ b/test/unit/agents/actuation/test_robot_speech_agent.py @@ -8,6 +8,11 @@ from control_backend.agents.actuation.robot_speech_agent import RobotSpeechAgent 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 def zmq_context(mocker): mock_context = mocker.patch( @@ -56,10 +61,10 @@ async def test_setup_connect(zmq_context, mocker): async def test_handle_message_sends_command(): """Internal message is forwarded to robot pub socket as JSON.""" pubsocket = AsyncMock() - agent = RobotSpeechAgent("robot_speech", "tcp://localhost:3498") + agent = mock_speech_agent() agent.pubsocket = pubsocket - payload = {"endpoint": "actuate/speech", "data": "hello"} + payload = {"endpoint": "actuate/speech", "data": "hello", "is_priority": False} msg = InternalMessage(to="robot", sender="tester", body=json.dumps(payload)) await agent.handle_message(msg) @@ -70,7 +75,7 @@ async def test_handle_message_sends_command(): @pytest.mark.asyncio async def test_zmq_command_loop_valid_payload(zmq_context): """UI command is read from SUB and published.""" - command = {"endpoint": "actuate/speech", "data": "hello"} + command = {"endpoint": "actuate/speech", "data": "hello", "is_priority": False} fake_socket = AsyncMock() async def recv_once(): @@ -80,7 +85,7 @@ async def test_zmq_command_loop_valid_payload(zmq_context): fake_socket.recv_multipart = recv_once fake_socket.send_json = AsyncMock() - agent = RobotSpeechAgent("robot_speech", "tcp://localhost:3498") + agent = mock_speech_agent() agent.subsocket = fake_socket agent.pubsocket = fake_socket agent._running = True @@ -101,7 +106,7 @@ async def test_zmq_command_loop_invalid_json(): fake_socket.recv_multipart = recv_once fake_socket.send_json = AsyncMock() - agent = RobotSpeechAgent("robot_speech", "tcp://localhost:3498") + agent = mock_speech_agent() agent.subsocket = fake_socket agent.pubsocket = fake_socket agent._running = True @@ -115,7 +120,7 @@ async def test_zmq_command_loop_invalid_json(): async def test_handle_message_invalid_payload(): """Invalid payload is caught and does not send.""" pubsocket = AsyncMock() - agent = RobotSpeechAgent("robot_speech", "tcp://localhost:3498") + agent = mock_speech_agent() agent.pubsocket = pubsocket 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(): pubsocket = MagicMock() subsocket = MagicMock() - agent = RobotSpeechAgent("robot_speech", "tcp://localhost:3498") + agent = mock_speech_agent() agent.pubsocket = pubsocket agent.subsocket = subsocket diff --git a/test/unit/agents/communication/test_ri_communication_agent.py b/test/unit/agents/communication/test_ri_communication_agent.py index 99210bc..06d8766 100644 --- a/test/unit/agents/communication/test_ri_communication_agent.py +++ b/test/unit/agents/communication/test_ri_communication_agent.py @@ -10,6 +10,10 @@ def speech_agent_path(): return "control_backend.agents.communication.ri_communication_agent.RobotSpeechAgent" +def gesture_agent_path(): + return "control_backend.agents.communication.ri_communication_agent.RobotGestureAgent" + + @pytest.fixture def zmq_context(mocker): mock_context = mocker.patch( @@ -22,7 +26,7 @@ def zmq_context(mocker): def negotiation_message( actuation_port: int = 5556, bind_main: bool = False, - bind_actuation: bool = True, + bind_actuation: bool = False, main_port: int = 5555, ): 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.send_multipart = AsyncMock() - with patch(speech_agent_path(), autospec=True) as MockRobot: - robot_instance = MockRobot.return_value - robot_instance.start = AsyncMock() + with ( + 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() agent = RICommunicationAgent("ri_comm", address="tcp://localhost:5555", bind=False) agent.add_behavior = MagicMock() @@ -52,9 +59,18 @@ async def test_setup_success_connects_and_starts_robot(zmq_context): fake_socket.connect.assert_any_call("tcp://localhost:5555") fake_socket.send_json.assert_any_call({"endpoint": "negotiate/ports", "data": {}}) - robot_instance.start.assert_awaited_once() - MockRobot.assert_called_once_with(ANY, address="tcp://*:5556", bind=True) + MockSpeech.return_value.start.assert_awaited_once() + 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=[], + single_gesture_data=[], + ) agent.add_behavior.assert_called_once() + assert agent.connected is True @@ -69,10 +85,13 @@ async def test_setup_binds_when_requested(zmq_context): agent.add_behavior = MagicMock() - with patch(speech_agent_path(), autospec=True) as MockRobot: - MockRobot.return_value.start = AsyncMock() + with ( + 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() - fake_socket.bind.assert_any_call("tcp://localhost:5555") agent.add_behavior.assert_called_once() @@ -88,7 +107,6 @@ async def test_negotiate_invalid_endpoint_retries(zmq_context): agent._req_socket = fake_socket success = await agent._negotiate_connection(max_retries=1) - assert success is False @@ -112,8 +130,12 @@ async def test_handle_negotiation_response_updates_req_socket(zmq_context): fake_socket = zmq_context.return_value.socket.return_value agent = RICommunicationAgent("ri_comm", address="tcp://localhost:5555", bind=False) agent._req_socket = fake_socket - with patch(speech_agent_path(), autospec=True) as MockRobot: - MockRobot.return_value.start = AsyncMock() + with ( + 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( negotiation_message( main_port=6000, @@ -135,7 +157,6 @@ async def test_handle_disconnection_publishes_and_reconnects(): agent._negotiate_connection = AsyncMock(return_value=True) await agent._handle_disconnection() - pub_socket.send_multipart.assert_awaited() assert agent.connected is True @@ -192,7 +213,7 @@ async def test_setup_warns_on_failed_negotiate(zmq_context, mocker): fake_socket.recv_json = AsyncMock() agent = RICommunicationAgent("ri_comm") - async def swallow(coro): + def swallow(coro): coro.close() agent.add_behavior = swallow diff --git a/test/unit/agents/llm/test_llm_agent.py b/test/unit/agents/llm/test_llm_agent.py index e2b6460..5e84d8d 100644 --- a/test/unit/agents/llm/test_llm_agent.py +++ b/test/unit/agents/llm/test_llm_agent.py @@ -49,6 +49,9 @@ async def test_llm_processing_success(mock_httpx_client, mock_settings): agent = LLMAgent("llm_agent") agent.send = AsyncMock() # Mock the send method to verify replies + mock_logger = MagicMock() + agent.logger = mock_logger + # Simulate receiving a message from BDI prompt = LLMPromptMessage(text="Hi", norms=[], goals=[]) msg = InternalMessage( @@ -194,6 +197,9 @@ async def test_query_llm_yields_final_tail_chunk(mock_settings): agent = LLMAgent("llm_agent") agent.send = AsyncMock() + agent.logger = MagicMock() + agent.logger.llm = MagicMock() + # Patch _stream_query_llm to yield tokens that do NOT end with punctuation async def fake_stream(messages): yield "Hello" diff --git a/test/unit/agents/user_interrupt/test_user_interrupt.py b/test/unit/agents/user_interrupt/test_user_interrupt.py new file mode 100644 index 0000000..7e3e700 --- /dev/null +++ b/test/unit/agents/user_interrupt/test_user_interrupt.py @@ -0,0 +1,146 @@ +import asyncio +import json +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from control_backend.agents.user_interrupt.user_interrupt_agent import UserInterruptAgent +from control_backend.core.agent_system import InternalMessage +from control_backend.core.config import settings +from control_backend.schemas.ri_message import RIEndpoint + + +@pytest.fixture +def agent(): + agent = UserInterruptAgent(name="user_interrupt_agent") + agent.send = AsyncMock() + agent.logger = MagicMock() + agent.sub_socket = AsyncMock() + return agent + + +@pytest.mark.asyncio +async def test_send_to_speech_agent(agent): + """Verify speech command format.""" + await agent._send_to_speech_agent("Hello World") + + agent.send.assert_awaited_once() + sent_msg: InternalMessage = agent.send.call_args.args[0] + + assert sent_msg.to == settings.agent_settings.robot_speech_name + body = json.loads(sent_msg.body) + assert body["data"] == "Hello World" + assert body["is_priority"] is True + + +@pytest.mark.asyncio +async def test_send_to_gesture_agent(agent): + """Verify gesture command format.""" + await agent._send_to_gesture_agent("wave_hand") + + agent.send.assert_awaited_once() + sent_msg: InternalMessage = agent.send.call_args.args[0] + + assert sent_msg.to == settings.agent_settings.robot_gesture_name + body = json.loads(sent_msg.body) + assert body["data"] == "wave_hand" + assert body["is_priority"] is True + assert body["endpoint"] == RIEndpoint.GESTURE_SINGLE.value + + +@pytest.mark.asyncio +async def test_send_to_program_manager(agent): + """Verify belief update format.""" + context_str = "2" + + await agent._send_to_program_manager(context_str) + + agent.send.assert_awaited_once() + sent_msg: InternalMessage = agent.send.call_args.args[0] + + assert sent_msg.to == settings.agent_settings.bdi_program_manager_name + assert sent_msg.thread == "belief_override_id" + + body = json.loads(sent_msg.body) + + assert body["belief"] == context_str + + +@pytest.mark.asyncio +async def test_receive_loop_routing_success(agent): + """ + Test that the loop correctly: + 1. Receives 'button_pressed' topic from ZMQ + 2. Parses the JSON payload to find 'type' and 'context' + 3. Calls the correct handler method based on 'type' + """ + # Prepare JSON payloads as bytes + payload_speech = json.dumps({"type": "speech", "context": "Hello Speech"}).encode() + payload_gesture = json.dumps({"type": "gesture", "context": "Hello Gesture"}).encode() + payload_override = json.dumps({"type": "override", "context": "Hello Override"}).encode() + + agent.sub_socket.recv_multipart.side_effect = [ + (b"button_pressed", payload_speech), + (b"button_pressed", payload_gesture), + (b"button_pressed", payload_override), + asyncio.CancelledError, # Stop the infinite loop + ] + + agent._send_to_speech_agent = AsyncMock() + agent._send_to_gesture_agent = AsyncMock() + agent._send_to_program_manager = AsyncMock() + + try: + await agent._receive_button_event() + except asyncio.CancelledError: + pass + + await asyncio.sleep(0) + + # Speech + agent._send_to_speech_agent.assert_awaited_once_with("Hello Speech") + + # Gesture + agent._send_to_gesture_agent.assert_awaited_once_with("Hello Gesture") + + # Override + agent._send_to_program_manager.assert_awaited_once_with("Hello Override") + + assert agent._send_to_speech_agent.await_count == 1 + assert agent._send_to_gesture_agent.await_count == 1 + assert agent._send_to_program_manager.await_count == 1 + + +@pytest.mark.asyncio +async def test_receive_loop_unknown_type(agent): + """Test that unknown 'type' values in the JSON log a warning and do not crash.""" + + # Prepare a payload with an unknown type + payload_unknown = json.dumps({"type": "unknown_thing", "context": "some_data"}).encode() + + agent.sub_socket.recv_multipart.side_effect = [ + (b"button_pressed", payload_unknown), + asyncio.CancelledError, + ] + + agent._send_to_speech_agent = AsyncMock() + agent._send_to_gesture_agent = AsyncMock() + agent._send_to_belief_collector = AsyncMock() + + try: + await agent._receive_button_event() + except asyncio.CancelledError: + pass + + await asyncio.sleep(0) + + # Ensure no handlers were called + agent._send_to_speech_agent.assert_not_called() + agent._send_to_gesture_agent.assert_not_called() + agent._send_to_belief_collector.assert_not_called() + + agent.logger.warning.assert_called_with( + "Received button press with unknown type '%s' (context: '%s').", + "unknown_thing", + "some_data", + ) diff --git a/test/unit/api/v1/endpoints/test_robot_endpoint.py b/test/unit/api/v1/endpoints/test_robot_endpoint.py index 0f71951..e9e637d 100644 --- a/test/unit/api/v1/endpoints/test_robot_endpoint.py +++ b/test/unit/api/v1/endpoints/test_robot_endpoint.py @@ -1,12 +1,14 @@ +# tests/test_robot_endpoints.py import json -from unittest.mock import AsyncMock, MagicMock +from unittest.mock import AsyncMock, MagicMock, patch import pytest +import zmq.asyncio from fastapi import FastAPI from fastapi.testclient import TestClient 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 @@ -26,7 +28,27 @@ def client(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 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) # Act - response = client.post("/command", json=command_data) + response = client.post("/command/speech", json=command_data) # Assert 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 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). """ # Missing required field(s) 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 @@ -69,6 +126,9 @@ def test_ping_check_returns_none(client): assert response.json() is None +# ---------------------------- +# ping_stream tests (unchanged behavior) +# ---------------------------- @pytest.mark.asyncio async def test_ping_stream_yields_ping_event(monkeypatch): """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 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.is_disconnected = AsyncMock(side_effect=[False, True]) @@ -94,7 +159,7 @@ async def test_ping_stream_yields_ping_event(monkeypatch): with pytest.raises(StopAsyncIteration): 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.recv_multipart.assert_awaited() @@ -111,6 +176,10 @@ async def test_ping_stream_handles_timeout(monkeypatch): mock_context.socket.return_value = mock_sub_socket 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.is_disconnected = AsyncMock(return_value=True) @@ -120,7 +189,7 @@ async def test_ping_stream_handles_timeout(monkeypatch): with pytest.raises(StopAsyncIteration): 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.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 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.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 "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.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 diff --git a/test/unit/schemas/test_ri_message.py b/test/unit/schemas/test_ri_message.py index 5078f9a..40601ec 100644 --- a/test/unit/schemas/test_ri_message.py +++ b/test/unit/schemas/test_ri_message.py @@ -1,26 +1,88 @@ import pytest 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(): 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(): 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(): command = valid_command_1() RIMessage.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(): command = invalid_command_1() RIMessage.model_validate(command) with pytest.raises(ValidationError): 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)