Merge pull request #28 from Unity-Technologies/yamato-test

AIRO-331a Adding smoke tests and Yamato hooks

Author: mrpropellers

.yamato/yamato-config.yml

@@ -0,0 +1,16 @@
+name: Endpoint Unit Tests
+agent:
+    type: Unity::VM
+    image: robotics/ci-ubuntu20:latest
+    flavor: i1.medium
+commands:
+    - source /opt/ros/noetic/setup.bash && echo "ROS_DISTRO == $ROS_DISTRO" &&
+        cd .. && mkdir -p catkin_ws/src && cp -r ./ROS-TCP-Endpoint catkin_ws/src &&
+        cd catkin_ws && catkin_make && source devel/setup.bash && python3 -m pytest
+triggers:
+    cancel_old_ci: true
+    expression: |
+        (pull_request.target eq "main"  AND
+        NOT pull_request.push.changes.all match "**/*.md") OR
+        (push.branch eq "dev" AND
+        NOT push.changes.all match "**/*.md")

src/ros_tcp_endpoint/subscriber.py

@@ -32,6 +32,7 @@ def __init__(self, topic, message_class, tcp_server, queue_size=10):
             message_class: The message class in catkin workspace
             queue_size:    Max number of entries to maintain in an outgoing queue
         """
+        RosReceiver.__init__(self)
         self.topic = topic
         self.node_name = "{}_subscriber".format(topic)
         self.msg = message_class

test/__init__.py

@@ -0,0 +1 @@
+

test/test_smoke.py

@@ -0,0 +1,23 @@
+import pytest
+from unittest import mock
+import ros_tcp_endpoint
+
+@mock.patch('ros_tcp_endpoint.publisher.rospy')
+def test_pub_constructor(mock_pub_rospy):
+    mock_msg_pub = mock.Mock()
+    pub = ros_tcp_endpoint.RosPublisher("/fake_topic", mock_msg_pub)
+    # Message is constructed in publisher, stored as object instance
+    mock_msg_pub.assert_called_once()
+
+@mock.patch('ros_tcp_endpoint.subscriber.rospy')
+def test_sub_constructor(mock_sub_rospy):
+    mock_msg_sub = mock.Mock()
+    sub = ros_tcp_endpoint.RosSubscriber("/fake_topic2", mock_msg_sub, None)
+    # Message is NOT constructed in subscriber, stored as class reference
+    mock_msg_sub.assert_not_called()
+
+@mock.patch('ros_tcp_endpoint.service.rospy')
+def test_srv_constructor(mock_srv_rospy):
+    mock_srv = mock.Mock()
+    src = ros_tcp_endpoint.RosService("fake_service", mock_srv)
+    mock_srv._request_class.assert_called_once()