From 67d5256ef41e958fd0524eb6307aaf49888f40ed Mon Sep 17 00:00:00 2001 From: Wolf Date: Thu, 27 Jun 2024 13:44:45 +0100 Subject: [PATCH 1/2] Add profiles --- CITATION.cff | 2 +- README.md | 1 + setup.py | 2 +- tests/conftest.py | 43 +++++++++-------- tests/test_get_aws_regions.py | 30 ++++++------ wolfsoftware/get_aws_regions/functions.py | 56 ++++++++++++++--------- 6 files changed, 73 insertions(+), 61 deletions(-) diff --git a/CITATION.cff b/CITATION.cff index 6850974..3ab659c 100644 --- a/CITATION.cff +++ b/CITATION.cff @@ -3,7 +3,7 @@ message: If you use this software, please cite it using these metadata. title: Get AWS Regions Package abstract: A simple package for retrieving a list of AWS regions. type: software -version: 0.1.0 +version: 0.1.1 date-released: 2024-06-27 repository-code: https://github.com/AWSToolbox/get-aws-regions-package keywords: diff --git a/README.md b/README.md index e011f4d..a0886d8 100644 --- a/README.md +++ b/README.md @@ -67,6 +67,7 @@ for all of the tools in our [AWS Toolbox](https://github.com/AWSToolbox). - `exclude_list`: Optional list of regions to exclude. - `all_regions`: Boolean flag to include all regions (default: True). - `details`: Boolean flag to return detailed information about each region (default: False). + - `profile_name`: String to specify the name of the profile to use. - Returns: - If `details=True`: Sorted list of dictionaries containing detailed region information. - If `details=False`: Sorted list of region names as strings. diff --git a/setup.py b/setup.py index f3f15a3..d483144 100644 --- a/setup.py +++ b/setup.py @@ -12,7 +12,7 @@ setup( name='wolfsoftware.get-aws-regions', - version='0.1.0', + version='0.1.1', author='Wolf Software', author_email='pypi@wolfsoftware.com', description='A simple package for retrieving a list of AWS regions.', diff --git a/tests/conftest.py b/tests/conftest.py index fec5ee5..694e06f 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -2,19 +2,18 @@ Configuration for pytest tests, including fixtures and mock data for testing the wolfsoftware.get-aws-regions package. Fixtures: -- boto3_client_mock: Mocks the boto3 client for AWS interactions. +- boto3_session_mock: Mocks the boto3 session for AWS interactions. +- boto3_session_mock_with_exception: Mocks the boto3 session raising an exception during describe_regions. Mock Data: - mock_regions: A list of dictionaries representing mock AWS regions. - mock_description: A dictionary representing the mock description of a specific AWS region. """ -from typing import Any, Dict, Generator, List, Union - -from unittest.mock import AsyncMock, MagicMock, patch +from typing import Any, Dict, Generator, List +from unittest.mock import MagicMock, patch import pytest - # Mock data mock_regions: List[Dict[str, str]] = [ {"RegionName": "us-east-1", "OptInStatus": "opt-in-not-required"}, @@ -30,14 +29,16 @@ @pytest.fixture -def boto3_client_mock(mock_exception: Union[Exception, None] = None) -> Generator[Union[MagicMock, AsyncMock], Any, None]: +def boto3_session_mock() -> Generator[MagicMock, None, None]: """ - Fixture to mock the boto3 client. + Fixture to mock the boto3 session. Yields: - MagicMock: A mock of the boto3 client. + MagicMock: A mock of the boto3 session. """ - with patch("boto3.client") as mock_client: + with patch("boto3.Session") as mock_session: + mock_session_instance: Any = mock_session.return_value + regions_mock = MagicMock() ssm_mock = MagicMock() @@ -47,27 +48,25 @@ def mock_get_parameter(Name) -> Dict[str, Dict[str, str]]: raise ValueError(f"Unknown parameter name: {Name}") ssm_mock.get_parameter.side_effect = mock_get_parameter - mock_client.side_effect = lambda service_name, *args, **kwargs: ssm_mock if service_name == "ssm" else regions_mock - # Conditionally set side effect for describe_regions based on mock_exception - if mock_exception: - print("HERE") - regions_mock.describe_regions.side_effect = mock_exception - regions_mock.describe_regions.side_effect = mock_exception + mock_session_instance.client.side_effect = lambda service_name, *args, **kwargs: ssm_mock if service_name == "ssm" else regions_mock + regions_mock.describe_regions.return_value = {"Regions": mock_regions} - yield mock_client + yield mock_session @pytest.fixture -def boto3_client_mock_with_exception() -> Generator[MagicMock, None, None]: +def boto3_session_mock_with_exception() -> Generator[MagicMock, None, None]: """ - Fixture to mock the boto3 client raising an exception during describe_regions. + Fixture to mock the boto3 session raising an exception during describe_regions. Yields: - MagicMock: A mock of the boto3 client. + MagicMock: A mock of the boto3 session. """ - with patch("boto3.client") as mock_client: + with patch("boto3.Session") as mock_session: + mock_session_instance: Any = mock_session.return_value + regions_mock = MagicMock() ssm_mock = MagicMock() @@ -80,6 +79,6 @@ def mock_get_parameter(Name) -> Dict[str, Dict[str, str]]: regions_mock.describe_regions.side_effect = Exception("Test Exception") - mock_client.side_effect = lambda service_name, *args, **kwargs: ssm_mock if service_name == "ssm" else regions_mock + mock_session_instance.client.side_effect = lambda service_name, *args, **kwargs: ssm_mock if service_name == "ssm" else regions_mock - yield mock_client + yield mock_session diff --git a/tests/test_get_aws_regions.py b/tests/test_get_aws_regions.py index 8e26c92..c83e9e3 100644 --- a/tests/test_get_aws_regions.py +++ b/tests/test_get_aws_regions.py @@ -37,19 +37,18 @@ def test_version() -> None: assert version != 'unknown', f"Expected version, but got {version}" # nosec: B101 -def test_get_region_list_all_regions(boto3_client_mock) -> None: +def test_get_region_list_all_regions(boto3_session_mock) -> None: """ Test fetching all regions with detailed information. Arguments: - boto3_client_mock (fixture): The mocked boto3 client. + boto3_session_mock (fixture): The mocked boto3 session. """ - regions_mock: Any = boto3_client_mock.return_value + regions_mock: Any = boto3_session_mock.return_value.client.return_value regions_mock.describe_regions.return_value = {"Regions": mock_regions} result: List[Dict[str, str | bool]] | List[str] = get_region_list(details=True) result.sort(key=lambda x: x["RegionName"]) # Sort the result for consistent ordering - print("Filtered Regions with Details:", result) expected_result: List[Dict[str, str]] = [ {"RegionName": "us-east-1", "OptInStatus": "opt-in-not-required", "GeographicalLocation": "US East (N. Virginia)"}, @@ -57,19 +56,18 @@ def test_get_region_list_all_regions(boto3_client_mock) -> None: {"RegionName": "eu-west-1", "OptInStatus": "opted-in", "GeographicalLocation": "EU West (Ireland)"} ] expected_result.sort(key=lambda x: x["RegionName"]) # Sort the expected result for consistent ordering - print("Expected Result:", expected_result) assert result == expected_result # nosec: B101 -def test_get_region_list_include_filter(boto3_client_mock) -> None: +def test_get_region_list_include_filter(boto3_session_mock) -> None: """ Test fetching regions with an include filter. Arguments: - boto3_client_mock (fixture): The mocked boto3 client. + boto3_session_mock (fixture): The mocked boto3 session. """ - regions_mock: Any = boto3_client_mock.return_value + regions_mock: Any = boto3_session_mock.return_value.client.return_value regions_mock.describe_regions.return_value = {"Regions": mock_regions} @@ -85,14 +83,14 @@ def test_get_region_list_include_filter(boto3_client_mock) -> None: assert result == expected_result # nosec: B101 -def test_get_region_list_exclude_filter(boto3_client_mock) -> None: +def test_get_region_list_exclude_filter(boto3_session_mock) -> None: """ Test fetching regions with an exclude filter. Arguments: - boto3_client_mock (fixture): The mocked boto3 client. + boto3_session_mock (fixture): The mocked boto3 session. """ - regions_mock: Any = boto3_client_mock.return_value + regions_mock: Any = boto3_session_mock.return_value.client.return_value regions_mock.describe_regions.return_value = {"Regions": mock_regions} @@ -108,14 +106,14 @@ def test_get_region_list_exclude_filter(boto3_client_mock) -> None: assert result == expected_result # nosec: B101 -def test_get_region_list_no_details(boto3_client_mock) -> None: +def test_get_region_list_no_details(boto3_session_mock) -> None: """ Test fetching region names without details. Arguments: - boto3_client_mock (fixture): The mocked boto3 client. + boto3_session_mock (fixture): The mocked boto3 session. """ - regions_mock: Any = boto3_client_mock.return_value + regions_mock: Any = boto3_session_mock.return_value.client.return_value regions_mock.describe_regions.return_value = {"Regions": mock_regions} @@ -128,12 +126,12 @@ def test_get_region_list_no_details(boto3_client_mock) -> None: assert result == expected_result # nosec: B101 -def test_get_region_list_exceptions(boto3_client_mock_with_exception) -> None: # pylint: disable=unused-argument +def test_get_region_list_exceptions(boto3_session_mock_with_exception) -> None: # pylint: disable=unused-argument """ Test exception handling when an error occurs in fetching regions. Arguments: - boto3_client_mock (fixture): The mocked boto3 client. + boto3_session_mock (fixture): The mocked boto3 session. """ # Use pytest.raises to catch RegionListingError with pytest.raises(RegionListingError) as excinfo: diff --git a/wolfsoftware/get_aws_regions/functions.py b/wolfsoftware/get_aws_regions/functions.py index 406e4c6..ac7e85c 100644 --- a/wolfsoftware/get_aws_regions/functions.py +++ b/wolfsoftware/get_aws_regions/functions.py @@ -1,18 +1,20 @@ """ -This module provides functionalities for retrieving and processing AWS region information. +This module provides functionalities for retrieving and processing AWS region information +with support for multiple AWS profiles. -It includes the following capabilities: +Capabilities: - Fetching a list of all AWS regions, with the option to include or exclude regions based on account opt-in status. - Retrieving the geographical location description for a specific AWS region from the AWS Systems Manager (SSM) Parameter Store. - Fetching geographical locations for multiple AWS regions using concurrent threading for improved performance. - Applying include and exclude filters to the list of regions to customize the output. - Generating a comprehensive list of AWS regions, with optional detailed information about each region. + - Utilizing different AWS profiles for the retrieval of region information. Functions: - - get_region_list: Main function to retrieve a list of AWS regions, optionally filtering by include and exclude lists and returning - detailed information if required. + - get_region_list: Main function to retrieve a list of AWS regions, optionally filtering by include and exclude lists, + and returning detailed information if required. - Private Functions: +Private Functions: - _fetch_all_regions: Retrieves a list of all AWS regions. - _fetch_region_description: Fetches the geographical location for a specific AWS region. - _fetch_region_descriptions: Fetches geographical locations for multiple AWS regions using threading. @@ -24,28 +26,34 @@ Dependencies: - boto3: AWS SDK for Python to interact with various AWS services like EC2 and SSM. - concurrent.futures: Standard library module to enable asynchronous execution using threading. + - botocore.exceptions: Exceptions for handling errors during boto3 operations. Usage: - This module is intended to be used as part of the wolfsoftware.get-aws-regions package. The main entry point is the `get_region_list` function, - which provides flexibility in retrieving and customizing the list of AWS regions based on user-defined criteria. + This module is intended to be used as part of the wolfsoftware.get-aws-regions package. + The main entry point is the `get_region_list` function, which provides flexibility in retrieving + and customizing the list of AWS regions based on user-defined criteria, including the ability + to specify different AWS profiles. """ + from typing import Any, List, Dict, Optional, Union from concurrent.futures._base import Future from concurrent.futures import ThreadPoolExecutor, as_completed import boto3 # pylint: disable=import-error +from botocore.exceptions import BotoCoreError, ClientError from .exceptions import RegionListingError -def _fetch_all_regions(all_regions: bool = True) -> List[Dict[str, Union[str, bool]]]: +def _fetch_all_regions(all_regions: bool = True, profile_name: Optional[str] = None) -> List[Dict[str, Union[str, bool]]]: """ Retrieve a list of all AWS regions. Arguments: all_regions (bool): If True, list all available regions, including those not opted into. If False, list only regions opted into by the account. + profile_name (Optional[str]): The name of the AWS profile to use. Returns: List[Dict[str, Union[str, bool]]]: A list of dictionaries containing information about each region. @@ -54,8 +62,9 @@ def _fetch_all_regions(all_regions: bool = True) -> List[Dict[str, Union[str, bo RegionListingError: If there is an error in retrieving the regions. """ try: - # Initialize a session using Amazon EC2 - ec2: Any = boto3.client('ec2') + # Initialize a session using Amazon EC2 with the specified profile + session = boto3.Session(profile_name=profile_name) if profile_name else boto3.Session() + ec2: Any = session.client('ec2') # Retrieve a list of all available regions or only opted-in regions based on the flag if all_regions: @@ -71,18 +80,19 @@ def _fetch_all_regions(all_regions: bool = True) -> List[Dict[str, Union[str, bo return regions - except boto3.exceptions.Boto3Error as e: + except (BotoCoreError, ClientError) as e: raise RegionListingError(f"An error occurred while listing regions: {str(e)}") from e except Exception as e: raise RegionListingError(f"An unexpected error occurred: {str(e)}") from e -def _fetch_region_description(region_name: str) -> Dict[str, str]: +def _fetch_region_description(region_name: str, profile_name: Optional[str] = None) -> Dict[str, str]: """ Fetch the geographical location for a specific AWS region from SSM Parameter Store. Arguments: region_name (str): The name of the region to fetch the geographical location for. + profile_name (Optional[str]): The name of the AWS profile to use. Returns: Dict[str, str]: A dictionary containing the region name and its geographical location. @@ -91,8 +101,9 @@ def _fetch_region_description(region_name: str) -> Dict[str, str]: RegionListingError: If there is an error in retrieving the region geographical location. """ try: - # Initialize a session using Amazon SSM - ssm: Any = boto3.client('ssm') + # Initialize a session using Amazon SSM with the specified profile + session = boto3.Session(profile_name=profile_name) if profile_name else boto3.Session() + ssm: Any = session.client('ssm') # Retrieve the parameter for the region description parameter_name: str = f"/aws/service/global-infrastructure/regions/{region_name}/longName" @@ -100,18 +111,19 @@ def _fetch_region_description(region_name: str) -> Dict[str, str]: return {region_name: response['Parameter']['Value']} - except boto3.exceptions.Boto3Error as e: + except (BotoCoreError, ClientError) as e: raise RegionListingError(f"An error occurred while retrieving geographical location for region {region_name}: {str(e)}") from e except Exception as e: raise RegionListingError(f"An unexpected error occurred: {str(e)}") from e -def _fetch_region_descriptions(region_names: List[str]) -> Dict[str, str]: +def _fetch_region_descriptions(region_names: List[str], profile_name: Optional[str] = None) -> Dict[str, str]: """ Fetch geographical locations for multiple AWS regions from SSM Parameter Store using threading for better performance. Arguments: region_names (List[str]): A list of region names to fetch geographical locations for. + profile_name (Optional[str]): The name of the AWS profile to use. Returns: Dict[str, str]: A dictionary mapping region codes to their geographical locations. @@ -122,7 +134,7 @@ def _fetch_region_descriptions(region_names: List[str]) -> Dict[str, str]: descriptions: Dict = {} with ThreadPoolExecutor() as executor: - future_to_region: Dict[Future[Dict[str, str]], str] = {executor.submit(_fetch_region_description, region): region for region in region_names} + future_to_region: Dict[Future[Dict[str, str]], str] = {executor.submit(_fetch_region_description, region, profile_name): region for region in region_names} for future in as_completed(future_to_region): region: str = future_to_region[future] @@ -166,8 +178,9 @@ def _apply_region_filters( def get_region_list( include_list: Optional[List[str]] = None, exclude_list: Optional[List[str]] = None, - all_regions: bool = True, - details: bool = False + all_regions: Optional[bool] = True, + details: Optional[bool] = False, + profile_name: Optional[str] = None ) -> Union[List[Dict[str, Union[str, bool]]], List[str]]: """ Retrieve a list of AWS regions, optionally filtering by include and exclude lists. @@ -179,6 +192,7 @@ def get_region_list( exclude_list (Optional[List[str]]): A list of regions to exclude. These regions will be omitted from the returned list if specified. all_regions (bool): If True, list all available regions, including those not opted into. If False, list only regions opted into by the account. details (bool): If True, return detailed information about each region. If False, return only the region names. + profile_name (Optional[str]): The name of the AWS profile to use. Returns: Union[List[Dict[str, Union[str, bool]]], List[str]]: A sorted list of regions with detailed information or just the region names. @@ -187,7 +201,7 @@ def get_region_list( RegionListingError: If there is an error in retrieving the regions. """ try: - all_regions_list: List[Dict[str, str | bool]] = _fetch_all_regions(all_regions) + all_regions_list: List[Dict[str, str | bool]] = _fetch_all_regions(all_regions, profile_name) except Exception as e: raise RegionListingError(f"An error occurred while retrieving regions: {str(e)}") from e @@ -195,7 +209,7 @@ def get_region_list( if details: region_names: List[str | bool] = [region['RegionName'] for region in filtered_regions] - region_descriptions: Dict[str, str] = _fetch_region_descriptions(region_names) + region_descriptions: Dict[str, str] = _fetch_region_descriptions(region_names, profile_name) for region in filtered_regions: region['GeographicalLocation'] = region_descriptions.get(region['RegionName'], "Unknown") print("Filtered Regions with Details:", filtered_regions) # Debug print From 2a2604ad5ab8e45b4b7264525ff4a8f4f3aafc67 Mon Sep 17 00:00:00 2001 From: Wolf Date: Thu, 27 Jun 2024 13:48:29 +0100 Subject: [PATCH 2/2] Fix pipelines errors --- wolfsoftware/get_aws_regions/functions.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/wolfsoftware/get_aws_regions/functions.py b/wolfsoftware/get_aws_regions/functions.py index ac7e85c..6f39f24 100644 --- a/wolfsoftware/get_aws_regions/functions.py +++ b/wolfsoftware/get_aws_regions/functions.py @@ -1,6 +1,5 @@ """ -This module provides functionalities for retrieving and processing AWS region information -with support for multiple AWS profiles. +This module provides functionalities for retrieving and processing AWS region information with support for multiple AWS profiles. Capabilities: - Fetching a list of all AWS regions, with the option to include or exclude regions based on account opt-in status. @@ -134,7 +133,10 @@ def _fetch_region_descriptions(region_names: List[str], profile_name: Optional[s descriptions: Dict = {} with ThreadPoolExecutor() as executor: - future_to_region: Dict[Future[Dict[str, str]], str] = {executor.submit(_fetch_region_description, region, profile_name): region for region in region_names} + future_to_region: Dict[Future[Dict[str, str]], str] = { + executor.submit(_fetch_region_description, region, profile_name): region + for region in region_names + } for future in as_completed(future_to_region): region: str = future_to_region[future]