makde updates to exceptions and tests

python/example_code/neptune/tests/database_tests/neptune_data_stubber.py

@@ -1,3 +1,6 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+
 import boto3
 import io
 from botocore.stub import Stubber

python/example_code/neptune/tests/database_tests/test_execute_gremlin_query.py

@@ -1,5 +1,8 @@
+
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+
 import types
-import pytest
 from botocore.exceptions import ClientError, BotoCoreError
 from database.neptune_execute_gremlin_query import execute_gremlin_query
 from neptune_data_stubber import NeptuneDateStubber
@@ -10,7 +13,6 @@ def test_execute_gremlin_query(capfd):
     stubber.activate()
 
     try:
-        # --- Success case with valid output ---
         stubber.add_execute_gremlin_query_stub(
             gremlin_query="g.V().has('code', 'ANC')",
             response_dict={"result": {"metrics": {"dur": 500, "steps": 3}}}
@@ -21,7 +23,6 @@ def test_execute_gremlin_query(capfd):
         assert "Response is:" in out
         assert '"dur": 500' in out or "'dur': 500" in out
 
-        # --- Success case with None result ---
         stubber.stubber.assert_no_pending_responses()
         stubber.add_execute_gremlin_query_stub(
             gremlin_query="g.V().has('code', 'ANC')",
@@ -32,7 +33,6 @@ def test_execute_gremlin_query(capfd):
         assert "Response is:" in out
         assert "None" in out
 
-        # --- ClientError case ---
         stubber.stubber.assert_no_pending_responses()
         stubber.stubber.add_client_error(
             method='execute_gremlin_query',
@@ -44,7 +44,6 @@ def test_execute_gremlin_query(capfd):
         out, _ = capfd.readouterr()
         assert "Neptune error: Invalid query" in out
 
-        # --- BotoCoreError case (Fix 1) ---
         stubber.stubber.assert_no_pending_responses()
 
         def raise_boto_core_error(*args, **kwargs):

python/example_code/neptune/tests/database_tests/test_opencypher_explain_query.py

@@ -1,3 +1,6 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+
 import types
 from botocore.exceptions import ClientError, BotoCoreError
 from botocore.response import StreamingBody

python/example_code/neptune/tests/neptune_stubber.py

@@ -41,16 +41,13 @@ def stub_create_db_subnet_group(self, group_name, subnet_ids, group_arn=None, er
             )
 
     def stub_create_db_cluster(self, cluster_id=None, error_code=None,
-                               backup_retention_period=None, deletion_protection=None, engine=None):
+                               backup_retention_period=1, deletion_protection=False, engine='neptune'):
         expected_params = {
             "DBClusterIdentifier": cluster_id,
+            "BackupRetentionPeriod": backup_retention_period,
+            "DeletionProtection": deletion_protection,
+            "Engine": engine
         }
-        if backup_retention_period is not None:
-            expected_params["BackupRetentionPeriod"] = backup_retention_period
-        if deletion_protection is not None:
-            expected_params["DeletionProtection"] = deletion_protection
-        if engine is not None:
-            expected_params["Engine"] = engine
 
         if error_code:
             self.stubber.add_client_error(

python/example_code/neptune/tests/test_create_db_cluster.py

@@ -7,21 +7,20 @@
 from neptune_stubber import Neptune
 from neptune_scenario import create_db_cluster  # Your actual function
 
+
 def test_create_db_cluster():
-    boto_client = boto3.client("neptune")
+    # Create Boto3 Neptune client and attach Stubber wrapper
+    boto_client = boto3.client("neptune", region_name="us-east-1")
     stubber = Neptune(boto_client)
 
     # --- Success case ---
     stubber.stub_create_db_cluster(
-        cluster_id="test-cluster",
-        engine="neptune",
-        deletion_protection=False,
-        backup_retention_period=1
+        cluster_id="test-cluster"
+        # engine, deletion_protection, backup_retention_period defaulted
     )
     cluster_id = create_db_cluster(stubber.client, "test-cluster")
     assert cluster_id == "test-cluster"
 
-    # --- Missing cluster ID raises RuntimeError ---
     stubber.stubber.add_response(
         "create_db_cluster",
         {"DBCluster": {}},
@@ -35,22 +34,19 @@ def test_create_db_cluster():
     with pytest.raises(RuntimeError, match="Cluster created but no ID returned"):
         create_db_cluster(stubber.client, "missing-id-cluster")
 
-    # --- ClientError is wrapped and re-raised ---
+    # --- ClientError is wrapped in RuntimeError ---
     stubber.stub_create_db_cluster(
         cluster_id="denied-cluster",
-        error_code="AccessDenied",
-        engine="neptune",
-        deletion_protection=False,
-        backup_retention_period=1
+        error_code="AccessDenied"
     )
-    with pytest.raises(ClientError) as exc_info:
+    with pytest.raises(RuntimeError) as exc_info:
         create_db_cluster(stubber.client, "denied-cluster")
-    assert "Failed to create DB cluster 'denied-cluster'" in str(exc_info.value)
+    assert "AWS error [AccessDenied]" in str(exc_info.value)
 
     # --- Unexpected exception raises RuntimeError ---
     def raise_generic_exception(**kwargs):
         raise Exception("Unexpected failure")
 
     stubber.client.create_db_cluster = raise_generic_exception
-    with pytest.raises(RuntimeError, match="Unexpected error creating DB cluster"):
-        create_db_cluster(stubber.client, "fail-cluster")
+    with pytest.raises(RuntimeError, match="Unexpected error creating DB cluster 'fail-cluster'"):
+        create_db_cluster(stubber.client, "fail-cluster")

python/example_code/neptune/tests/test_create_db_instance.py

@@ -7,6 +7,12 @@
 from neptune_stubber import Neptune
 from neptune_scenario import create_db_instance
 
+class DummyWaiter:
+    def __init__(self, name):
+        self.name = name
+    def wait(self, **kwargs):
+        return None  # Simulate successful wait
+
 def test_create_db_instance():
     boto_client = boto3.client("neptune")
     stubber = Neptune(boto_client)
@@ -23,7 +29,7 @@ def test_create_db_instance():
     assert result == instance_id
 
     # --- Missing ID raises RuntimeError ---
-    stubber.stubber.add_response(  # can't use your stub_create_db_instance here because it always returns an ID
+    stubber.stubber.add_response(
         "create_db_instance",
         {"DBInstance": {}},
         expected_params={
@@ -36,21 +42,15 @@ def test_create_db_instance():
     with pytest.raises(RuntimeError, match="no ID returned"):
         create_db_instance(stubber.client, "no-id-instance", cluster_id)
 
-    # --- ClientError is re-raised with wrapped message ---
+    # --- ClientError is re-raised ---
     stubber.stub_create_db_instance("fail-instance", cluster_id, error_code="AccessDenied")
     with pytest.raises(ClientError) as e:
         create_db_instance(stubber.client, "fail-instance", cluster_id)
-    assert "Failed to create DB instance 'fail-instance'" in str(e.value)
+    assert "AccessDenied error" in str(e.value)
 
     # --- Unexpected exception case ---
     def broken_call(**kwargs):
         raise Exception("DB is on fire")
     stubber.client.create_db_instance = broken_call
     with pytest.raises(RuntimeError, match="Unexpected error creating DB instance 'boom-instance'"):
         create_db_instance(stubber.client, "boom-instance", cluster_id)
-
-class DummyWaiter:
-    def __init__(self, name):
-        self.name = name
-    def wait(self, **kwargs):
-        return None  # Simulate successful wait

python/example_code/neptune/tests/test_create_subnet_group.py

@@ -4,9 +4,7 @@
 import pytest
 import boto3
 from unittest.mock import patch
-from botocore.exceptions import ClientError
 from neptune_stubber import Neptune
-from neptune_scenario import create_subnet_group  # Your real function to test
 
 @patch("neptune_scenario.get_subnet_ids")
 @patch("neptune_scenario.get_default_vpc_id")

python/example_code/neptune/tests/test_delete_db_cluster.py

@@ -4,7 +4,6 @@
 import pytest
 import boto3
 from botocore.exceptions import ClientError
-
 from neptune_scenario import delete_db_cluster  # Your actual module
 from neptune_stubber import Neptune  # Update path if needed
 

python/example_code/neptune/tests/test_delete_db_subnet_group.py

@@ -9,15 +9,12 @@
 
 
 def test_delete_db_subnet_group():
-    # Create a real boto3 client and wrap it with the custom stubber
     boto_client = boto3.client("neptune", region_name="us-east-1")
     stubber = Neptune(boto_client)
 
-    # --- Success case ---
     stubber.stub_delete_db_subnet_group("my-subnet-group")
     delete_db_subnet_group(stubber.client, "my-subnet-group")
 
-    # --- ClientError case ---
     stubber.stub_delete_db_subnet_group(
         "unauthorized-subnet",
         error_code="AccessDenied"

python/example_code/neptune/tests/test_hello.py

@@ -45,10 +45,7 @@ def test_describe_db_clusters_unit(mock_neptune_client, capsys):
     ]
 
     try:
-        # Call the function with the mocked client
         describe_db_clusters(mock_neptune_client)
-
-        # Capture stdout
         captured = capsys.readouterr()
 
         # Check that expected outputs from both pages were printed
@@ -57,10 +54,7 @@ def test_describe_db_clusters_unit(mock_neptune_client, capsys):
         assert "my-second-cluster" in captured.out
         assert "modifying" in captured.out
 
-        # Ensure get_paginator was called with correct operation
         mock_neptune_client.get_paginator.assert_called_once_with("describe_db_clusters")
-
-        # Ensure paginate method was called
         mock_paginator.paginate.assert_called_once()
 
     except ClientError as e:

python/example_code/neptune/tests/test_start_db_cluster.py

@@ -21,7 +21,6 @@ def test_start_db_cluster_success(mock_sleep):
         {"DBClusterIdentifier": cluster_id}
     )
 
-    # Stub 9 "starting" statuses and 1 "available"
     statuses = ["starting"] * 9 + ["available"]
     for status in statuses:
         neptune.stubber.add_response(
@@ -32,8 +31,6 @@ def test_start_db_cluster_success(mock_sleep):
             {"DBClusterIdentifier": cluster_id}
         )
 
-    # Run the service method
     start_db_cluster(client, cluster_id)
-
     neptune.stubber.deactivate()