Update with status checking for cancellation, and simplified integration test.

Author: rlhagerm

python/example_code/s3/scenarios/batch/s3_batch_wrapper.py

@@ -316,12 +316,28 @@ def cancel_job(self, job_id: str, account_id: str) -> None:
             account_id (str): AWS account ID
         """
         try:
-            self.s3control_client.update_job_status(
+            response = self.s3control_client.describe_job(
                 AccountId=account_id,
-                JobId=job_id,
-                RequestedJobStatus='Cancelled'
+                JobId=job_id
             )
-            print(f"Job {job_id} was successfully canceled.")
+            current_status = response['Job']['Status']
+            print(f"Current job status: {current_status}")
+
+            if current_status in ['Ready', 'Suspended', 'Active']:
+                self.s3control_client.update_job_status(
+                    AccountId=account_id,
+                    JobId=job_id,
+                    RequestedJobStatus='Cancelled'
+                )
+                print(f"Job {job_id} was successfully canceled.")
+            elif current_status in ['Completing', 'Complete']:
+                print(f"Job is in '{current_status}' state - cannot be cancelled")
+                if current_status == 'Completing':
+                    print("Job is finishing up and will complete soon.")
+                elif current_status == 'Complete':
+                    print("Job has already completed successfully.")
+            else:
+                print(f"Job is in '{current_status}' state - cancel not allowed")
         except ClientError as e:
             print(f"Error canceling job: {e}")
             raise

python/example_code/s3/scenarios/batch/test/conftest.py

@@ -1,36 +1,14 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: Apache-2.0
 
-"""Shared test fixtures for S3 batch operations tests."""
+"""
+Contains common test fixtures used to run unit tests.
+"""
 
-import boto3
-import pytest
-from moto import mock_s3, mock_s3control, mock_sts
+import sys
 
-from test_s3_batch_stubber import S3BatchStubber
-from s3_batch_wrapper import S3BatchWrapper
-from cloudformation_helper import CloudFormationHelper
+# This is needed so Python can find test_tools on the path.
+sys.path.append("../../../..")
+from test_tools.fixtures.common import *
 
 
-class ScenarioData:
-    """Holds data for scenario tests."""
-    
-    def __init__(self, wrapper, cfn_helper, stubber):
-        self.wrapper = wrapper
-        self.cfn_helper = cfn_helper
-        self.stubber = stubber
-
-
-@pytest.fixture
-def scenario_data(make_stubber):
-    """Create scenario data with stubbed clients."""
-    s3_client = boto3.client("s3", region_name="us-east-1")
-    s3control_client = boto3.client("s3control", region_name="us-east-1")
-    sts_client = boto3.client("sts", region_name="us-east-1")
-    cfn_client = boto3.client("cloudformation", region_name="us-east-1")
-    
-    wrapper = S3BatchWrapper(s3_client, s3control_client, sts_client)
-    cfn_helper = CloudFormationHelper(cfn_client)
-    stubber = make_stubber(S3BatchStubber, s3_client, s3control_client, sts_client)
-    
-    return ScenarioData(wrapper, cfn_helper, stubber)

python/example_code/s3/scenarios/batch/test/test_batch_scenario.py

@@ -0,0 +1,31 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+import pytest
+
+import boto3
+
+from s3_batch_wrapper import S3BatchWrapper
+from cloudformation_helper import CloudFormationHelper
+from s3_batch_scenario import S3BatchScenario
+
+
+@pytest.mark.integ
+def test_run_batch_wrapper_scenario_integ(input_mocker, capsys):
+    s3_client = boto3.client('s3')
+    s3control_client = boto3.client('s3control')
+    sts_client = boto3.client('sts')
+    cfn_client = boto3.client('cloudformation')
+
+    batch_wrapper = S3BatchWrapper(s3_client, s3control_client, sts_client)
+    cf_helper = CloudFormationHelper(cfn_client)
+    scenario = S3BatchScenario(batch_wrapper, cf_helper)
+
+    input_mocker.mock_answers(
+        ["y", "y", "y", "n", "y", "y", "y", "y", "y", "y", "y"] # yes to proceed, no to cancel, yes to cleanup
+    )
+
+    scenario.run_scenario()
+
+    capt = capsys.readouterr()
+    assert "has successfully completed" in capt.out