Add test for GGV2 client using mocked rpc connection

.gitignore

@@ -1,2 +1,3 @@
 docs/.buildinfo
 docs/.doctrees/
+__pycache__/

test/test_ggv2.py

@@ -0,0 +1,85 @@
+import concurrent.futures
+import threading
+from unittest import TestCase
+from unittest.mock import patch
+
+from awsiot.greengrasscoreipc.client import SubscribeToTopicStreamHandler
+from awsiot.greengrasscoreipc.model import CreateLocalDeploymentResponse, SubscribeToTopicResponse, \
+    SubscriptionResponseMessage, BinaryMessage
+
+TIMEOUT = 10.0 # seconds
+
+
+class GGV2Test(TestCase):
+
+    def _mock_operation(self, mock_op, response):
+        activate_fut = concurrent.futures.Future()
+        activate_fut.set_result(None)
+        mock_op.activate.return_value = activate_fut
+        response_fut = concurrent.futures.Future()
+        response_fut.set_result(response)
+        mock_op.get_response.return_value = response_fut
+        return mock_op
+
+    @patch('awsiot.greengrasscoreipc.client.GreengrassCoreIPCClient')
+    @patch('awsiot.greengrasscoreipc.client.CreateLocalDeploymentOperation')
+    @patch('awsiot.greengrasscoreipc.client.SubscribeToTopicOperation')
+    def test_connect(self, mock_client, mock_deployment_op, mock_subscribe_op):
+        from awsiot.greengrasscoreipc.clientv2 import GreengrassCoreIPCClientV2 as Client
+        c = Client(client=mock_client)
+
+        self._mock_operation(mock_deployment_op, CreateLocalDeploymentResponse(deployment_id="deployment"))
+        mock_client.new_create_local_deployment.return_value = mock_deployment_op
+        resp = c.create_local_deployment()
+        self.assertEqual("deployment", resp.deployment_id)
+
+        # Verify subscription works and callback is called on the executor thread
+        self._mock_operation(mock_subscribe_op, SubscribeToTopicResponse(topic_name="abc"))
+        mock_client.new_subscribe_to_topic.return_value = mock_subscribe_op
+
+        subscription_fut = concurrent.futures.Future()
+        thread_id_fut = concurrent.futures.Future()
+
+        def on_stream_event(r):
+            subscription_fut.set_result(r)
+            thread_id_fut.set_result(threading.get_ident())
+        resp, op = c.subscribe_to_topic(topic="abc", on_stream_event=on_stream_event)
+        self.assertEqual("abc", resp.topic_name)
+
+        sub_handler = mock_client.new_subscribe_to_topic.call_args[0][0]
+        sub_handler.on_stream_event(SubscriptionResponseMessage(binary_message=BinaryMessage(message="xyz")))
+
+        self.assertEqual("xyz".encode("utf-8"), subscription_fut.result(TIMEOUT).binary_message.message)
+        self.assertNotEqual(threading.get_ident(), thread_id_fut.result(TIMEOUT))
+
+        # Verify that when using the stream_handler option, the callback is run in the executor
+
+        subscription_fut = concurrent.futures.Future()
+        thread_id_fut = concurrent.futures.Future()
+
+        class handler(SubscribeToTopicStreamHandler):
+            def on_stream_event(self, event):
+                on_stream_event(event)
+        resp, op = c.subscribe_to_topic(topic="abc", stream_handler=handler())
+        self.assertEqual("abc", resp.topic_name)
+
+        sub_handler = mock_client.new_subscribe_to_topic.call_args[0][0]
+        sub_handler.on_stream_event(SubscriptionResponseMessage(binary_message=BinaryMessage(message="xyz")))
+
+        self.assertEqual("xyz".encode("utf-8"), subscription_fut.result(TIMEOUT).binary_message.message)
+        self.assertNotEqual(threading.get_ident(), thread_id_fut.result(TIMEOUT))
+
+        # Remove executor from client to verify that we are not running the callback in a different thread
+        c = Client(client=mock_client, executor=None)
+
+        subscription_fut = concurrent.futures.Future()
+        thread_id_fut = concurrent.futures.Future()
+
+        resp, op = c.subscribe_to_topic(topic="abc", on_stream_event=on_stream_event)
+        self.assertEqual("abc", resp.topic_name)
+
+        sub_handler = mock_client.new_subscribe_to_topic.call_args[0][0]
+        sub_handler.on_stream_event(SubscriptionResponseMessage(binary_message=BinaryMessage(message="xyz")))
+
+        self.assertEqual("xyz".encode("utf-8"), subscription_fut.result(TIMEOUT).binary_message.message)
+        self.assertEqual(threading.get_ident(), thread_id_fut.result(TIMEOUT))