|
1 | 1 | # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. |
2 | 2 | # SPDX-License-Identifier: Apache-2.0 |
3 | 3 |
|
| 4 | +import boto3 |
4 | 5 | import pytest |
5 | | -from unittest.mock import MagicMock |
| 6 | +from botocore.stub import Stubber |
6 | 7 | from botocore.exceptions import ClientError |
7 | | - |
8 | | -from hello_neptune import describe_db_clusters # replace with actual import |
| 8 | +from hello_neptune import describe_db_clusters |
9 | 9 |
|
10 | 10 |
|
11 | 11 | @pytest.fixture |
12 | | -def mock_neptune_client(): |
13 | | - """Return a mocked boto3 Neptune client.""" |
14 | | - return MagicMock() |
15 | | - |
16 | | - |
17 | | -def test_describe_db_clusters_unit(mock_neptune_client, capsys): |
18 | | - """ |
19 | | - Unit test for describe_db_clusters with paginator. |
20 | | - Mocks the Neptune client's paginator and verifies expected output is printed. |
21 | | - """ |
22 | | - |
23 | | - # Create a mock paginator |
24 | | - mock_paginator = MagicMock() |
25 | | - mock_neptune_client.get_paginator.return_value = mock_paginator |
26 | | - |
27 | | - # Mock pages returned by paginate() |
28 | | - mock_paginator.paginate.return_value = [ |
29 | | - { |
30 | | - "DBClusters": [ |
31 | | - { |
32 | | - "DBClusterIdentifier": "my-test-cluster", |
33 | | - "Status": "available" |
34 | | - } |
35 | | - ] |
36 | | - }, |
37 | | - { |
38 | | - "DBClusters": [ |
39 | | - { |
40 | | - "DBClusterIdentifier": "my-second-cluster", |
41 | | - "Status": "modifying" |
42 | | - } |
43 | | - ] |
44 | | - } |
45 | | - ] |
46 | | - |
47 | | - try: |
48 | | - describe_db_clusters(mock_neptune_client) |
49 | | - captured = capsys.readouterr() |
50 | | - |
51 | | - # Check that expected outputs from both pages were printed |
52 | | - assert "my-test-cluster" in captured.out |
53 | | - assert "available" in captured.out |
54 | | - assert "my-second-cluster" in captured.out |
55 | | - assert "modifying" in captured.out |
56 | | - |
57 | | - mock_neptune_client.get_paginator.assert_called_once_with("describe_db_clusters") |
58 | | - mock_paginator.paginate.assert_called_once() |
59 | | - |
60 | | - except ClientError as e: |
61 | | - pytest.fail(f"AWS ClientError occurred: {e.response['Error']['Message']}") |
62 | | - except Exception as e: |
63 | | - pytest.fail(f"Unexpected error: {str(e)}") |
| 12 | +def neptune_client_stub(): |
| 13 | + client = boto3.client("neptune", region_name="us-east-1") |
| 14 | + stubber = Stubber(client) |
| 15 | + stubber.activate() |
| 16 | + yield client, stubber |
| 17 | + stubber.deactivate() |
| 18 | + |
| 19 | + |
| 20 | +def test_describe_db_clusters_with_stubber_single_page(neptune_client_stub, capsys): |
| 21 | + client, stubber = neptune_client_stub |
| 22 | + |
| 23 | + # Simulate a single-page paginator result with both clusters in one call |
| 24 | + stubber.add_response("describe_db_clusters", { |
| 25 | + "DBClusters": [ |
| 26 | + {"DBClusterIdentifier": "my-test-cluster", "Status": "available"}, |
| 27 | + {"DBClusterIdentifier": "my-second-cluster", "Status": "modifying"} |
| 28 | + ] |
| 29 | + }) |
| 30 | + |
| 31 | + describe_db_clusters(client) |
| 32 | + |
| 33 | + captured = capsys.readouterr() |
| 34 | + |
| 35 | + assert "my-test-cluster" in captured.out |
| 36 | + assert "available" in captured.out |
| 37 | + assert "my-second-cluster" in captured.out |
| 38 | + assert "modifying" in captured.out |
0 commit comments