PYT-1103 No error displayed by the dvp upload command when file upload job fails

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 2.0.0-internal-001
+current_version = 2.0.0-internal-2
 commit = False
 tag = False
 parse = (?P<major>\d+)\.(?P<minor>\d+)\.(?P<patch>\d+)(\-(?P<release>[a-z]+)\-(?P<build>\d+))?

common/src/main/python/dlpx/virtualization/common/VERSION

@@ -1 +1 @@
-2.0.0-internal-001
+2.0.0-internal-2

dvp/src/main/python/dlpx/virtualization/VERSION

@@ -1 +1 @@
-2.0.0-internal-001
+2.0.0-internal-2

libs/src/main/python/dlpx/virtualization/libs/VERSION

@@ -1 +1 @@
-2.0.0-internal-001
+2.0.0-internal-2

platform/src/main/python/dlpx/virtualization/platform/VERSION

@@ -1 +1 @@
-2.0.0-internal-001
+2.0.0-internal-2

tools/src/main/python/dlpx/virtualization/_internal/VERSION

@@ -1 +1 @@
-2.0.0-internal-001
+2.0.0-internal-2

tools/src/main/python/dlpx/virtualization/_internal/cli.py

@@ -240,19 +240,22 @@ def build(plugin_config, upload_artifact, generate_only, skip_id_validation,
                     resolve_path=True),
     callback=click_util.validate_option_exists,
     help='Path to the upload artifact that was generated through build.')
+@click.option('--wait',
+              is_flag=True,
+              help='Wait for the upload job to complete before returning.')
 @click.password_option(cls=click_util.PasswordPromptIf,
                        default=DVP_CONFIG_MAP.get('password'),
                        confirmation_prompt=False,
                        help='Authenticate using the provided password.')
-def upload(engine, user, upload_artifact, password):
+def upload(engine, user, upload_artifact, password, wait):
     """
     Upload the generated upload artifact (the plugin JSON file) that was built
     to a target Delphix Engine.
     Note that the upload artifact should be the file created after running
     the build command and will fail if it's not readable or valid.
     """
     with command_error_handler():
-        upload_internal.upload(engine, user, upload_artifact, password)
+        upload_internal.upload(engine, user, upload_artifact, password, wait)
 
 
 @delphix_sdk.command()

tools/src/main/python/dlpx/virtualization/_internal/commands/upload.py

@@ -13,7 +13,7 @@
 UNKNOWN_ERR = 'UNKNOWN_ERR'
 
 
-def upload(engine, user, upload_artifact, password):
+def upload(engine, user, upload_artifact, password, wait):
     """
     Takes in the engine hostname/ip address, logs on and uploads the artifact
     passed in. The upload artifact should have been generated via the build
@@ -26,11 +26,14 @@ def upload(engine, user, upload_artifact, password):
         InvalidArtifactError
         HttpError
         UnexpectedError
+        PluginUploadJobFailed
+        PluginUploadWaitTimedOut
     """
     logger.debug('Upload parameters include'
                  ' engine: {},'
                  ' user: {},'
-                 ' upload_artifact: {}'.format(engine, user, upload_artifact))
+                 ' upload_artifact: {},'
+                 ' wait: {}'.format(engine, user, upload_artifact, wait))
     logger.info('Uploading plugin artifact {} ...'.format(upload_artifact))
 
     # Read content of upload artifact
@@ -54,4 +57,4 @@ def upload(engine, user, upload_artifact, password):
     client = delphix_client.DelphixClient(engine)
     engine_api = client.get_engine_api(content)
     client.login(engine_api, user, password)
-    client.upload_plugin(os.path.basename(upload_artifact), content)
+    client.upload_plugin(os.path.basename(upload_artifact), content, wait)

tools/src/main/python/dlpx/virtualization/_internal/delphix_client.py

@@ -4,6 +4,8 @@
 
 import json
 import logging
+import threading
+import time
 
 import requests
 from dlpx.virtualization._internal import exceptions, plugin_util
@@ -20,10 +22,14 @@ class DelphixClient(object):
     """
     __BOUNDARY = '----------boundary------'
     __UPLOAD_CONTENT = 'multipart/form-data; boundary={}'.format(__BOUNDARY)
+    __JOB_POLLING_INTERVAL = 5
+    __WAIT_TIMEOUT_SECONDS = 3600
     __cookie = None
 
-    def __init__(self, engine):
+    def __init__(self, engine, timeout=None):
         self.__engine = engine
+        if timeout is not None:
+            self.__WAIT_TIMEOUT_SECONDS = timeout
 
     def login(self, engine_api, user, password):
         """
@@ -207,13 +213,12 @@ def __download_logs(self, plugin_name, token, directory):
             for chunk in download_zip_data:
                 f.write(chunk)
 
-    def upload_plugin(self, name, content):
+    def upload_plugin(self, name, content, wait):
         """
         Takes in the plugin name and content (as a json). Attempts to upload
         the plugin onto the connected Delphix Engine. Can raise HttpPostError
         and UnexpectedError.
         """
-
         # Get the upload token.
         logger.debug('Getting token to do upload.')
         response = self.__post('delphix/toolkit/requestUploadToken')
@@ -222,10 +227,60 @@ def upload_plugin(self, name, content):
 
         logger.info('Uploading plugin {!r}.'.format(name))
         # Encode plugin content.
-        self.__post('delphix/data/upload',
-                    content_type=self.__UPLOAD_CONTENT,
-                    data=self.__encode(json.dumps(content), token, name))
-        logger.info('Plugin was successfully uploaded.')
+        upload_response = self.__post('delphix/data/upload',
+                                      content_type=self.__UPLOAD_CONTENT,
+                                      data=self.__encode(
+                                          json.dumps(content), token, name))
+        if wait:
+            self._wait_for_upload_to_complete(name,
+                                              upload_response.get('action'),
+                                              upload_response.get('job'))
+
+    def _wait_for_upload_to_complete(self, name, upload_action, upgrade_job):
+        """
+        Waits a maximum of 60 minutes for the plugin upload to complete before
+        returning from the cli command. If the upload response contains a job,
+        this means that the plugin will be upgraded. We log additional details
+        regarding events if the job exists (i.e. event code, details, and
+        action), but only if we haven't seen the job event before. We will
+        return when the job succeeds, fails, or times out. Can raise
+        PluginUploadJobFailed or PluginUploadWaitTimedOut
+        """
+        ticker = threading.Event()
+        start_time = time.time()
+        event_tuples = set()
+        failed_statuses = ('FAILED', 'SUSPENDED', 'CANCELLED')
+        while not ticker.wait(self.__JOB_POLLING_INTERVAL):
+            if upgrade_job:
+                status_response = self.__get(
+                    'delphix/action/{}/getJob'.format(upload_action)).json()
+                events = status_response.get('result').get('events')
+                for event in events:
+                    event_tuple = (event.get('timestamp'),
+                                   event.get('messageCode'))
+                    if event_tuple not in event_tuples:
+                        logger.info('Timestamp: {}, Code: {}'.format(
+                            event.get('timestamp'), event.get('messageCode')))
+                        logger.warn(event.get('messageDetails'))
+                        if event.get('messageAction') is not None:
+                            logger.warn(event.get('messageAction'))
+                        event_tuples.add(event_tuple)
+                status = status_response.get('result').get('jobState')
+            else:
+                status_response = self.__get(
+                    'delphix/action/{}'.format(upload_action)).json()
+                status = status_response.get('result').get('state')
+
+            if status == 'COMPLETED':
+                logger.warn(
+                    'Plugin {} was successfully uploaded.'.format(name))
+                ticker.set()
+            elif status in failed_statuses:
+                ticker.set()
+                raise exceptions.PluginUploadJobFailed(name)
+            elif (time.time() - start_time) > self.__WAIT_TIMEOUT_SECONDS:
+                ticker.set()
+                raise exceptions.PluginUploadWaitTimedOut(name)
 
     def download_plugin_logs(self, directory, plugin_config):
         """

tools/src/main/python/dlpx/virtualization/_internal/exceptions.py

@@ -36,6 +36,30 @@ def __init__(self, message):
         super(UserError, self).__init__(message)
 
 
+class PluginUploadJobFailed(UserError):
+    """
+    PluginUploadJobFailed is raised in the upload command if the action/job
+    that is being monitored returns with a status other than 'COMPLETED' or
+    'RUNNING'.
+    """
+    def __init__(self, plugin_name):
+        message = "Failed trying to upload plugin {}."\
+            .format(plugin_name)
+        super(PluginUploadJobFailed, self).__init__(message)
+
+
+class PluginUploadWaitTimedOut(UserError):
+    """
+    PluginUploadWaitTimedOut is raised in the upload command if the
+    action/job that is being monitored does not complete or fail within a
+    30 minute timeout window.
+    """
+    def __init__(self, plugin):
+        message = "Timed out waiting for upload of plugin {} to complete."\
+            .format(plugin)
+        super(PluginUploadWaitTimedOut, self).__init__(message)
+
+
 class PathIsAbsoluteError(UserError):
     def __init__(self, path):
         self.path = path

tools/src/test/python/dlpx/virtualization/_internal/commands/test_build.py

@@ -66,9 +66,9 @@ def test_build_success(mock_relative_path, mock_install_deps,
         'dlpx.virtualization._internal.plugin_dependency_util.install_deps')
     @mock.patch('os.path.isabs', return_value=False)
     def test_build_success_non_default_output_file(
-        mock_relative_path, mock_install_deps, mock_generate_python,
-        mock_import_plugin, plugin_config_file, artifact_file,
-        artifact_content, codegen_gen_py_inputs):
+            mock_relative_path, mock_install_deps, mock_generate_python,
+            mock_import_plugin, plugin_config_file, artifact_file,
+            artifact_content, codegen_gen_py_inputs):
         gen_py = codegen_gen_py_inputs
 
         # Before running build assert that the artifact file does not exist.
@@ -217,9 +217,9 @@ def test_build_prepare_artifact_fail(mock_relative_path, mock_install_deps,
         'dlpx.virtualization._internal.plugin_dependency_util.install_deps')
     @mock.patch('os.path.isabs', return_value=False)
     def test_build_generate_artifact_fail(
-        mock_relative_path, mock_install_deps, mock_generate_python,
-        mock_plugin_manifest, mock_gen_artifact, plugin_config_file,
-        artifact_file, codegen_gen_py_inputs):
+            mock_relative_path, mock_install_deps, mock_generate_python,
+            mock_plugin_manifest, mock_gen_artifact, plugin_config_file,
+            artifact_file, codegen_gen_py_inputs):
         gen_py = codegen_gen_py_inputs
 
         # Before running build assert that the artifact file does not exist.
@@ -320,8 +320,7 @@ def test_zip_and_encode_source_files_invalid_dir(src_dir):
 
     @staticmethod
     @mock.patch('compileall.compile_dir')
-    def test_zip_and_encode_source_files_compileall_fail(
-        mock_compile, src_dir):
+    def test_zip_and_encode_source_files_compileall_fail(mock_compile, src_dir):
         mock_compile.return_value = 0
         with pytest.raises(exceptions.UserError) as err_info:
             build.zip_and_encode_source_files(src_dir)