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

Docs for environment variables, parameterize some constants #38

Merged
0950726 merged 5 commits from feat/environment-variables into dev 2026-01-06 09:02:49 +00:00
Conversation 25 Commits 5 Files Changed 9 +28 -24
2 changed files with 28 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 7d798f2e77 - Show all commits

2
src/control_backend/agents/actuation/robot_gesture_agent.py
View File

@@ -33,7 +33,7 @@ class RobotGestureAgent(BaseAgent):
def __init__( def __init__(
self, self,
name: str, name: str,
address=settings.zmq_settings.ri_command_address, address: str,
bind=False, bind=False,
gesture_data=None, gesture_data=None,
single_gesture_data=None, single_gesture_data=None,

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(): async def test_handle_message_sends_valid_gesture_command():
"""Internal message with valid gesture tag is forwarded to robot pub socket.""" """Internal message with valid gesture tag is forwarded to robot pub socket."""
pubsocket = AsyncMock() pubsocket = AsyncMock()
agent = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"]) agent = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"], address="")
agent.pubsocket = pubsocket agent.pubsocket = pubsocket
payload = { payload = {
@@ -91,7 +91,7 @@ async def test_handle_message_sends_valid_gesture_command():
async def test_handle_message_sends_non_gesture_command(): async def test_handle_message_sends_non_gesture_command():
"""Internal message with non-gesture endpoint is not forwarded by this agent.""" """Internal message with non-gesture endpoint is not forwarded by this agent."""
pubsocket = AsyncMock() pubsocket = AsyncMock()
agent = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"]) agent = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"], address="")
agent.pubsocket = pubsocket agent.pubsocket = pubsocket
payload = {"endpoint": "some_other_endpoint", "data": "invalid_tag_not_in_list"} 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(): async def test_handle_message_rejects_invalid_gesture_tag():
"""Internal message with invalid gesture tag is not forwarded.""" """Internal message with invalid gesture tag is not forwarded."""
pubsocket = AsyncMock() pubsocket = AsyncMock()
agent = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"]) agent = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"], address="")
agent.pubsocket = pubsocket agent.pubsocket = pubsocket
# Use a tag that's not in gesture_data # 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(): 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 = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"]) agent = RobotGestureAgent("robot_gesture", gesture_data=["hello", "yes", "no"], address="")
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"}))
@@ -142,12 +142,12 @@ async def test_zmq_command_loop_valid_gesture_payload():
async def recv_once(): async def recv_once():
# stop after first iteration # stop after first iteration
agent._running = False 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.recv_multipart = recv_once
fake_socket.send_json = AsyncMock() 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.subsocket = fake_socket
agent.pubsocket = fake_socket agent.pubsocket = fake_socket
agent._running = True agent._running = True
@@ -165,12 +165,12 @@ async def test_zmq_command_loop_valid_non_gesture_payload():
async def recv_once(): async def recv_once():
agent._running = False 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.recv_multipart = recv_once
fake_socket.send_json = AsyncMock() 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.subsocket = fake_socket
agent.pubsocket = fake_socket agent.pubsocket = fake_socket
agent._running = True agent._running = True
@@ -188,12 +188,12 @@ async def test_zmq_command_loop_invalid_gesture_tag():
async def recv_once(): async def recv_once():
agent._running = False 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.recv_multipart = recv_once
fake_socket.send_json = AsyncMock() 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.subsocket = fake_socket
agent.pubsocket = fake_socket agent.pubsocket = fake_socket
agent._running = True agent._running = True
@@ -210,12 +210,12 @@ async def test_zmq_command_loop_invalid_json():
async def recv_once(): async def recv_once():
agent._running = False agent._running = False
return (b"command", b"{not_json}") return b"command", b"{not_json}"
fake_socket.recv_multipart = recv_once fake_socket.recv_multipart = recv_once
fake_socket.send_json = AsyncMock() 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.subsocket = fake_socket
agent.pubsocket = fake_socket agent.pubsocket = fake_socket
agent._running = True agent._running = True
@@ -232,12 +232,12 @@ async def test_zmq_command_loop_ignores_send_gestures_topic():
async def recv_once(): async def recv_once():
agent._running = False agent._running = False
return (b"send_gestures", b"{}") return b"send_gestures", b"{}"
fake_socket.recv_multipart = recv_once fake_socket.recv_multipart = recv_once
fake_socket.send_json = AsyncMock() 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.subsocket = fake_socket
agent.pubsocket = fake_socket agent.pubsocket = fake_socket
agent._running = True agent._running = True
@@ -259,7 +259,9 @@ async def test_fetch_gestures_loop_without_amount():
fake_repsocket.recv = recv_once fake_repsocket.recv = recv_once
fake_repsocket.send = AsyncMock() 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.repsocket = fake_repsocket
agent._running = True agent._running = True
@@ -287,7 +289,9 @@ async def test_fetch_gestures_loop_with_amount():
fake_repsocket.recv = recv_once fake_repsocket.recv = recv_once
fake_repsocket.send = AsyncMock() 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.repsocket = fake_repsocket
agent._running = True agent._running = True
@@ -315,7 +319,7 @@ async def test_fetch_gestures_loop_with_integer_request():
fake_repsocket.recv = recv_once fake_repsocket.recv = recv_once
fake_repsocket.send = AsyncMock() 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.repsocket = fake_repsocket
agent._running = True agent._running = True
@@ -340,7 +344,7 @@ async def test_fetch_gestures_loop_with_invalid_json():
fake_repsocket.recv = recv_once fake_repsocket.recv = recv_once
fake_repsocket.send = AsyncMock() 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.repsocket = fake_repsocket
agent._running = True agent._running = True
@@ -365,7 +369,7 @@ async def test_fetch_gestures_loop_with_non_integer_json():
fake_repsocket.recv = recv_once fake_repsocket.recv = recv_once
fake_repsocket.send = AsyncMock() 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.repsocket = fake_repsocket
agent._running = True agent._running = True
@@ -381,7 +385,7 @@ async def test_fetch_gestures_loop_with_non_integer_json():
def test_gesture_data_attribute(): def test_gesture_data_attribute():
"""Test that gesture_data returns the expected list.""" """Test that gesture_data returns the expected list."""
gesture_data = ["hello", "yes", "no", "wave"] 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 agent.gesture_data == gesture_data
assert isinstance(agent.gesture_data, list) assert isinstance(agent.gesture_data, list)
@@ -398,7 +402,7 @@ async def test_stop_closes_sockets():
pubsocket = MagicMock() pubsocket = MagicMock()
subsocket = MagicMock() subsocket = MagicMock()
repsocket = MagicMock() repsocket = MagicMock()
agent = RobotGestureAgent("robot_gesture") agent = RobotGestureAgent("robot_gesture", address="")
agent.pubsocket = pubsocket agent.pubsocket = pubsocket
agent.subsocket = subsocket agent.subsocket = subsocket
agent.repsocket = repsocket agent.repsocket = repsocket
@@ -415,7 +419,7 @@ async def test_stop_closes_sockets():
async def test_initialization_with_custom_gesture_data(): async def test_initialization_with_custom_gesture_data():
"""Agent can be initialized with custom gesture data.""" """Agent can be initialized with custom gesture data."""
custom_gestures = ["custom1", "custom2", "custom3"] 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 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.recv = recv_once
fake_repsocket.send = AsyncMock() 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.repsocket = fake_repsocket
agent.logger = MagicMock() agent.logger = MagicMock()
agent._running = True agent._running = True