Refacter get_aws_connection_info / get_aws_region (#1231)

Refacter get_aws_connection_info / get_aws_region

Depends-On: #1246
SUMMARY
Splits up the logic for get_aws_region and get_aws_connection_info so that it can be reused by non-module plugins.
ISSUE TYPE

Feature Pull Request

COMPONENT NAME
plugins/module_utils/botocore.py
ADDITIONAL INFORMATION

Reviewed-by: Alina Buzachis <None>

changelogs/fragments/1231-boto3_connections.yml

@@ -0,0 +1,2 @@
+minor_changes:
+- module_utils.botocore - refactorization of ``get_aws_region``, ``get_aws_connection_info`` so that the code can be reused by non-module plugins (https://github.com/ansible-collections/amazon.aws/pull/1231).

plugins/module_utils/botocore.py

@@ -142,42 +142,55 @@ def boto_exception(err):
     return error
 
 
-def get_aws_region(module, boto3=None):
-    region = module.params.get('region')
+def _aws_region(params):
+    region = params.get('region')
 
     if region:
         return region
 
     if not HAS_BOTO3:
-        module.fail_json(msg=missing_required_lib('boto3'), exception=BOTO3_IMP_ERR)
+        raise AnsibleBotocoreError(
+            message=missing_required_lib('boto3 and botocore'),
+            exception=BOTO3_IMP_ERR
+        )
 
     # here we don't need to make an additional call, will default to 'us-east-1' if the below evaluates to None.
     try:
         # Botocore doesn't like empty strings, make sure we default to None in the case of an empty
         # string.
-        profile_name = module.params.get('profile') or None
+        profile_name = params.get('profile') or None
         return botocore.session.Session(profile=profile_name).get_config_variable('region')
     except botocore.exceptions.ProfileNotFound:
         return None
 
 
-def get_aws_connection_info(module, boto3=None):
-
-    # Check module args for credentials, then check environment vars
-    # access_key
-
-    endpoint_url = module.params.get('endpoint_url')
-    access_key = module.params.get('access_key')
-    secret_key = module.params.get('secret_key')
-    session_token = module.params.get('session_token')
-    region = get_aws_region(module)
-    profile_name = module.params.get('profile')
-    validate_certs = module.params.get('validate_certs')
-    ca_bundle = module.params.get('aws_ca_bundle')
-    config = module.params.get('aws_config')
+def get_aws_region(module, boto3=None):
 
+    try:
+        return _aws_region(module.params)
+    except AnsibleBotocoreError as e:
+        if e.exception:
+            module.fail_json(msg=e.message, exception=e.exception)
+        else:
+            module.fail_json(msg=e.message)
+
+
+def _aws_connection_info(params):
+
+    endpoint_url = params.get('endpoint_url')
+    access_key = params.get('access_key')
+    secret_key = params.get('secret_key')
+    session_token = params.get('session_token')
+    region = _aws_region(params)
+    profile_name = params.get('profile')
+    validate_certs = params.get('validate_certs')
+    ca_bundle = params.get('aws_ca_bundle')
+    config = params.get('aws_config')
+
+    # Caught here so that they can be deliberately set to '' to avoid conflicts when environment
+    # variables are also being used
     if profile_name and (access_key or secret_key or session_token):
-        module.fail_json(msg="Passing both a profile and access tokens is not supported.")
+        raise AnsibleBotocoreError(message="Passing both a profile and access tokens is not supported.")
 
     # Botocore doesn't like empty strings, make sure we default to None in the case of an empty
     # string.
@@ -217,6 +230,16 @@ def get_aws_connection_info(module, boto3=None):
     return region, endpoint_url, boto_params
 
 
+def get_aws_connection_info(module, boto3=None):
+    try:
+        return _aws_connection_info(module.params)
+    except AnsibleBotocoreError as e:
+        if e.exception:
+            module.fail_json(msg=e.message, exception=e.exception)
+        else:
+            module.fail_json(msg=e.message)
+
+
 def _paginated_query(client, paginator_name, **params):
     paginator = client.get_paginator(paginator_name)
     result = paginator.paginate(**params).build_full_result()

tests/unit/module_utils/botocore/test_aws_region.py

@@ -0,0 +1,200 @@
+# (c) 2022 Red Hat Inc.
+#
+# This file is part of Ansible
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+import pytest
+from unittest.mock import MagicMock
+from unittest.mock import sentinel
+from unittest.mock import call
+
+try:
+    import botocore
+except ImportError:
+    # Handled by HAS_BOTO3
+    pass
+
+import ansible_collections.amazon.aws.plugins.module_utils.botocore as utils_botocore
+from ansible_collections.amazon.aws.plugins.module_utils.exceptions import AnsibleBotocoreError
+
+
+class FailException(Exception):
+    pass
+
+
+@pytest.fixture
+def aws_module(monkeypatch):
+    aws_module = MagicMock()
+    aws_module.fail_json.side_effect = FailException()
+    aws_module.fail_json_aws.side_effect = FailException()
+    monkeypatch.setattr(aws_module, 'params', sentinel.MODULE_PARAMS)
+    return aws_module
+
+
+@pytest.fixture
+def fake_botocore(monkeypatch):
+    # Note: this isn't a monkey-patched real-botocore, this is a complete fake.
+    fake_session = MagicMock()
+    fake_session.get_config_variable.return_value = sentinel.BOTO3_REGION
+    fake_session_module = MagicMock()
+    fake_session_module.Session.return_value = fake_session
+    fake_botocore = MagicMock()
+    monkeypatch.setattr(fake_botocore, 'session', fake_session_module)
+    # Patch exceptions back in
+    monkeypatch.setattr(fake_botocore, 'exceptions', botocore.exceptions)
+
+    return fake_botocore
+
+
+@pytest.fixture
+def botocore_utils(monkeypatch):
+    return utils_botocore
+
+
+###############################################################
+# module_utils.botocore.get_aws_region
+###############################################################
+def test_get_aws_region_simple(monkeypatch, aws_module, botocore_utils):
+    region_method = MagicMock(name='_aws_region')
+    monkeypatch.setattr(botocore_utils, '_aws_region', region_method)
+    region_method.return_value = sentinel.RETURNED_REGION
+
+    assert botocore_utils.get_aws_region(aws_module) is sentinel.RETURNED_REGION
+    passed_args = region_method.call_args
+    assert passed_args == call(sentinel.MODULE_PARAMS)
+    # args[0]
+    assert passed_args[0][0] is sentinel.MODULE_PARAMS
+
+
+def test_get_aws_region_exception_nested(monkeypatch, aws_module, botocore_utils):
+    region_method = MagicMock(name='_aws_region')
+    monkeypatch.setattr(botocore_utils, '_aws_region', region_method)
+
+    exception_nested = AnsibleBotocoreError(message=sentinel.ERROR_MSG, exception=sentinel.ERROR_EX)
+    region_method.side_effect = exception_nested
+
+    with pytest.raises(FailException):
+        assert botocore_utils.get_aws_region(aws_module)
+
+    passed_args = region_method.call_args
+    assert passed_args == call(sentinel.MODULE_PARAMS)
+    # call_args[0] == positional args
+    assert passed_args[0][0] is sentinel.MODULE_PARAMS
+
+    fail_args = aws_module.fail_json.call_args
+    assert fail_args == call(msg=sentinel.ERROR_MSG, exception=sentinel.ERROR_EX)
+    # call_args[1] == kwargs
+    assert fail_args[1]['msg'] is sentinel.ERROR_MSG
+    assert fail_args[1]['exception'] is sentinel.ERROR_EX
+
+
+def test_get_aws_region_exception_msg(monkeypatch, aws_module, botocore_utils):
+    region_method = MagicMock(name='_aws_region')
+    monkeypatch.setattr(botocore_utils, '_aws_region', region_method)
+
+    exception_nested = AnsibleBotocoreError(message=sentinel.ERROR_MSG)
+    region_method.side_effect = exception_nested
+
+    with pytest.raises(FailException):
+        assert botocore_utils.get_aws_region(aws_module)
+
+    passed_args = region_method.call_args
+    assert passed_args == call(sentinel.MODULE_PARAMS)
+    # call_args[0] == positional args
+    assert passed_args[0][0] is sentinel.MODULE_PARAMS
+
+    fail_args = aws_module.fail_json.call_args
+    assert fail_args == call(msg=sentinel.ERROR_MSG)
+    # call_args[1] == kwargs
+    assert fail_args[1]['msg'] is sentinel.ERROR_MSG
+
+
+###############################################################
+# module_utils.botocore._aws_region
+###############################################################
+def test_aws_region_no_boto(monkeypatch, botocore_utils):
+    monkeypatch.setattr(botocore_utils, 'HAS_BOTO3', False)
+    monkeypatch.setattr(botocore_utils, 'BOTO3_IMP_ERR', sentinel.BOTO3_IMPORT_EXCEPTION)
+
+    assert botocore_utils._aws_region(dict(region=sentinel.PARAM_REGION)) is sentinel.PARAM_REGION
+
+    with pytest.raises(AnsibleBotocoreError) as e:
+        utils_botocore._aws_region(dict())
+    assert 'boto3' in e.value.message
+    assert 'botocore' in e.value.message
+    assert e.value.exception is sentinel.BOTO3_IMPORT_EXCEPTION
+
+
+def test_aws_region_no_profile(monkeypatch, botocore_utils, fake_botocore):
+
+    monkeypatch.setattr(botocore_utils, 'botocore', fake_botocore)
+    fake_session_module = fake_botocore.session
+    fake_session = fake_session_module.Session(sentinel.RETRIEVAL)
+
+    assert botocore_utils._aws_region(dict(region=sentinel.PARAM_REGION)) is sentinel.PARAM_REGION
+    assert fake_session_module.Session.call_args == call(sentinel.RETRIEVAL)
+
+    assert botocore_utils._aws_region(dict()) is sentinel.BOTO3_REGION
+    assert fake_session_module.Session.call_args == call(profile=None)
+    assert fake_session.get_config_variable.call_args == call('region')
+
+
+def test_aws_region_none_profile(monkeypatch, botocore_utils, fake_botocore):
+
+    monkeypatch.setattr(botocore_utils, 'botocore', fake_botocore)
+    fake_session_module = fake_botocore.session
+    fake_session = fake_session_module.Session(sentinel.RETRIEVAL)
+
+    assert botocore_utils._aws_region(dict(region=sentinel.PARAM_REGION, profile=None)) is sentinel.PARAM_REGION
+    assert fake_session_module.Session.call_args == call(sentinel.RETRIEVAL)
+
+    assert utils_botocore._aws_region(dict(profile=None)) is sentinel.BOTO3_REGION
+    assert fake_session_module.Session.call_args == call(profile=None)
+    assert fake_session.get_config_variable.call_args == call('region')
+
+
+def test_aws_region_empty_profile(monkeypatch, botocore_utils, fake_botocore):
+
+    monkeypatch.setattr(botocore_utils, 'botocore', fake_botocore)
+    fake_session_module = fake_botocore.session
+    fake_session = fake_session_module.Session(sentinel.RETRIEVAL)
+
+    assert botocore_utils._aws_region(dict(region=sentinel.PARAM_REGION, profile='')) is sentinel.PARAM_REGION
+    assert fake_session_module.Session.call_args == call(sentinel.RETRIEVAL)
+
+    assert utils_botocore._aws_region(dict(profile='')) is sentinel.BOTO3_REGION
+    assert fake_session_module.Session.call_args == call(profile=None)
+    assert fake_session.get_config_variable.call_args == call('region')
+
+
+def test_aws_region_with_profile(monkeypatch, botocore_utils, fake_botocore):
+
+    monkeypatch.setattr(botocore_utils, 'botocore', fake_botocore)
+    fake_session_module = fake_botocore.session
+    fake_session = fake_session_module.Session(sentinel.RETRIEVAL)
+
+    assert botocore_utils._aws_region(dict(region=sentinel.PARAM_REGION, profile=sentinel.PARAM_PROFILE)) is sentinel.PARAM_REGION
+    assert fake_session_module.Session.call_args == call(sentinel.RETRIEVAL)
+
+    assert utils_botocore._aws_region(dict(profile=sentinel.PARAM_PROFILE)) is sentinel.BOTO3_REGION
+    assert fake_session_module.Session.call_args == call(profile=sentinel.PARAM_PROFILE)
+    assert fake_session.get_config_variable.call_args == call('region')
+
+
+def test_aws_region_bad_profile(monkeypatch, botocore_utils, fake_botocore):
+
+    not_found_exception = botocore.exceptions.ProfileNotFound(profile=sentinel.ERROR_PROFILE)
+
+    monkeypatch.setattr(botocore_utils, 'botocore', fake_botocore)
+    fake_session_module = fake_botocore.session
+
+    assert botocore_utils._aws_region(dict(region=sentinel.PARAM_REGION, profile=sentinel.PARAM_PROFILE)) is sentinel.PARAM_REGION
+    # We've always just returned a blank region if we're passed a bad profile.
+    # However, it's worth noting however that once someone tries to build a connection passing the
+    # bad profile name they'll see the ProfileNotFound exception
+    fake_session_module.Session.side_effect = not_found_exception
+    assert utils_botocore._aws_region(dict(profile=sentinel.PARAM_PROFILE)) is None
+    assert fake_session_module.Session.call_args == call(profile=sentinel.PARAM_PROFILE)