Merge pull request #82 from TaskarCenterAtUW/feature-stop-at-one-multiprocessing

Feature stop at one multiprocessing

Author: sujata-m

requirements.txt

@@ -1,6 +1,6 @@
 fastapi==0.88.0
 pydantic==1.10.4
-python-ms-core==0.0.24
+python-ms-core==0.0.25
 uvicorn==0.20.0
 html_testRunner==1.2.1
 geopandas==0.14.4

src/osw_validator.py

@@ -49,13 +49,16 @@ def process(message) -> None:
                 self.validate(received_message=upload_message)
 
         self.listening_topic.subscribe(subscription=self.subscription_name, callback=process, max_receivable_messages=self._settings.max_receivable_messages)
+        if self._settings.max_receivable_messages > 0:
+            logger.info('Listener finished processing available messages; stopping server/container.')
+            self._stop_server_and_container(delay_seconds=2)
 
     def validate(self, received_message: Upload):
         tdei_record_id: str = ''
         status_sent = False
         try:
             tdei_record_id = received_message.message_id
-            logger.info(f'Received message for : {tdei_record_id} Message received for OSW validation !')
+            logger.info(f'Received message for : {tdei_record_id} Message received for OSW validation! Core version: {Core.__version__}')
 
             if received_message.data.file_upload_path is None:
                 error_msg = 'Request does not have valid file path specified.'
@@ -86,10 +89,9 @@ def validate(self, received_message: Upload):
             status_sent = True
         finally:
             if status_sent:
-                logger.info('Triggering server shutdown after status send.')
+                logger.info('Validation status sent for %s.', tdei_record_id)
             else:
-                logger.warning('Server shutdown skipped because status was not sent.')
-            self._stop_server_and_container(delay_seconds=2)
+                logger.warning('Validation status was not sent for %s.', tdei_record_id)
 
     def send_status(self, result: ValidationResult, upload_message: Upload):
         upload_message.data.success = result.is_valid

tests/unit_tests/test_service.py

@@ -1,5 +1,5 @@
 import unittest
-from unittest.mock import patch, MagicMock
+from unittest.mock import patch, MagicMock, ANY
 from src.osw_validator import OSWValidator
 from src.models.queue_message_content import Upload
 from src.models.queue_message_content import ValidationResult
@@ -16,6 +16,7 @@ def setUp(self, mock_core, mock_settings, mock_thread):
         mock_settings.return_value.event_bus.upload_topic = 'test_request_topic'
         mock_settings.return_value.event_bus.validation_topic = 'test_response_topic'
         mock_settings.return_value.max_concurrent_messages = 10
+        mock_settings.return_value.max_receivable_messages = -1
         mock_settings.return_value.get_download_directory.return_value = '/tmp'
         mock_settings.return_value.event_bus.container_name = 'test_container'
 
@@ -63,6 +64,18 @@ def test_subscribe_with_valid_message(self, mock_upload, mock_queue_message):
         # Assert
         self.service.validate.assert_called_once_with(received_message=mock_upload_message)
 
+    def test_start_listening_stops_container_after_subscribe_returns(self):
+        self.service._settings.max_receivable_messages = 1
+
+        self.service.start_listening()
+
+        self.service.listening_topic.subscribe.assert_called_once_with(
+            subscription=self.service.subscription_name,
+            callback=ANY,
+            max_receivable_messages=1,
+        )
+        self.service._stop_server_and_container.assert_called_once_with(delay_seconds=2)
+
     @patch('src.osw_validator.Validation')
     def test_validate_with_valid_file_path(self, mock_validation):
         # Arrange