114 lines
3.3 KiB
Python
114 lines
3.3 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
This program has been developed by students from the bachelor Computer Science at Utrecht
|
|
University within the Software Project course.
|
|
© Copyright Utrecht University (Department of Information and Computing Sciences)
|
|
"""
|
|
|
|
import mock
|
|
import pytest
|
|
import zmq
|
|
|
|
from robot_interface.endpoints.main_receiver import MainReceiver
|
|
|
|
|
|
@pytest.fixture
|
|
def zmq_context():
|
|
"""
|
|
A pytest fixture that creates and yields a ZMQ context.
|
|
|
|
:return: An initialized ZeroMQ context.
|
|
:rtype: zmq.Context
|
|
"""
|
|
context = zmq.Context()
|
|
yield context
|
|
|
|
|
|
def test_handle_ping(zmq_context):
|
|
"""
|
|
Tests the receiver's ability to handle the "ping" endpoint with data.
|
|
"""
|
|
receiver = MainReceiver(zmq_context)
|
|
response = receiver.handle_message({"endpoint": "ping", "data": "pong"})
|
|
|
|
assert "endpoint" in response
|
|
assert response["endpoint"] == "ping"
|
|
assert "data" in response
|
|
assert response["data"] == "pong"
|
|
|
|
|
|
def test_handle_ping_none(zmq_context):
|
|
"""
|
|
Tests the receiver's ability to handle the ping endpoint when the
|
|
data field is explicitly set to None.
|
|
"""
|
|
receiver = MainReceiver(zmq_context)
|
|
response = receiver.handle_message({"endpoint": "ping", "data": None})
|
|
|
|
assert "endpoint" in response
|
|
assert response["endpoint"] == "ping"
|
|
assert "data" in response
|
|
assert response["data"] == None
|
|
|
|
|
|
@mock.patch("robot_interface.endpoints.main_receiver.state")
|
|
def test_handle_negotiate_ports(mock_state, zmq_context):
|
|
"""
|
|
Tests the handling of the "negotiate/ports" endpoint.
|
|
"""
|
|
receiver = MainReceiver(zmq_context)
|
|
mock_state.sockets = [receiver]
|
|
|
|
response = receiver.handle_message({"endpoint": "negotiate/ports", "data": None})
|
|
|
|
assert "endpoint" in response
|
|
assert response["endpoint"] == "negotiate/ports"
|
|
assert "data" in response
|
|
assert isinstance(response["data"], list)
|
|
for port in response["data"]:
|
|
assert "id" in port
|
|
assert isinstance(port["id"], str)
|
|
assert "port" in port
|
|
assert isinstance(port["port"], int)
|
|
assert "bind" in port
|
|
assert isinstance(port["bind"], bool)
|
|
|
|
assert any(port["id"] == "main" for port in response["data"])
|
|
|
|
|
|
def test_handle_unimplemented_endpoint(zmq_context):
|
|
"""
|
|
Tests that the receiver correctly handles a request to a completely
|
|
unknown or non-existent endpoint.
|
|
"""
|
|
receiver = MainReceiver(zmq_context)
|
|
response = receiver.handle_message({
|
|
"endpoint": "some_endpoint_that_definitely_does_not_exist",
|
|
"data": None,
|
|
})
|
|
|
|
assert "endpoint" in response
|
|
assert response["endpoint"] == "error"
|
|
assert "data" in response
|
|
assert isinstance(response["data"], str)
|
|
|
|
|
|
def test_handle_unimplemented_negotiation_endpoint(zmq_context):
|
|
"""
|
|
Tests handling a request to an unknown sub-endpoint within a known
|
|
group
|
|
|
|
The expected behavior is to return a specific "negotiate/error" response
|
|
with a descriptive error string.
|
|
"""
|
|
receiver = MainReceiver(zmq_context)
|
|
response = receiver.handle_message({
|
|
"endpoint": "negotiate/but_some_subpath_that_definitely_does_not_exist",
|
|
"data": None,
|
|
})
|
|
|
|
assert "endpoint" in response
|
|
assert response["endpoint"] == "negotiate/error"
|
|
assert "data" in response
|
|
assert isinstance(response["data"], str)
|