fix!: Add empty paths to referenced paths

Signed-off-by: Caden Marofke <132690522+marofke@users.noreply.github.com>

src/deadline/client/api/_submit_job_bundle.py

@@ -37,6 +37,7 @@
     JobAttachmentsFileSystem,
     AssetRootGroup,
     AssetRootManifest,
+    AssetUploadGroup,
     JobAttachmentS3Settings,
 )
 from ...job_attachments.progress_tracker import SummaryStatistics, ProgressReportMetadata
@@ -58,11 +59,12 @@ def create_job_from_job_bundle(
     max_retries_per_task: Optional[int] = None,
     print_function_callback: Callable[[str], None] = lambda msg: None,
     decide_cancel_submission_callback: Callable[
-        [dict[str, int], int, int], bool
-    ] = lambda paths, files, bytes: False,
+        [AssetUploadGroup], bool
+    ] = lambda upload_group: False,
     hashing_progress_callback: Optional[Callable[[ProgressReportMetadata], bool]] = None,
     upload_progress_callback: Optional[Callable[[ProgressReportMetadata], bool]] = None,
     create_job_result_callback: Optional[Callable[[], bool]] = None,
+    require_paths_exist: bool = False,
 ) -> Union[str, None]:
     """
     Creates a job in the AWS Deadline Cloud farm/queue configured as default for the
@@ -212,10 +214,15 @@ def create_job_from_job_bundle(
     if asset_references and "jobAttachmentSettings" in queue:
         # Extend input_filenames with all the files in the input_directories
         missing_directories: set[str] = set()
-        empty_directories: set[str] = set()
         for directory in asset_references.input_directories:
             if not os.path.isdir(directory):
-                missing_directories.add(directory)
+                if require_paths_exist:
+                    missing_directories.add(directory)
+                else:
+                    logger.info(
+                        f"Input path '{directory}' does not exist. Adding to referenced paths."
+                    )
+                    asset_references.referenced_paths.add(directory)
                 continue
 
             is_dir_empty = True
@@ -226,30 +233,21 @@ def create_job_from_job_bundle(
                 asset_references.input_filenames.update(
                     os.path.normpath(os.path.join(root, file)) for file in files
                 )
+            # Empty directories just become references since there's nothing to upload
             if is_dir_empty:
-                empty_directories.add(directory)
+                logger.info(f"Input directory '{directory}' is empty. Adding to referenced paths.")
+                asset_references.referenced_paths.add(directory)
         asset_references.input_directories.clear()
 
-        misconfigured_directories = missing_directories or empty_directories
-        if misconfigured_directories:
-            all_misconfigured_inputs = ""
+        if missing_directories:
+            all_missing_directories = "\n\t".join(sorted(list(missing_directories)))
             misconfigured_directories_msg = (
                 "Job submission contains misconfigured input directories and cannot be submitted."
-                " All input directories must exist and cannot be empty."
+                " All input directories must exist."
+                f"\nNon-existent directories:\n\t{all_missing_directories}"
             )
 
-            if missing_directories:
-                missing_directory_list = sorted(list(missing_directories))
-                all_missing_directories = "\n\t".join(missing_directory_list)
-                all_misconfigured_inputs += (
-                    f"\nNon-existent directories:\n\t{all_missing_directories}"
-                )
-            if empty_directories:
-                empty_directory_list = sorted(list(empty_directories))
-                all_empty_directories = "\n\t".join(empty_directory_list)
-                all_misconfigured_inputs += f"\nEmpty directories:\n\t{all_empty_directories}"
-
-            raise MisconfiguredInputsError(misconfigured_directories_msg + all_misconfigured_inputs)
+            raise MisconfiguredInputsError(misconfigured_directories_msg)
 
         queue_role_session = api.get_queue_user_boto3_session(
             deadline=deadline,
@@ -267,18 +265,14 @@ def create_job_from_job_bundle(
         )
 
         upload_group = asset_manager.prepare_paths_for_upload(
-            job_bundle_path=job_bundle_dir,
             input_paths=sorted(asset_references.input_filenames),
             output_paths=sorted(asset_references.output_directories),
             referenced_paths=sorted(asset_references.referenced_paths),
             storage_profile=storage_profile,
+            require_paths_exist=require_paths_exist,
         )
         if upload_group.asset_groups:
-            if decide_cancel_submission_callback(
-                upload_group.num_outside_files_by_root,
-                upload_group.total_input_files,
-                upload_group.total_input_bytes,
-            ):
+            if decide_cancel_submission_callback(upload_group):
                 print_function_callback("Job submission canceled.")
                 return None
 

src/deadline/client/cli/_groups/bundle_group.py

@@ -13,13 +13,13 @@
 from botocore.exceptions import ClientError
 
 from deadline.client import api
-from deadline.client.config import set_setting
+from deadline.client.config import config_file, get_setting, set_setting
 from deadline.job_attachments.exceptions import (
     AssetSyncError,
     AssetSyncCancelledError,
     MisconfiguredInputsError,
 )
-from deadline.job_attachments.models import JobAttachmentsFileSystem
+from deadline.job_attachments.models import AssetUploadGroup, JobAttachmentsFileSystem
 from deadline.job_attachments.progress_tracker import ProgressReportMetadata
 from deadline.job_attachments._utils import _human_readable_file_size
 
@@ -91,6 +91,11 @@ def validate_parameters(ctx, param, value):
     is_flag=True,
     help="Skip any confirmation prompts",
 )
+@click.option(
+    "--require-paths-exist",
+    is_flag=True,
+    help="Require all input paths to exist",
+)
 @click.argument("job_bundle_dir")
 @_handle_error
 def bundle_submit(
@@ -102,6 +107,7 @@ def bundle_submit(
     priority,
     max_failed_tasks_count,
     max_retries_per_task,
+    require_paths_exist,
     **args,
 ):
     """
@@ -116,36 +122,54 @@ def bundle_submit(
     def _check_create_job_wait_canceled() -> bool:
         return sigint_handler.continue_operation
 
-    def _decide_cancel_submission(
-        deviated_file_count_by_root: dict[str, int],
-        num_files: int,
-        upload_size: int,
-    ):
+    def _decide_cancel_submission(upload_group: AssetUploadGroup) -> bool:
+        """
+        Callback to decide if submission should be cancelled or not. Return 'True' to cancel.
+        Prints a warning that requires confirmation if paths are found outside of configured storage profile locations.
+        """
+        warning_message = ""
+        for group in upload_group.asset_groups:
+            if not group.file_system_location_name:
+                warning_message += f"\n\nUnder the directory '{group.root_path}':"
+                warning_message += (
+                    f"\n\t{len(group.inputs)} input file{'' if len(group.inputs) == 1 else 's'}"
+                    if len(group.inputs) > 0
+                    else ""
+                )
+                warning_message += (
+                    f"\n\t{len(group.outputs)} output director{'y' if len(group.outputs) == 1 else 'ies'}"
+                    if len(group.outputs) > 0
+                    else ""
+                )
+                warning_message += (
+                    f"\n\t{len(group.references)} referenced file{'' if len(group.references) == 1 else 's'} and/or director{'y' if len(group.outputs) == 1 else 'ies'}"
+                    if len(group.references) > 0
+                    else ""
+                )
+
+        # Exit early if there are no warnings and we've either set auto accept or there's no files to confirm
+        if not warning_message and (
+            yes
+            or config_file.str2bool(get_setting("settings.auto_accept", config=config))
+            or upload_group.total_input_files == 0
+        ):
+            return False
+
         message_text = (
-            f"Job submission contains {num_files} files totaling {_human_readable_file_size(upload_size)}. "
-            " All files will be uploaded to S3 if they are not already present in the job attachments bucket."
+            f"Job submission contains {upload_group.total_input_files} input files totaling {_human_readable_file_size(upload_group.total_input_bytes)}. "
+            " All input files will be uploaded to S3 if they are not already present in the job attachments bucket."
         )
-        if deviated_file_count_by_root:
-            root_by_count_message = "\n\n".join(
-                [
-                    f"{file_count} files from: '{directory}'"
-                    for directory, file_count in deviated_file_count_by_root.items()
-                ]
-            )
+        if warning_message:
             message_text += (
-                f"\n\nFiles were found outside of the configured storage profile location(s). "
-                " Please confirm that you intend to upload files from the following directories:\n\n"
-                f"{root_by_count_message}\n\n"
-                "To permanently remove this warning you must only upload files located within a storage profile location."
+                f"\n\nFiles were specified outside of the configured storage profile location(s). "
+                " Please confirm that you intend to submit a job that uses files from the following directories:"
+                f"{warning_message}\n\n"
+                "To permanently remove this warning you must only use files located within a storage profile location."
             )
         message_text += "\n\nDo you wish to proceed?"
-        return (
-            not yes
-            and num_files > 0
-            and not click.confirm(
-                message_text,
-                default=not deviated_file_count_by_root,
-            )
+        return not click.confirm(
+            message_text,
+            default=not warning_message,
         )
 
     try:
@@ -163,6 +187,7 @@ def _decide_cancel_submission(
             create_job_result_callback=_check_create_job_wait_canceled,
             print_function_callback=click.echo,
             decide_cancel_submission_callback=_decide_cancel_submission,
+            require_paths_exist=require_paths_exist,
         )
 
         # Check Whether the CLI options are modifying any of the default settings that affect