moved stubber to propet location

Author: scmacdon

python/example_code/neptune/tests/analytics_tests/test_create_graph.py

@@ -1,7 +1,9 @@
-import pytest
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+
 from botocore.exceptions import ClientError, BotoCoreError
-from analytics.create_neptune_graph_example import execute_create_graph
-from neptune_graph_stubber import NeptuneGraphStubber
+from example_code.neptune.analytics.create_neptune_graph_example import execute_create_graph
+from test_tools.neptune_graph_stubber import NeptuneGraphStubber
 
 class MockBotoCoreError(BotoCoreError):
     def __init__(self, message="BotoCore error occurred"):
@@ -12,7 +14,6 @@ def __str__(self):
         return self.message
 
 def test_execute_create_graph(capfd):
-    # --- Success case ---
     stubber = NeptuneGraphStubber()
     client = stubber.get_client()
     stubber.activate()
@@ -26,9 +27,8 @@ def test_execute_create_graph(capfd):
     assert "Graph ARN: arn:aws:neptune-graph:us-east-1:123456789012:graph/test-graph" in out
     assert "Graph Endpoint: https://test-graph.cluster-neptune.amazonaws.com" in out
 
-    stubber.deactivate()  # deactivate the stubber before mocking
+    stubber.deactivate()
 
-    # --- ClientError case ---
     def raise_client_error(**kwargs):
         raise ClientError(
             {"Error": {"Message": "Client error occurred"}}, "CreateGraph"
@@ -38,15 +38,13 @@ def raise_client_error(**kwargs):
     out, _ = capfd.readouterr()
     assert "Failed to create graph: Client error occurred" in out
 
-    # --- BotoCoreError case ---
     def raise_boto_core_error(**kwargs):
         raise MockBotoCoreError()
     client.create_graph = raise_boto_core_error
     execute_create_graph(client, "test-graph")
     out, _ = capfd.readouterr()
     assert "Failed to create graph: BotoCore error occurred" in out
 
-    # --- Generic Exception case ---
     def raise_generic_error(**kwargs):
         raise Exception("Generic failure")
     client.create_graph = raise_generic_error

python/example_code/neptune/tests/analytics_tests/test_execute_gremlin_profile_query.py

@@ -1,17 +1,17 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+
 import io
-import pytest
 from botocore.response import StreamingBody
-from botocore.exceptions import ClientError
-from analytics.neptune_analytics_query_example import run_open_cypher_query
-from neptune_graph_stubber import NeptuneGraphStubber
+from example_code.neptune.analytics.neptune_analytics_query_example import run_open_cypher_query
+from test_tools.neptune_graph_stubber import NeptuneGraphStubber
 
 GRAPH_ID = "test-graph-id"
 
 def test_execute_gremlin_profile_query(capfd):
     stubber = NeptuneGraphStubber()
     client = stubber.get_client()
 
-    # --- Success case with payload ---
     stubber.activate()
     payload_bytes = b'{"results": "some data"}'
     response_body = StreamingBody(io.BytesIO(payload_bytes), len(payload_bytes))
@@ -30,7 +30,6 @@ def test_execute_gremlin_profile_query(capfd):
     assert '{"results": "some data"}' in out
     stubber.deactivate()
 
-    # --- Success case with empty payload ---
     stubber.activate()
     empty_payload = StreamingBody(io.BytesIO(b''), 0)
 
@@ -45,14 +44,13 @@ def test_execute_gremlin_profile_query(capfd):
     )
     run_open_cypher_query(client, GRAPH_ID)
     out, _ = capfd.readouterr()
-    assert out.strip() == ""  # Empty line
+    assert out.strip() == ""
     stubber.deactivate()
 
-    # --- ClientError case ---
     stubber.activate()
     stubber.stubber.add_client_error(
         "execute_query",
-        service_error_code="ValidationException",  # <-- Updated to a valid exception code
+        service_error_code="ValidationException",
         service_message="Client error occurred",
         http_status_code=400,
         expected_params={
@@ -66,8 +64,6 @@ def test_execute_gremlin_profile_query(capfd):
     assert "ClientError: Client error occurred" in out
     stubber.deactivate()
 
-    # --- Generic Exception case ---
-    # Deactivate stubber to allow monkeypatching client.execute_query
     stubber.deactivate()
 
     def raise_generic_error(**kwargs):
@@ -77,4 +73,3 @@ def raise_generic_error(**kwargs):
     run_open_cypher_query(client, GRAPH_ID)
     out, _ = capfd.readouterr()
     assert "Unexpected error: Generic failure" in out
-

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

@@ -2,10 +2,9 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import types
-import pytest
 from botocore.exceptions import ClientError, BotoCoreError
-from neptune_data_stubber import NeptuneDateStubber
-from database.neptune_execute_gremlin_profile_query import run_profile_query  # Adjust import path as needed
+from test_tools.neptune_data_stubber import NeptuneDateStubber
+from example_code.neptune.database.neptune_execute_gremlin_profile_query import run_profile_query
 
 def test_run_profile_query(capfd):
     stubber = NeptuneDateStubber()

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

@@ -1,11 +1,10 @@
-
 # 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 database.neptune_execute_gremlin_query import execute_gremlin_query
-from neptune_data_stubber import NeptuneDateStubber
+from test_tools.neptune_data_stubber import NeptuneDateStubber
+from example_code.neptune.database.neptune_execute_gremlin_query import execute_gremlin_query
 
 def test_execute_gremlin_query(capfd):
     stubber = NeptuneDateStubber()

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

@@ -1,19 +1,17 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: Apache-2.0
 
-import io
 import types
 from botocore.exceptions import ClientError, EndpointConnectionError
-from neptune_data_stubber import NeptuneDateStubber
-from database.neptune_execute_gremlin_explain_query import execute_gremlin_query
+from test_tools.neptune_data_stubber import NeptuneDateStubber
+from example_code.neptune.database.neptune_execute_gremlin_explain_query import execute_gremlin_query
 
 def test_execute_gremlin_query(capfd):
     stubber = NeptuneDateStubber()
     client = stubber.get_client()
     stubber.activate()
 
     try:
-        # --- Success case with valid StreamingBody output ---
         response_payload = '{"explain": "details"}'
         stubber.add_execute_gremlin_explain_query_stub(
             gremlin_query="g.V().has('code', 'ANC')",
@@ -26,7 +24,6 @@ def test_execute_gremlin_query(capfd):
         assert "Full Response:" in out
         assert '{"explain": "details"}' in out
 
-        # --- ClientError case ---
         stubber.stubber.assert_no_pending_responses()
         stubber.stubber.add_client_error(
             method='execute_gremlin_explain_query',
@@ -38,7 +35,6 @@ def test_execute_gremlin_query(capfd):
         out, _ = capfd.readouterr()
         assert "Error calling Neptune: Invalid query" in out
 
-        # --- EndpointConnectionError (BotoCoreError subclass) case ---
         stubber.stubber.assert_no_pending_responses()
 
         def raise_endpoint_connection_error(*args, **kwargs):

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

@@ -4,8 +4,8 @@
 import types
 from botocore.exceptions import ClientError, BotoCoreError
 from botocore.response import StreamingBody
-from database.neptune_execute_open_cypher_query import execute_open_cypher_explain_query
-from neptune_data_stubber import NeptuneDateStubber  # adjust import path accordingly
+from test_tools.neptune_data_stubber import NeptuneDateStubber
+from example_code.neptune.database.neptune_execute_open_cypher_query import execute_open_cypher_explain_query
 import io
 
 def test_execute_opencypher_explain_query(capfd):
@@ -14,7 +14,6 @@ def test_execute_opencypher_explain_query(capfd):
     stubber.activate()
 
     try:
-        # --- Case 1: Successful result (StreamingBody with bytes) ---
         explain_payload = 'mocked byte explain output'
         stubber.add_execute_open_cypher_explain_query_stub(
             open_cypher_query="MATCH (n {code: 'ANC'}) RETURN n",
@@ -28,7 +27,6 @@ def test_execute_opencypher_explain_query(capfd):
 
         stubber.stubber.assert_no_pending_responses()
 
-        # --- Case 2: No results (empty StreamingBody) ---
         empty_stream = StreamingBody(io.BytesIO(b""), 0)
         stubber.stubber.add_response(
             "execute_open_cypher_explain_query",
@@ -48,7 +46,6 @@ def test_execute_opencypher_explain_query(capfd):
     finally:
         stubber.deactivate()
 
-    # --- Case 3: ClientError ---
     def run_client_error_case():
         stubber = NeptuneDateStubber()
         client = stubber.get_client()
@@ -71,7 +68,6 @@ def run_client_error_case():
 
     run_client_error_case()
 
-    # --- Case 4: BotoCoreError (monkeypatch) ---
     def raise_boto_core_error(*args, **kwargs):
         raise BotoCoreError()
 
@@ -80,7 +76,6 @@ def raise_boto_core_error(*args, **kwargs):
     out, _ = capfd.readouterr()
     assert "BotoCore error:" in out
 
-    # --- Case 5: Generic Exception (monkeypatch) ---
     def raise_generic_exception(*args, **kwargs):
         raise Exception("Some generic error")
 

python/example_code/neptune/tests/test_check_instance_status.py

@@ -4,31 +4,18 @@
 import pytest
 from botocore.exceptions import ClientError
 import boto3
-from neptune_stubber import Neptune
-from neptune_scenario import check_instance_status  # your function to test
-
-# Constants for polling & timeout - patch if needed
-TIMEOUT_SECONDS = 10
-POLL_INTERVAL_SECONDS = 1
-
+from test_tools.neptune_stubber import Neptune
+from example_code.neptune.neptune_scenario import check_instance_status
 
 def test_check_instance_status_with_neptune_stubber(monkeypatch):
-    # Create real boto3 client + wrap with Neptune stubber
     client = boto3.client("neptune", region_name="us-east-1")
     stubber = Neptune(client)
 
     instance_id = "instance-1"
 
-    # Prepare stubbed responses for describe_db_instances paginator pages
-    # Each call to paginate() will return these pages in order:
-    # First call returns status 'starting', second returns 'available'
-    # Because the paginator returns an iterator of pages, each page is a dict
-
     stubbed_pages_starting = [{"DBInstances": [{"DBInstanceStatus": "starting"}]}]
     stubbed_pages_available = [{"DBInstances": [{"DBInstanceStatus": "available"}]}]
 
-    # We need to stub `describe_db_instances` for each paginator page request
-    # So stub two responses in sequence to simulate status change on subsequent polls
     stubber.stubber.add_response(
         "describe_db_instances",
         stubbed_pages_starting[0],
@@ -40,17 +27,18 @@ def test_check_instance_status_with_neptune_stubber(monkeypatch):
         expected_params={"DBInstanceIdentifier": instance_id},
     )
 
-    # Patch time.time to simulate time passing quickly (simulate elapsed time)
     times = [0, 1, 2, 3, 4, 5]
-    monkeypatch.setattr("neptune_scenario.time.time", lambda: times.pop(0) if times else 5)
-
-    # Patch time.sleep to avoid real wait during test
-    monkeypatch.setattr("neptune_scenario.time.sleep", lambda s: None)
-
-    # Patch format_elapsed_time to just return seconds + 's' string
-    monkeypatch.setattr("neptune_scenario.format_elapsed_time", lambda x: f"{x}s")
+    monkeypatch.setattr(
+        "example_code.neptune.neptune_scenario.time.time",
+        lambda: times.pop(0) if times else 5,
+    )
+    monkeypatch.setattr(
+        "example_code.neptune.neptune_scenario.time.sleep", lambda s: None
+    )
+    monkeypatch.setattr(
+        "example_code.neptune.neptune_scenario.format_elapsed_time", lambda x: f"{x}s"
+    )
 
-    # Run the check_instance_status function (should exit once status 'available' is found)
     check_instance_status(stubber.client, instance_id, "available")
 
 
@@ -60,50 +48,54 @@ def test_check_instance_status_timeout(monkeypatch):
 
     instance_id = "instance-timeout"
 
-    # Always return status 'starting' to simulate never reaching 'available'
     stub_response = {"DBInstances": [{"DBInstanceStatus": "starting"}]}
 
-    # Stub multiple responses (enough for timeout loops)
     for _ in range(10):
         stubber.stubber.add_response(
             "describe_db_instances",
             stub_response,
             expected_params={"DBInstanceIdentifier": instance_id},
         )
 
-    # Patch time.time to simulate time passing beyond timeout (simulate elapsed time)
-    times = list(range(15))  # simulate 15 seconds
-    monkeypatch.setattr("neptune_scenario.time.time", lambda: times.pop(0) if times else 15)
-
-    monkeypatch.setattr("neptune_scenario.time.sleep", lambda s: None)
-    monkeypatch.setattr("neptune_scenario.format_elapsed_time", lambda x: f"{x}s")
+    times = list(range(15))
+    monkeypatch.setattr(
+        "example_code.neptune.neptune_scenario.time.time",
+        lambda: times.pop(0) if times else 15,
+    )
+    monkeypatch.setattr(
+        "example_code.neptune.neptune_scenario.time.sleep", lambda s: None
+    )
+    monkeypatch.setattr(
+        "example_code.neptune.neptune_scenario.format_elapsed_time", lambda x: f"{x}s"
+    )
 
-    # Patch timeout and poll interval inside your module (adjust as needed)
-    monkeypatch.setattr("neptune_scenario.TIMEOUT_SECONDS", 5)
-    monkeypatch.setattr("neptune_scenario.POLL_INTERVAL_SECONDS", 1)
+    monkeypatch.setattr("example_code.neptune.neptune_scenario.TIMEOUT_SECONDS", 5)
+    monkeypatch.setattr("example_code.neptune.neptune_scenario.POLL_INTERVAL_SECONDS", 1)
 
-    with pytest.raises(RuntimeError, match="Timeout waiting for 'instance-timeout'"):
+    with pytest.raises(RuntimeError, match=f"Timeout waiting for '{instance_id}'"):
         check_instance_status(stubber.client, instance_id, "available")
 
-
 def test_check_instance_status_client_error(monkeypatch):
-    client = boto3.client("neptune")
+    client = boto3.client("neptune", region_name="us-east-1")
     stubber = Neptune(client)
 
     instance_id = "not-there"
 
-    # Stub a ClientError for describe_db_instances
     stubber.stubber.add_client_error(
         "describe_db_instances",
         service_error_code="DBInstanceNotFound",
         service_message="Instance not found",
         expected_params={"DBInstanceIdentifier": instance_id},
     )
 
-    # Patch time.sleep and format_elapsed_time to avoid delays and keep output clean
-    monkeypatch.setattr("neptune_scenario.time.sleep", lambda s: None)
-    monkeypatch.setattr("neptune_scenario.format_elapsed_time", lambda x: f"{x}s")
+    monkeypatch.setattr(
+        "example_code.neptune.neptune_scenario.time.sleep", lambda s: None
+    )
+    monkeypatch.setattr(
+        "example_code.neptune.neptune_scenario.format_elapsed_time", lambda x: f"{x}s"
+    )
 
     with pytest.raises(ClientError, match="Instance not found"):
         check_instance_status(stubber.client, instance_id, "available")
 
+

python/example_code/neptune/tests/test_create_db_cluster.py

@@ -3,20 +3,15 @@
 
 import pytest
 import boto3
-from botocore.exceptions import ClientError
-from neptune_stubber import Neptune
-from neptune_scenario import create_db_cluster  # Your actual function
-
+from test_tools.neptune_stubber import Neptune
+from example_code.neptune.neptune_scenario import create_db_cluster
 
 def test_create_db_cluster():
-    # 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, deletion_protection, backup_retention_period defaulted
     )
     cluster_id = create_db_cluster(stubber.client, "test-cluster")
     assert cluster_id == "test-cluster"
@@ -34,7 +29,6 @@ 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 in RuntimeError ---
     stubber.stub_create_db_cluster(
         cluster_id="denied-cluster",
         error_code="AccessDenied"
@@ -43,7 +37,6 @@ def test_create_db_cluster():
         create_db_cluster(stubber.client, "denied-cluster")
     assert "AWS error [AccessDenied]" in str(exc_info.value)
 
-    # --- Unexpected exception raises RuntimeError ---
     def raise_generic_exception(**kwargs):
         raise Exception("Unexpected failure")