I have a setup where I am developing a roslibpy.Ros client wrapper. I am not a robotics developer so it might be that I am missing some basic things with the Ros client.
The setup is as follows:
I have a RosBridgeClientWrapper:
import roslibpy
class _PayloadRosBridgeClient():
def __init__(self, hostname: str, port: int) -> None:
self._hostname = hostname
self._port = port
self._ros = None
self._timeout = 10
def __del__(self):
self._ros.terminate()
def connect(self):
if self._ros is None:
self._ros = Ros(self._hostname, self._port)
self._ros.run(self._timeout)
else:
self._ros.connect()
time.sleep(1)
....
This client is being used by a PayloadHandlerClass which has other types of clients to interface with the physical device:
class _PayloadHandler_4_1(PayloadHandlerInterface.PayloadHandlerInterface):
def __init__(self, hostname: str, username: str, password: str, ssh_port: int, ros_bridge_port: int):
self._lock = threading.Lock()
self._ros_bridge_client = PayloadRosBridgeClient.make(hostname, ros_bridge_port)
self._ssh_client = PayloadSSHClient.make(
hostname, username, password, PathConfigurations.Paths(), ssh_port)
self._username = username
self._password = password
def get_version(self) -> None:
return PayloadVersions.Version_4_1
def connect(self) -> bool:
try:
self._ros_bridge_client.connect()
self._ssh_client.connect()
return True
except Exception:
print("PayloadHandler_4_1: could not connect to ros or ssh service")
return False
...
To automate testing of this PayloadHandler class, I created a unittest class:
class PayloadHandler_4_1_IntegrationTest(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls._payload_address = socket.gethostbyname("container_name")
...
cls._payload_manager = PayloadHandler_4_1.make(
cls._payload_address, cls._payload_username, cls._payload_password, cls._ssh_port, cls._ros_port)
def setUp(self):
PayloadHandler_4_1_IntegrationTest._payload_manager.connect()
def tearDown(self):
if PayloadHandler_4_1_IntegrationTest._payload_manager.is_connected():
PayloadHandler_4_1_IntegrationTest._payload_manager.disconnect()
def test_make(self):
result = PayloadHandler_4_1_IntegrationTest._payload_manager
self.assertIsInstance(result, PayloadHandler_4_1._PayloadHandler_4_1)
def test_connect(self):
handler = PayloadHandler_4_1_IntegrationTest._payload_manager
self.assertTrue(handler.is_connected())
The issue that I am facing is due to the roslibpy.Ros client. It seems that for some reason it can not connect multiple times in a row. I am running a Ros instance in a docker container to test against it. When I run my test class, the test_make method runs fine, the client is connected by the setUp method, the test runs against the container, then the tearDown method disconnects the client. All other following tests however fail. In the test_connect method for example, the Ros client attempts connection to the container, and I can see that by doing rostopic echo /connected_clients I can see that new clients have connected, however the roslibpy.Ros client, notifies me that it failed to connect to the service and fails with error "Could not connect to Ros".
Another piece of information is that I am running these in 2 separate containers. When I run the tests on my local machine against a docker container running on my local machine, everything seems to run fine. If I run my test setup in my gitlab-runners however things fail. My unit tests run on one container, and the Ros instance runs in a different container. I have checked that the 2 containers can see eachother by ssh-ing into them and manually testing the connection and all the services and everything runs as expected. When I run the unit tests however they fail. Any input on this would be very appreciated.