def upload()

in src/sfctl/custom_app.py [0:0]

• 69 lines of code
• 26 McCabe index (conditional complexity)

def upload(path, imagestore_string='fabric:ImageStore', show_progress=False, timeout=300,  # pylint: disable=too-many-locals,missing-docstring,too-many-arguments,too-many-branches,too-many-statements
           compress=False, keep_compressed=False, compressed_location=None):

    from sfctl.config import (client_endpoint, no_verify_setting, ca_cert_info,
                              cert_info)
    import requests

    path = _normalize_path(path)
    if compressed_location is not None:
        compressed_location = _normalize_path(compressed_location)

    abspath = validate_app_path(path)
    basename = os.path.basename(abspath)

    endpoint = client_endpoint()
    cert = cert_info()
    ca_cert = True
    if no_verify_setting():
        ca_cert = False
    elif ca_cert_info():
        ca_cert = ca_cert_info()

    if all([no_verify_setting(), ca_cert_info()]):
        raise CLIError('Cannot specify both CA cert info and no verify')

    if not compress and (keep_compressed or compressed_location is not None):
        raise CLIError('--keep-compressed and --compressed-location options are only applicable '
                       'if the --compress option is set')

    compressed_pkg_location = None
    created_dir_path = None

    if compress:

        parent_folder = os.path.dirname(path)
        file_or_folder_name = os.path.basename(path)

        compressed_pkg_location = os.path.join(parent_folder, 'sfctl_compressed_temp')

        if compressed_location is not None:
            compressed_pkg_location = compressed_location

        # Check if a zip file has already been created
        created_dir_path = os.path.join(compressed_pkg_location, file_or_folder_name)

        if os.path.exists(created_dir_path):
            if get_user_confirmation(str.format('Deleting previously generated compressed files at '
                                                '{0}. If this folder has anything else, those will be '
                                                'deleted as well. Allow? ["y", "n"]: ', created_dir_path)):
                shutil.rmtree(created_dir_path)
            else:
                # We can consider adding an option to number the packages in the future.
                print('Stopping upload operation. Cannot compress to the following location '
                      'because the path already exists: ' + created_dir_path)
                return

        # Let users know where to find the compressed app package before starting the
        # copy / compression, in case the process crashes in the middle, so users
        # will know where to clean up items from, or where to upload already compressed
        # app packages from
        if show_progress:
            print('Starting package compression into location: ' + compressed_pkg_location)
            print()  # New line for formatting purposes
        compress_package(path, compressed_pkg_location)

        # Change the path to the path with the compressed package
        compressed_path = os.path.join(compressed_pkg_location, file_or_folder_name)

        # re-do validation and reset the variables
        abspath = validate_app_path(compressed_path)
        basename = os.path.basename(abspath)

    # Note: pressing ctrl + C during upload does not end the current upload in progress, but only
    # stops the next one from occurring. This will be fixed in the future.

    # Upload to either to a folder, or native image store only
    if 'file:' in imagestore_string:
        dest_path = path_from_imagestore_string(imagestore_string)

        process = Process(target=upload_to_fileshare,
                          args=(abspath, os.path.join(dest_path, basename), show_progress))

        process.start()
        process.join(timeout)  # If timeout is None then there is no timeout.

        if process.is_alive():
            process.terminate()  # This will leave any children of process orphaned.
            raise SFCTLInternalException('Upload has timed out. Consider passing a longer '
                                         'timeout duration.')

    elif imagestore_string == 'fabric:ImageStore':

        with requests.Session() as sesh:
            sesh.verify = ca_cert
            sesh.cert = cert

            # There is no need for a new process here since
            upload_to_native_imagestore(sesh, endpoint, abspath, basename, show_progress, timeout)

    else:
        raise CLIError('Unsupported image store connection string. Value should be either '
                       '"fabric:ImageStore", or start with "file:"')

    # If code has reached here, it means that upload was successful
    # To reach here, user must have agreed to clear this folder or exist the API
    # So we can safely delete the contents
    # User is expected to not create a folder by the same name during the upload duration
    # If needed, we can consider adding our content under a GUID in the future
    if compress and not keep_compressed:
        # Remove the generated files
        if show_progress:
            print('Removing generated folder ' + created_dir_path)
        shutil.rmtree(created_dir_path)