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

Merge branch 'feat/environment-variables' into 'dev'

Browse Source 
Docs for environment variables, parameterize some constants

See merge request ics/sp/2025/n25b/pepperplus-cb!38
This commit was merged in pull request #38.
This commit is contained in:
Björn Otgaar
2026-01-06 09:02:49 +00:00
parent 4c20656c75 7d798f2e77
commit 612a96940d
9 changed files with 104 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
test/unit/agents/actuation/test_robot_gesture_agent.py
View File

@@ -73,7 +73,7 @@ async def test_setup_connect(zmq_context, mocker):
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 = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"], address="")
agent.pubsocket = pubsocket
payload = {
@@ -91,7 +91,7 @@ async def test_handle_message_sends_valid_gesture_command():
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 = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"], address="")
agent.pubsocket = pubsocket
payload = {"endpoint": "some_other_endpoint", "data": "invalid_tag_not_in_list"}
@@ -107,7 +107,7 @@ async def test_handle_message_sends_non_gesture_command():
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 = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"], address="")
agent.pubsocket = pubsocket
# Use a tag that's not in gesture_data
@@ -123,7 +123,7 @@ async def test_handle_message_rejects_invalid_gesture_tag():
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 = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"], address="")
agent.pubsocket = pubsocket
msg = InternalMessage(to="robot", sender="tester", body=json.dumps({"bad": "data"}))
@@ -142,12 +142,12 @@ async def test_zmq_command_loop_valid_gesture_payload():
async def recv_once():
# stop after first iteration
agent._running = False
return (b"command", json.dumps(command).encode("utf-8"))
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 = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"], address="")
agent.subsocket = fake_socket
agent.pubsocket = fake_socket
agent._running = True
@@ -165,12 +165,12 @@ async def test_zmq_command_loop_valid_non_gesture_payload():
async def recv_once():
agent._running = False
return (b"command", json.dumps(command).encode("utf-8"))
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 = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"], address="")
agent.subsocket = fake_socket
agent.pubsocket = fake_socket
agent._running = True
@@ -188,12 +188,12 @@ async def test_zmq_command_loop_invalid_gesture_tag():
async def recv_once():
agent._running = False
return (b"command", json.dumps(command).encode("utf-8"))
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 = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"], address="")
agent.subsocket = fake_socket
agent.pubsocket = fake_socket
agent._running = True
@@ -210,12 +210,12 @@ async def test_zmq_command_loop_invalid_json():
async def recv_once():
agent._running = False
return (b"command", b"{not_json}")
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 = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"], address="")
agent.subsocket = fake_socket
agent.pubsocket = fake_socket
agent._running = True
@@ -232,12 +232,12 @@ async def test_zmq_command_loop_ignores_send_gestures_topic():
async def recv_once():
agent._running = False
return (b"send_gestures", b"{}")
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 = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"], address="")
agent.subsocket = fake_socket
agent.pubsocket = fake_socket
agent._running = True
@@ -259,7 +259,9 @@ async def test_fetch_gestures_loop_without_amount():
fake_repsocket.recv = recv_once
fake_repsocket.send = AsyncMock()
agent = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no", "wave", "point"])
agent = RobotGestureAgent(
"robot_gesture", gesture_data=["hello", "yes", "no", "wave", "point"], address=""
)
agent.repsocket = fake_repsocket
agent._running = True
@@ -287,7 +289,9 @@ async def test_fetch_gestures_loop_with_amount():
fake_repsocket.recv = recv_once
fake_repsocket.send = AsyncMock()
agent = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no", "wave", "point"])
agent = RobotGestureAgent(
"robot_gesture", gesture_data=["hello", "yes", "no", "wave", "point"], address=""
)
agent.repsocket = fake_repsocket
agent._running = True
@@ -315,7 +319,7 @@ async def test_fetch_gestures_loop_with_integer_request():
fake_repsocket.recv = recv_once
fake_repsocket.send = AsyncMock()
agent = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"])
agent = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"], address="")
agent.repsocket = fake_repsocket
agent._running = True
@@ -340,7 +344,7 @@ async def test_fetch_gestures_loop_with_invalid_json():
fake_repsocket.recv = recv_once
fake_repsocket.send = AsyncMock()
agent = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"])
agent = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"], address="")
agent.repsocket = fake_repsocket
agent._running = True
@@ -365,7 +369,7 @@ async def test_fetch_gestures_loop_with_non_integer_json():
fake_repsocket.recv = recv_once
fake_repsocket.send = AsyncMock()
agent = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"])
agent = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"], address="")
agent.repsocket = fake_repsocket
agent._running = True
@@ -381,7 +385,7 @@ async def test_fetch_gestures_loop_with_non_integer_json():
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)
agent = RobotGestureAgent("robot_gesture", gesture_data=gesture_data, address="")
assert agent.gesture_data == gesture_data
assert isinstance(agent.gesture_data, list)
@@ -398,7 +402,7 @@ async def test_stop_closes_sockets():
pubsocket = MagicMock()
subsocket = MagicMock()
repsocket = MagicMock()
agent = RobotGestureAgent("robot_gesture")
agent = RobotGestureAgent("robot_gesture", address="")
agent.pubsocket = pubsocket
agent.subsocket = subsocket
agent.repsocket = repsocket
@@ -415,7 +419,7 @@ async def test_stop_closes_sockets():
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)
agent = RobotGestureAgent("robot_gesture", gesture_data=custom_gestures, address="")
assert agent.gesture_data == custom_gestures
@@ -432,7 +436,7 @@ async def test_fetch_gestures_loop_handles_exception():
fake_repsocket.recv = recv_once
fake_repsocket.send = AsyncMock()
agent = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"])
agent = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"], address="")
agent.repsocket = fake_repsocket
agent.logger = MagicMock()
agent._running = True

15
test/unit/agents/perception/vad_agent/test_vad_streaming.py
View File

@@ -7,6 +7,15 @@ import zmq
from control_backend.agents.perception.vad_agent import VADAgent
# We don't want to use real ZMQ in unit tests, for example because it can give errors when sockets
# aren't closed properly.
@pytest.fixture(autouse=True)
def mock_zmq():
with patch("zmq.asyncio.Context") as mock:
mock.instance.return_value = MagicMock()
yield mock
@pytest.fixture
def audio_out_socket():
return AsyncMock()
@@ -140,12 +149,10 @@ async def test_vad_model_load_failure_stops_agent(vad_agent):
# Patch stop to an AsyncMock so we can check it was awaited
vad_agent.stop = AsyncMock()
result = await vad_agent.setup()
await vad_agent.setup()
# Assert stop was called
vad_agent.stop.assert_awaited_once()
# Assert setup returned None
assert result is None
@pytest.mark.asyncio
@@ -155,7 +162,7 @@ async def test_audio_out_bind_failure_sets_none_and_logs(vad_agent, caplog):
audio_out_socket is set to None, None is returned, and an error is logged.
"""
mock_socket = MagicMock()
mock_socket.bind_to_random_port.side_effect = zmq.ZMQBindError()
mock_socket.bind.side_effect = zmq.ZMQBindError()
with patch("control_backend.agents.perception.vad_agent.azmq.Context.instance") as mock_ctx:
mock_ctx.return_value.socket.return_value = mock_socket