Merge pull request #5 from cloudblue/CP_1

Cumulative fixes

Author: ffaraoneim

Dockerfile

@@ -18,6 +18,9 @@ RUN poetry build
 
 RUN pip install dist/*.whl
 
-WORKDIR /
-
 RUN rm -rf /install_temp
+
+COPY ./entrypoint.sh /entrypoint.sh
+RUN chmod 755 /entrypoint.sh
+
+ENTRYPOINT [ "/entrypoint.sh" ]

connect/eaas/helpers.py

@@ -13,19 +13,6 @@
 from connect.eaas.exceptions import EaaSError
 
 
-def install_extension(root_dir):
-    result = subprocess.run(
-        ['poetry', 'install'],
-        cwd=root_dir,
-        stdin=subprocess.DEVNULL,
-        start_new_session=True,
-    )
-    try:
-        result.check_returncode()
-    except subprocess.CalledProcessError:
-        raise EaaSError(result.stderr.decode())
-
-
 def get_container_id():
     result = subprocess.run(
         ['cat', '/proc/1/cpuset'],

connect/eaas/main.py

@@ -10,7 +10,6 @@
 import pathlib
 import signal
 
-from connect.eaas.helpers import install_extension
 from connect.eaas.worker import Worker
 
 
@@ -42,18 +41,20 @@ def configure_logger(debug):
                     'handlers': ['console'],
                     'level': 'DEBUG' if debug else 'INFO',
                 },
+                'eaas.extension': {
+                    'handlers': ['console'],
+                    'level': 'DEBUG' if debug else 'INFO',
+                },
             },
         },
     )
 
 
 def start(data):
     logger.info('Starting Connect EaaS runtime....')
-    logger.info('Installing the extension package...')
-    install_extension(data.extension_dir)
-    logger.info('The extension has been installed, starting the control worker...')
     if data.unsecure:
         logger.warning('Websocket connections will be established using unsecure protocol (ws).')
+
     worker = Worker(secure=not data.unsecure)
     loop = asyncio.get_event_loop()
     loop.add_signal_handler(

connect/eaas/manager.py

@@ -41,9 +41,8 @@ class TasksManager:
     Once completed, it will push the result to a queue from which, the
     result_sender task will pick it and send the result to the backend.
     """
-    def __init__(self, worker, extension):
+    def __init__(self, worker):
         self.worker = worker
-        self.extension = extension
         self.background_executor = ThreadPoolExecutor()
         self.interactive_executor = ThreadPoolExecutor()
         self.run_event = asyncio.Event()
@@ -87,17 +86,22 @@ async def submit_task(self, data):
         object_id = data.object_id
         task_type = data.task_type
         method_name = TASK_TYPE_EXT_METHOD_MAP[task_type]
-        method = getattr(self.extension, method_name)
+        extension = self.worker.get_extension()
+        method = getattr(extension, method_name)
         logger.debug(f'invoke {method_name}')
         self.running_tasks += 1
-        try:
-            request = await self.get_request(object_id, task_type)
-        except ClientError as e:
-            logger.warning(f'Cannot retrieve object {data.object_id} for task {data.task_id}')
-            future = Future()
-            future.set_exception(e)
-            asyncio.create_task(self.enqueue_result(data, future))
-            return
+        request = None
+        if data.task_category == TaskCategory.BACKGROUND:
+            try:
+                request = await self.get_request(object_id, task_type)
+            except ClientError as e:
+                logger.warning(f'Cannot retrieve object {data.object_id} for task {data.task_id}')
+                future = Future()
+                future.set_exception(e)
+                asyncio.create_task(self.enqueue_result(data, future))
+                return
+        else:
+            request = data.data
 
         if inspect.iscoroutinefunction(method):
             future = asyncio.create_task(method(request))

connect/eaas/worker.py

@@ -118,12 +118,7 @@ async def stop_tasks_manager(self):
 
     def start_tasks_manager(self):
         logger.info('Starting tasks worker...')
-        extension = self.extension_class(
-            self.get_client(),
-            self.get_extension_logger(self.logging_api_key),
-            self.extension_config,
-        )
-        self.tasks_manager = TasksManager(self, extension)
+        self.tasks_manager = TasksManager(self)
         self.tasks_manager.start()
         logger.info('Task worker started')
 
@@ -134,6 +129,13 @@ def ensure_tasks_manager_running(self):
     def get_url(self):
         return f'{self.base_ws_url}/{self.environment_id}/{self.instance_id}'
 
+    def get_extension(self):
+        return self.extension_class(
+            self.get_client(),
+            self.get_extension_logger(self.logging_api_key),
+            self.extension_config,
+        )
+
     async def run(self):  # noqa: CCR001
         """
         Main loop for the websocket connection.
@@ -195,12 +197,8 @@ async def configuration(self, data):
         It will stop the tasks manager so the extension can be
         reconfigured, then restart the tasks manager.
         """
-        self.paused = True
-        await self.stop_tasks_manager()
         self.extension_config = data.configuration
         self.logging_api_key = data.logging_api_key
-        self.paused = False
-        self.start_tasks_manager()
 
     async def pause(self):
         """

entrypoint.sh

@@ -0,0 +1,24 @@
+#!/bin/bash
+set -e
+
+EXTENSION_DIR=${EXTENSION_DIR:-'/extension'}
+
+DIST_DIR=$EXTENSION_DIR/dist
+
+echo "Installing the extension package from $EXTENSION_DIR..."
+
+pushd .
+
+cd $EXTENSION_DIR
+
+if [ -d "$DIST_DIR" ]; then rm -Rf $DIST_DIR; fi
+
+poetry build
+
+pip install -U pip && pip install $DIST_DIR/*.whl
+
+popd
+
+echo "Extension installed."
+
+exec "$@"

tests/test_helpers.py

@@ -15,33 +15,9 @@
     get_environment,
     get_extension_class,
     get_extension_type,
-    install_extension,
 )
 
 
-def test_install_extension(mocker):
-    result = mocker.MagicMock()
-    result.returnvalue = 0
-    mocked = mocker.patch('connect.eaas.helpers.subprocess.run', return_value=result)
-    install_extension('/root_dir')
-
-    assert mocked.mock_calls[0].args[0] == ['poetry', 'install']
-    assert mocked.mock_calls[0].kwargs['cwd'] == '/root_dir'
-
-
-def test_install_extension_ko(mocker):
-    result = mocker.MagicMock()
-    result.check_returncode = mocker.MagicMock(
-        side_effect=subprocess.CalledProcessError(128, cmd=[]),
-    )
-    result.stderr = 'error message'.encode('utf-8')
-    mocker.patch('connect.eaas.helpers.subprocess.run', return_value=result)
-    with pytest.raises(EaaSError) as cv:
-        install_extension('/root_dir')
-
-    assert str(cv.value) == 'error message'
-
-
 def test_get_container_id(mocker):
     result = mocker.MagicMock()
     result.returnvalue = 0

tests/test_main.py

@@ -31,7 +31,7 @@ def process_asset_purchase_request(self, request):
         def validate_asset_purchase_request(self, request):
             pass
     run_mock = mocker.AsyncMock()
-    mocker.patch('connect.eaas.main.install_extension')
+
     mocker.patch('connect.eaas.worker.get_extension_class', return_value=MyExtension)
     mocker.patch.object(Worker, 'run', run_mock)
     parsed_args = namedtuple('_Args', ('unsecure', 'extension_dir'))

tests/test_manager.py

@@ -22,7 +22,7 @@
 
 @pytest.mark.asyncio
 async def test_start_stop_is_running(mocker, caplog):
-    manager = TasksManager(mocker.MagicMock(), mocker.MagicMock())
+    manager = TasksManager(mocker.MagicMock())
     manager.start()
     assert manager.run_event.is_set() is True
     assert manager.is_running() is True
@@ -43,9 +43,10 @@ async def test_background_task_sync(mocker, extension_cls, task_type):
     extension = extension_class(None, None, None)
 
     worker = mocker.MagicMock()
+    worker.get_extension.return_value = extension
     worker.send = mocker.AsyncMock()
 
-    manager = TasksManager(worker, extension)
+    manager = TasksManager(worker)
     manager.get_request = mocker.AsyncMock(return_value={'id': 'PR-000'})
 
     manager.start()
@@ -77,9 +78,10 @@ async def test_background_task_async(mocker, extension_cls, task_type):
     extension = extension_class(None, None, None)
 
     worker = mocker.MagicMock()
+    worker.get_extension.return_value = extension
     worker.send = mocker.AsyncMock()
 
-    manager = TasksManager(worker, extension)
+    manager = TasksManager(worker)
     manager.get_request = mocker.AsyncMock(return_value={'id': 'PR-000'})
 
     manager.start()
@@ -111,9 +113,10 @@ async def test_interactive_task_sync(mocker, extension_cls, task_type):
     extension = extension_class(None, None, None)
 
     worker = mocker.MagicMock()
+    worker.get_extension.return_value = extension
     worker.send = mocker.AsyncMock()
 
-    manager = TasksManager(worker, extension)
+    manager = TasksManager(worker)
     manager.get_request = mocker.AsyncMock(return_value={'id': 'PR-000'})
 
     manager.start()
@@ -145,9 +148,10 @@ async def test_interactive_task_async(mocker, extension_cls, task_type):
     extension = extension_class(None, None, None)
 
     worker = mocker.MagicMock()
+    worker.get_extension.return_value = extension
     worker.send = mocker.AsyncMock()
 
-    manager = TasksManager(worker, extension)
+    manager = TasksManager(worker)
     manager.get_request = mocker.AsyncMock(return_value={'id': 'PR-000'})
 
     manager.start()
@@ -175,9 +179,10 @@ async def test_background_task_request_error(mocker, extension_cls):
     extension = extension_class(None, None, None)
 
     worker = mocker.MagicMock()
+    worker.get_extension.return_value = extension
     worker.send = mocker.AsyncMock()
 
-    manager = TasksManager(worker, extension)
+    manager = TasksManager(worker)
     manager.get_request = mocker.AsyncMock(side_effect=ClientError('Request not found', 404))
 
     manager.start()
@@ -199,45 +204,16 @@ async def test_background_task_request_error(mocker, extension_cls):
     worker.send.assert_awaited_once_with(json_msg)
 
 
-@pytest.mark.asyncio
-async def test_interactive_task_request_error(mocker, extension_cls):
-    extension_class = extension_cls('validate_asset_purchase_request')
-    extension = extension_class(None, None, None)
-
-    worker = mocker.MagicMock()
-    worker.send = mocker.AsyncMock()
-
-    manager = TasksManager(worker, extension)
-    manager.get_request = mocker.AsyncMock(side_effect=ClientError('Request not found', 404))
-
-    manager.start()
-
-    task = TaskPayload(
-        'TQ-000',
-        TaskCategory.INTERACTIVE,
-        TaskType.ASSET_PURCHASE_REQUEST_VALIDATION,
-        'PR-000',
-    )
-
-    await manager.submit_task(task)
-    await asyncio.sleep(.1)
-    await manager.stop()
-    message = Message(message_type=MessageType.TASK, data=task)
-    json_msg = message.to_json()
-    json_msg['data']['result'] = 'failed'
-    json_msg['data']['failure_output'] = 'Request not found'
-    worker.send.assert_awaited_once_with(json_msg)
-
-
 @pytest.mark.asyncio
 async def test_result_sender_retries(mocker, extension_cls):
     extension_class = extension_cls('process_asset_purchase_request')
     extension = extension_class(None, None, None)
 
     worker = mocker.MagicMock()
+    worker.get_extension.return_value = extension
     worker.send = mocker.AsyncMock(side_effect=[Exception('retry'), None])
 
-    manager = TasksManager(worker, extension)
+    manager = TasksManager(worker)
     manager.get_request = mocker.AsyncMock(return_value={'id': 'PR-000'})
 
     manager.start()
@@ -264,11 +240,12 @@ async def test_result_sender_max_retries_exceeded(mocker, extension_cls, caplog)
     extension = extension_class(None, None, None)
 
     worker = mocker.MagicMock()
+    worker.get_extension.return_value = extension
     worker.send = mocker.AsyncMock(
         side_effect=[Exception('retry') for _ in range(RESULT_SENDER_MAX_RETRIES)],
     )
 
-    manager = TasksManager(worker, extension)
+    manager = TasksManager(worker)
     manager.get_request = mocker.AsyncMock(return_value={'id': 'PR-000'})
 
     manager.start()