Testing done

python/example_code/s3-directory-buckets/requirements.txt

@@ -1,3 +1,4 @@
-boto3>=1.35.49
+boto3>=1.35.77
+botocore>=1.35.77
 pytest>=7.2.1
 requests>=2.28.2

python/example_code/s3-directory-buckets/s3_express_getting_started.py

@@ -243,7 +243,7 @@ def create_session_and_add_objects(self) -> None:
             "Next, we'll copy the object into the Directory bucket using the regular client."
         )
         print(
-            "This works fine, because Copy operations are not restricted for Directory buckets."
+            "This works fine, because copy operations are not restricted for Directory buckets."
         )
         press_enter_to_continue()
         bucket_object = "basic-text-object"
@@ -775,6 +775,8 @@ def tear_done_vpc(self) -> None:
                     self.vpc_id,
                     client_error.response["Error"]["Message"],
                 )
+
+
 # snippet-end:[python.example_code.s3.s3_express_basics]
 
 if __name__ == "__main__":
@@ -811,5 +813,3 @@ def tear_done_vpc(self) -> None:
         logging.exception("Type error in demo!")
         if s3_express_scenario is not None:
             s3_express_scenario.cleanup()
-
-

python/example_code/s3-directory-buckets/tests/test_s3_express_getting_started_cleanup.py

@@ -5,23 +5,122 @@
 Tests for s3_express_getting_started.py.
 """
 
+
 import pytest
 
-import boto3
 from botocore import waiter
-import os
-import sys
 
-script_dir = os.path.dirname(os.path.abspath(__file__))
 
-# Append directory for s3_express_getting_started.py
-sys.path.append(os.path.join(script_dir, ".."))
-import s3_express_getting_started
+class MockManager:
+    def __init__(self, stub_runner, scenario_data):
+        self.scenario_data = scenario_data
+        self.my_uuid = "0000"
+        self.bucket_name_prefix = "amzn-s3-demo-bucket"
+        self.directory_bucket_name = (
+            f"{self.bucket_name_prefix}-{self.my_uuid}--use1-az2--x-s3"
+        )
+        self.regular_bucket_name = f"{self.bucket_name_prefix}-regular-{self.my_uuid}"
+
+        self.object_name = "basic-text-object"
+        self.other_object = f"other/{self.object_name}"
+        self.alt_object = f"alt/{self.object_name}"
+        self.other_alt_object = f"other/alt/{self.object_name}"
+
+        self.object_keys = [
+            self.object_name,
+            self.other_object,
+            self.alt_object,
+            self.other_alt_object,
+        ]
+
+        self.vpc_id = "XXXXXXXXXXXXXXXXXXXXX"
+        self.vpc_endpoint_id = f"vpce-{self.vpc_id}"
+
+        self.stack_name = f"cfn-stack-s3-express-basics--{self.my_uuid}"
+        self.stack = scenario_data.cloud_formation_resource.Stack(self.stack_name)
+        self.stub_runner = stub_runner
+
+    def setup_stubs(
+        self,
+        error,
+        stop_on,
+        ec2_stubber,
+        cloud_formation_stubber,
+        s3_stubber,
+        monkeypatch,
+    ):
+        with self.stub_runner(error, stop_on) as runner:
+            runner.add(
+                s3_stubber.stub_list_objects_v2,
+                self.directory_bucket_name,
+                self.object_keys,
+            )
+            runner.add(
+                s3_stubber.stub_delete_objects,
+                self.directory_bucket_name,
+                self.object_keys,
+            )
+            runner.add(s3_stubber.stub_delete_bucket, self.directory_bucket_name)
+            runner.add(
+                s3_stubber.stub_list_objects_v2,
+                self.regular_bucket_name,
+                self.object_keys,
+            )
+            runner.add(
+                s3_stubber.stub_delete_objects,
+                self.regular_bucket_name,
+                self.object_keys,
+            )
+            runner.add(s3_stubber.stub_delete_bucket, self.regular_bucket_name)
+            runner.add(cloud_formation_stubber.stub_delete_stack, self.stack_name)
+            runner.add(ec2_stubber.stub_delete_vpc_endpoints, [self.vpc_endpoint_id])
+            runner.add(ec2_stubber.stub_delete_vpc, self.vpc_id)
+
+        # Mock the waiters.
+        monkeypatch.setattr(waiter.Waiter, "wait", lambda arg, **kwargs: None)
+
+        self.scenario_data.scenario.s3_express_client = self.scenario_data.s3_client
+        self.scenario_data.scenario.s3_regular_client = self.scenario_data.s3_client
+        self.scenario_data.scenario.directory_bucket_name = self.directory_bucket_name
+        # The cleanup function keeps executing after an exception is raised.
+        # This allows the app to clean up all possible resources.
+        # Because of this, exception testing fails because actions are called after the exception is raised.
+        # To avoid false negatives when testing exceptions, certain fields are set only when they should be tested.
+
+        if stop_on is None or stop_on > 2:
+            self.scenario_data.scenario.regular_bucket_name = self.regular_bucket_name
+
+        if stop_on is None or stop_on > 5:
+            self.scenario_data.scenario.stack = self.stack
+
+        if stop_on is None or stop_on > 6:
+            self.scenario_data.scenario.vpc_endpoint_id = self.vpc_endpoint_id
+
+        if stop_on is None or stop_on > 7:
+            self.scenario_data.scenario.vpc_id = self.vpc_id
+
+
+@pytest.fixture
+def mock_mgr(stub_runner, scenario_data):
+    return MockManager(stub_runner, scenario_data)
 
 
 @pytest.mark.integ
+def test_scenario_cleanup(mock_mgr, capsys, monkeypatch):
+    mock_mgr.setup_stubs(
+        None,
+        None,
+        mock_mgr.scenario_data.ec2_stubber,
+        mock_mgr.scenario_data.cloud_formation_stubber,
+        mock_mgr.scenario_data.s3_stubber,
+        monkeypatch,
+    )
+
+    mock_mgr.scenario_data.scenario.cleanup()
+
+
 @pytest.mark.parametrize(
-    "error_code, stop_on_index",
+    "error, stop_on_index",
     [
         ("TESTERROR-stub_list_objects_directory", 0),
         ("TESTERROR-stub_delete_objects_directory", 1),
@@ -32,81 +131,17 @@
         ("TESTERROR-stub_delete_stack", 6),
         ("TESTERROR-stub_delete_vpc_endpoints", 7),
         ("TESTERROR-stub_delete_vpc", 8),
-        (None, 8),
     ],
 )
-def test_s3_express_scenario_cleanup(
-    make_stubber, stub_runner, error_code, stop_on_index, monkeypatch
-):
-    region = "us-east-1"
-    cloud_formation_resource = boto3.resource("cloudformation")
-    cloud_formation_stubber = make_stubber(cloud_formation_resource.meta.client)
-
-    ec2_client = boto3.client("ec2", region_name=region)
-    ec2_stubber = make_stubber(ec2_client)
-
-    iam_client = boto3.client("iam")
-
-    s3_client = boto3.client("s3")
-    s3_stubber = make_stubber(s3_client)
-
-    my_uuid = "0000"
-
-    availability_zone_ids = ["use1-az2"]
-
-    bucket_name_prefix = "amzn-s3-demo-bucket"
-    directory_bucket_name = (
-        f"{bucket_name_prefix}-{my_uuid}--{availability_zone_ids[0]}--x-s3"
-    )
-    regular_bucket_name = f"{bucket_name_prefix}-regular-{my_uuid}"
-    object_name = "basic-text-object"
-    other_object = f"other/{object_name}"
-    alt_object = f"alt/{object_name}"
-    other_alt_object = f"other/alt/{object_name}"
-
-    object_keys = [object_name, other_object, alt_object, other_alt_object]
-
-    my_uuid = "0000"
-
-    stack_name = f"cfn-stack-s3-express-basics--{my_uuid}"
-    stack = cloud_formation_resource.Stack(stack_name)
-    vpc_id = "XXXXXXXXXXXXXXXXXXXXX"
-    vpc_endpoint_id = f"vpce-{vpc_id}"
-
-    with stub_runner(error_code, stop_on_index) as runner:
-        runner.add(s3_stubber.stub_list_objects_v2, directory_bucket_name, object_keys)
-        runner.add(s3_stubber.stub_delete_objects, directory_bucket_name, object_keys)
-        runner.add(s3_stubber.stub_delete_bucket, directory_bucket_name)
-        runner.add(s3_stubber.stub_list_objects_v2, regular_bucket_name, object_keys)
-        runner.add(s3_stubber.stub_delete_objects, regular_bucket_name, object_keys)
-        runner.add(s3_stubber.stub_delete_bucket, regular_bucket_name)
-        runner.add(cloud_formation_stubber.stub_delete_stack, stack_name)
-        runner.add(ec2_stubber.stub_delete_vpc_endpoints, [vpc_endpoint_id])
-        runner.add(ec2_stubber.stub_delete_vpc, vpc_id)
-
-    scenario = s3_express_getting_started.S3ExpressScenario(
-        cloud_formation_resource, ec2_client, iam_client
+@pytest.mark.integ
+def test_scenario_cleanup_error(mock_mgr, caplog, error, stop_on_index, monkeypatch):
+    mock_mgr.setup_stubs(
+        error,
+        stop_on_index,
+        mock_mgr.scenario_data.ec2_stubber,
+        mock_mgr.scenario_data.cloud_formation_stubber,
+        mock_mgr.scenario_data.s3_stubber,
+        monkeypatch,
     )
 
-    def mock_wait(self, **kwargs):
-        return
-
-    # Mock the waiters.
-    monkeypatch.setattr(waiter.Waiter, "wait", mock_wait)
-
-    scenario.s3_express_client = s3_client
-    scenario.s3_regular_client = s3_client
-    scenario.directory_bucket_name = directory_bucket_name
-    if stop_on_index > 2:
-        scenario.regular_bucket_name = regular_bucket_name
-
-    if stop_on_index > 5:
-        scenario.stack = stack
-
-    if stop_on_index > 6:
-        scenario.vpc_endpoint_id = vpc_endpoint_id
-
-    if stop_on_index > 7:
-        scenario.vpc_id = vpc_id
-
-    scenario.cleanup()
+    mock_mgr.scenario_data.scenario.cleanup()

python/example_code/s3-directory-buckets/tests/test_s3_express_getting_started_create_session_and_add_objects.py

@@ -7,16 +7,8 @@
 
 import pytest
 
-import boto3
 from botocore.exceptions import ClientError
-import os
-import sys
 
-script_dir = os.path.dirname(os.path.abspath(__file__))
-
-# Append directory for s3_express_getting_started.py
-sys.path.append(os.path.join(script_dir, ".."))
-import s3_express_getting_started
 
 class MockManager:
     def __init__(self, stub_runner, scenario_data, input_mocker):
@@ -26,14 +18,11 @@ def __init__(self, stub_runner, scenario_data, input_mocker):
         self.availability_zone_ids = ["use1-az2"]
 
         self.bucket_name_prefix = "amzn-s3-demo-bucket"
-        self.directory_bucket_name = (
-            f"{self.bucket_name_prefix}-{self.my_uuid}--{self.availability_zone_ids[0]}--x-s3"
-        )
+        self.directory_bucket_name = f"{self.bucket_name_prefix}-{self.my_uuid}--{self.availability_zone_ids[0]}--x-s3"
         self.regular_bucket_name = f"{self.bucket_name_prefix}-regular-{self.my_uuid}"
 
         self.object_name = "basic-text-object"
 
-
         answers = []
         input_mocker.mock_answers(answers)
         self.stub_runner = stub_runner
@@ -67,7 +56,7 @@ def mock_mgr(stub_runner, scenario_data, input_mocker):
 
 
 @pytest.mark.integ
-def test_scenario_create_one_time_schedule(mock_mgr, capsys, monkeypatch):
+def test_scenario_create_session_and_add_objects(mock_mgr, capsys, monkeypatch):
     mock_mgr.setup_stubs(None, None, mock_mgr.scenario_data.s3_stubber)
 
     mock_mgr.scenario_data.scenario.create_session_and_add_objects()
@@ -79,12 +68,13 @@ def test_scenario_create_one_time_schedule(mock_mgr, capsys, monkeypatch):
         ("TESTERROR-stub_put_object", 0),
         ("TESTERROR-stub_create_session", 1),
         ("TESTERROR-stub_copy_object", 2),
-    ]
+    ],
 )
-
 @pytest.mark.integ
-def test_scenario_create_one_time_schedule_error(mock_mgr, caplog, error, stop_on_index, monkeypatch):
+def test_scenario_create_session_and_add_objects_error(
+    mock_mgr, caplog, error, stop_on_index, monkeypatch
+):
     mock_mgr.setup_stubs(error, stop_on_index, mock_mgr.scenario_data.s3_stubber)
 
-    with pytest.raises(ClientError) as exc_info:
+    with pytest.raises(ClientError):
         mock_mgr.scenario_data.scenario.create_session_and_add_objects()