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

chore: fix style guide max characters

Browse Source
This commit is contained in:
Björn Otgaar
2025-10-23 17:12:49 +02:00
parent c1217a9017
commit 31e77de26b
4 changed files with 83 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -10,10 +10,12 @@ from control_backend.schemas.ri_message import SpeechCommand
async def test_setup_bind(monkeypatch):
"""Test setup with bind=True"""
fake_socket = MagicMock()
monkeypatch.setattr("control_backend.agents.ri_command_agent.context.socket", lambda _: fake_socket)
monkeypatch.setattr("control_backend.agents.ri_command_agent.context.socket",
lambda _: fake_socket)
agent = RICommandAgent("test@server", "password", address="tcp://localhost:5555", bind=True)
monkeypatch.setattr("control_backend.agents.ri_command_agent.settings", MagicMock(zmq_settings=MagicMock(internal_comm_address="tcp://internal:1234")))
monkeypatch.setattr("control_backend.agents.ri_command_agent.settings",
MagicMock(zmq_settings=MagicMock(internal_comm_address="tcp://internal:1234")))
await agent.setup()
@@ -30,10 +32,12 @@ async def test_setup_bind(monkeypatch):
async def test_setup_connect(monkeypatch):
"""Test setup with bind=False"""
fake_socket = MagicMock()
monkeypatch.setattr("control_backend.agents.ri_command_agent.context.socket", lambda _: fake_socket)
monkeypatch.setattr("control_backend.agents.ri_command_agent.context.socket",
lambda _: fake_socket)
agent = RICommandAgent("test@server", "password", address="tcp://localhost:5555", bind=False)
monkeypatch.setattr("control_backend.agents.ri_command_agent.settings", MagicMock(zmq_settings=MagicMock(internal_comm_address="tcp://internal:1234")))
monkeypatch.setattr("control_backend.agents.ri_command_agent.settings",
MagicMock(zmq_settings=MagicMock(internal_comm_address="tcp://internal:1234")))
await agent.setup()
@@ -45,7 +49,8 @@ async def test_send_commands_behaviour_valid_message():
"""Test behaviour with valid JSON message"""
fake_socket = AsyncMock()
message_dict = {"message": "hello"}
fake_socket.recv_multipart = AsyncMock(return_value=(b"command", json.dumps(message_dict).encode("utf-8")))
fake_socket.recv_multipart = AsyncMock(return_value=(b"command",
json.dumps(message_dict).encode("utf-8")))
fake_socket.send_json = AsyncMock()
agent = RICommandAgent("test@server", "password")

84
test/integration/agents/test_ri_communication_agent.py
View File

@@ -69,15 +69,18 @@ async def test_setup_creates_socket_and_negotiate_1(monkeypatch):
fake_socket.recv_json = fake_json_correct_negototiate_1()
# Mock context.socket to return our fake socket
monkeypatch.setattr("control_backend.agents.ri_communication_agent.context.socket", lambda _: fake_socket)
monkeypatch.setattr("control_backend.agents.ri_communication_agent.context.socket",
lambda _: fake_socket)
# Mock RICommandAgent agent startup
with patch("control_backend.agents.ri_communication_agent.RICommandAgent", autospec=True) as MockCommandAgent:
with patch("control_backend.agents.ri_communication_agent.RICommandAgent",
autospec=True) as MockCommandAgent:
fake_agent_instance = MockCommandAgent.return_value
fake_agent_instance.start = AsyncMock()
# --- Act ---
agent = RICommunicationAgent("test@server", "password", address="tcp://localhost:5555", bind=False)
agent = RICommunicationAgent("test@server", "password",
address="tcp://localhost:5555", bind=False)
await agent.setup()
# --- Assert ---
@@ -105,15 +108,18 @@ async def test_setup_creates_socket_and_negotiate_2(monkeypatch):
fake_socket.recv_json = fake_json_correct_negototiate_2()
# Mock context.socket to return our fake socket
monkeypatch.setattr("control_backend.agents.ri_communication_agent.context.socket", lambda _: fake_socket)
monkeypatch.setattr("control_backend.agents.ri_communication_agent.context.socket",
lambda _: fake_socket)
# Mock RICommandAgent agent startup
with patch("control_backend.agents.ri_communication_agent.RICommandAgent", autospec=True) as MockCommandAgent:
with patch("control_backend.agents.ri_communication_agent.RICommandAgent",
autospec=True) as MockCommandAgent:
fake_agent_instance = MockCommandAgent.return_value
fake_agent_instance.start = AsyncMock()
# --- Act ---
agent = RICommunicationAgent("test@server", "password", address="tcp://localhost:5555", bind=False)
agent = RICommunicationAgent("test@server", "password",
address="tcp://localhost:5555", bind=False)
await agent.setup()
# --- Assert ---
@@ -141,20 +147,23 @@ async def test_setup_creates_socket_and_negotiate_3(monkeypatch, caplog):
fake_socket.recv_json = fake_json_wrong_negototiate_1()
# Mock context.socket to return our fake socket
monkeypatch.setattr("control_backend.agents.ri_communication_agent.context.socket", lambda _: fake_socket)
monkeypatch.setattr("control_backend.agents.ri_communication_agent.context.socket",
lambda _: fake_socket)
# Mock RICommandAgent agent startup
# We are sending wrong negotiation info to the communication agent, so we should retry and expect a
# better response, within a limited time.
with patch("control_backend.agents.ri_communication_agent.RICommandAgent", autospec=True) as MockCommandAgent:
with patch("control_backend.agents.ri_communication_agent.RICommandAgent",
autospec=True) as MockCommandAgent:
fake_agent_instance = MockCommandAgent.return_value
fake_agent_instance.start = AsyncMock()
# --- Act ---
with caplog.at_level("ERROR"):
agent = RICommunicationAgent("test@server", "password", address="tcp://localhost:5555", bind=False)
agent = RICommunicationAgent("test@server", "password",
address="tcp://localhost:5555", bind=False)
await agent.setup(max_retries=1)
# --- Assert ---
@@ -179,15 +188,18 @@ async def test_setup_creates_socket_and_negotiate_4(monkeypatch):
fake_socket.recv_json = fake_json_correct_negototiate_3()
# Mock context.socket to return our fake socket
monkeypatch.setattr("control_backend.agents.ri_communication_agent.context.socket", lambda _: fake_socket)
monkeypatch.setattr("control_backend.agents.ri_communication_agent.context.socket",
lambda _: fake_socket)
# Mock RICommandAgent agent startup
with patch("control_backend.agents.ri_communication_agent.RICommandAgent", autospec=True) as MockCommandAgent:
with patch("control_backend.agents.ri_communication_agent.RICommandAgent",
autospec=True) as MockCommandAgent:
fake_agent_instance = MockCommandAgent.return_value
fake_agent_instance.start = AsyncMock()
# --- Act ---
agent = RICommunicationAgent("test@server", "password", address="tcp://localhost:5555", bind=True)
agent = RICommunicationAgent("test@server", "password",
address="tcp://localhost:5555", bind=True)
await agent.setup()
# --- Assert ---
@@ -215,15 +227,18 @@ async def test_setup_creates_socket_and_negotiate_5(monkeypatch):
fake_socket.recv_json = fake_json_correct_negototiate_4()
# Mock context.socket to return our fake socket
monkeypatch.setattr("control_backend.agents.ri_communication_agent.context.socket", lambda _: fake_socket)
monkeypatch.setattr("control_backend.agents.ri_communication_agent.context.socket",
lambda _: fake_socket)
# Mock RICommandAgent agent startup
with patch("control_backend.agents.ri_communication_agent.RICommandAgent", autospec=True) as MockCommandAgent:
with patch("control_backend.agents.ri_communication_agent.RICommandAgent",
autospec=True) as MockCommandAgent:
fake_agent_instance = MockCommandAgent.return_value
fake_agent_instance.start = AsyncMock()
# --- Act ---
agent = RICommunicationAgent("test@server", "password", address="tcp://localhost:5555", bind=False)
agent = RICommunicationAgent("test@server", "password",
address="tcp://localhost:5555", bind=False)
await agent.setup()
# --- Assert ---
@@ -251,15 +266,18 @@ async def test_setup_creates_socket_and_negotiate_6(monkeypatch):
fake_socket.recv_json = fake_json_correct_negototiate_5()
# Mock context.socket to return our fake socket
monkeypatch.setattr("control_backend.agents.ri_communication_agent.context.socket", lambda _: fake_socket)
monkeypatch.setattr("control_backend.agents.ri_communication_agent.context.socket",
lambda _: fake_socket)
# Mock RICommandAgent agent startup
with patch("control_backend.agents.ri_communication_agent.RICommandAgent", autospec=True) as MockCommandAgent:
with patch("control_backend.agents.ri_communication_agent.RICommandAgent",
autospec=True) as MockCommandAgent:
fake_agent_instance = MockCommandAgent.return_value
fake_agent_instance.start = AsyncMock()
# --- Act ---
agent = RICommunicationAgent("test@server", "password", address="tcp://localhost:5555", bind=False)
agent = RICommunicationAgent("test@server", "password",
address="tcp://localhost:5555", bind=False)
await agent.setup()
# --- Assert ---
@@ -287,20 +305,23 @@ async def test_setup_creates_socket_and_negotiate_7(monkeypatch, caplog):
fake_socket.recv_json = fake_json_invalid_id_negototiate()
# Mock context.socket to return our fake socket
monkeypatch.setattr("control_backend.agents.ri_communication_agent.context.socket", lambda _: fake_socket)
monkeypatch.setattr("control_backend.agents.ri_communication_agent.context.socket",
lambda _: fake_socket)
# Mock RICommandAgent agent startup
# We are sending wrong negotiation info to the communication agent, so we should retry and expect a
# better response, within a limited time.
with patch("control_backend.agents.ri_communication_agent.RICommandAgent", autospec=True) as MockCommandAgent:
with patch("control_backend.agents.ri_communication_agent.RICommandAgent",
autospec=True) as MockCommandAgent:
fake_agent_instance = MockCommandAgent.return_value
fake_agent_instance.start = AsyncMock()
# --- Act ---
with caplog.at_level("WARNING"):
agent = RICommunicationAgent("test@server", "password", address="tcp://localhost:5555", bind=False)
agent = RICommunicationAgent("test@server", "password",
address="tcp://localhost:5555", bind=False)
await agent.setup(max_retries=1)
# --- Assert ---
@@ -323,15 +344,18 @@ async def test_setup_creates_socket_and_negotiate_timeout(monkeypatch, caplog):
fake_socket.recv_json = AsyncMock(side_effect=asyncio.TimeoutError)
# Mock context.socket to return our fake socket
monkeypatch.setattr("control_backend.agents.ri_communication_agent.context.socket", lambda _: fake_socket)
monkeypatch.setattr("control_backend.agents.ri_communication_agent.context.socket",
lambda _: fake_socket)
with patch("control_backend.agents.ri_communication_agent.RICommandAgent", autospec=True) as MockCommandAgent:
with patch("control_backend.agents.ri_communication_agent.RICommandAgent",
autospec=True) as MockCommandAgent:
fake_agent_instance = MockCommandAgent.return_value
fake_agent_instance.start = AsyncMock()
# --- Act ---
with caplog.at_level("WARNING"):
agent = RICommunicationAgent("test@server", "password", address="tcp://localhost:5555", bind=False)
agent = RICommunicationAgent("test@server", "password",
address="tcp://localhost:5555", bind=False)
await agent.setup(max_retries=1)
# --- Assert ---
@@ -453,7 +477,8 @@ async def test_setup_unexpected_exception(monkeypatch, caplog):
lambda _: fake_socket
)
agent = RICommunicationAgent("test@server", "password", address="tcp://localhost:5555", bind=False)
agent = RICommunicationAgent("test@server", "password",
address="tcp://localhost:5555", bind=False)
with caplog.at_level("ERROR"):
await agent.setup(max_retries=1)
@@ -468,7 +493,8 @@ async def test_setup_unpacking_exception(monkeypatch, caplog):
fake_socket.send_json = AsyncMock()
# Make recv_json return malformed negotiation data to trigger unpacking exception
malformed_data = {"endpoint": "negotiate/ports", "data": [ {"id": "main"} ]} # missing 'port' and 'bind'
malformed_data = {"endpoint": "negotiate/ports",
"data": [ {"id": "main"} ]} # missing 'port' and 'bind'
fake_socket.recv_json = AsyncMock(return_value=malformed_data)
# Patch context.socket
@@ -478,11 +504,13 @@ async def test_setup_unpacking_exception(monkeypatch, caplog):
)
# Patch RICommandAgent so it won't actually start
with patch("control_backend.agents.ri_communication_agent.RICommandAgent", autospec=True) as MockCommandAgent:
with patch("control_backend.agents.ri_communication_agent.RICommandAgent",
autospec=True) as MockCommandAgent:
fake_agent_instance = MockCommandAgent.return_value
fake_agent_instance.start = AsyncMock()
agent = RICommunicationAgent("test@server", "password", address="tcp://localhost:5555", bind=False)
agent = RICommunicationAgent("test@server", "password",
address="tcp://localhost:5555", bind=False)
# --- Act & Assert ---
with caplog.at_level("ERROR"):