d31ded47fb
* aws ssm parameter lookup test case - fails demonstrating no exception when parameter missing * aws ssm parameter lookup - fail in case parameter doesn't exist * aws ssm parameter lookup test case - failing case for nice return from path lookup * aws ssm parameter lookup - convert incoming taglist to a key-value dictionary * aws ssm parameter lookup - pep8 / style clean up * aws_ssm lookup plugin rewrite for more standard interface * aws_ssm module and lookup - introduce integration test and fix: * aws_ssm module and lookup - error case integraton test and many PEP8 and other cleanups * aws ssm parameter lookup - Various fixes in response to review + recursive fix & test * aws ssm parameter lookup - more in response to review - shertel/abadger * aws ssm parameter lookup unit test - move to mocker according to abadger * aws ssm parameter lookup - integrate with new documentation fragment * aws ssm parameter lookup - accept either aws_profile or boto_profile * aws ssm parameter lookup - eliminate lookup document fragment until env vars are fixed later
137 lines
5.1 KiB
Python
137 lines
5.1 KiB
Python
#
|
|
# (c) 2017 Michael De La Rue
|
|
#
|
|
# This file is part of Ansible
|
|
#
|
|
# Ansible is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU General Public License as published by
|
|
# the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# Ansible is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU General Public License
|
|
# along with Ansible. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
# Make coding more python3-ish
|
|
from __future__ import (absolute_import, division, print_function)
|
|
|
|
import pytest
|
|
from copy import copy
|
|
|
|
from ansible.errors import AnsibleError
|
|
|
|
import ansible.plugins.lookup.aws_ssm as aws_ssm
|
|
|
|
try:
|
|
import boto3
|
|
from botocore.exceptions import ClientError
|
|
except ImportError:
|
|
pytestmark = pytest.mark.skip("This test requires the boto3 and botocore Python libraries")
|
|
|
|
simple_variable_success_response = {
|
|
'Parameters': [
|
|
{
|
|
'Name': 'simple_variable',
|
|
'Type': 'String',
|
|
'Value': 'simplevalue',
|
|
'Version': 1
|
|
}
|
|
],
|
|
'InvalidParameters': [],
|
|
'ResponseMetadata': {
|
|
'RequestId': '12121212-3434-5656-7878-9a9a9a9a9a9a',
|
|
'HTTPStatusCode': 200,
|
|
'HTTPHeaders': {
|
|
'x-amzn-requestid': '12121212-3434-5656-7878-9a9a9a9a9a9a',
|
|
'content-type': 'application/x-amz-json-1.1',
|
|
'content-length': '116',
|
|
'date': 'Tue, 23 Jan 2018 11:04:27 GMT'
|
|
},
|
|
'RetryAttempts': 0
|
|
}
|
|
}
|
|
|
|
path_success_response = copy(simple_variable_success_response)
|
|
path_success_response['Parameters'] = [
|
|
{'Name': '/testpath/too', 'Type': 'String', 'Value': 'simple_value_too', 'Version': 1},
|
|
{'Name': '/testpath/won', 'Type': 'String', 'Value': 'simple_value_won', 'Version': 1}
|
|
]
|
|
|
|
missing_variable_fail_response = copy(simple_variable_success_response)
|
|
missing_variable_fail_response['Parameters'] = []
|
|
missing_variable_fail_response['InvalidParameters'] = ['missing_variable']
|
|
|
|
|
|
dummy_credentials = {}
|
|
dummy_credentials['boto_profile'] = None
|
|
dummy_credentials['aws_secret_key'] = "notasecret"
|
|
dummy_credentials['aws_access_key'] = "notakey"
|
|
dummy_credentials['aws_security_token'] = None
|
|
dummy_credentials['region'] = 'eu-west-1'
|
|
|
|
|
|
def test_lookup_variable(mocker):
|
|
lookup = aws_ssm.LookupModule()
|
|
lookup._load_name = "aws_ssm"
|
|
|
|
boto3_double = mocker.MagicMock()
|
|
boto3_double.Session.return_value.client.return_value.get_parameters.return_value = simple_variable_success_response
|
|
boto3_client_double = boto3_double.Session.return_value.client
|
|
|
|
with mocker.patch.object(boto3, 'session', boto3_double):
|
|
retval = lookup.run(["simple_variable"], {}, **dummy_credentials)
|
|
assert(retval[0] == "simplevalue")
|
|
boto3_client_double.assert_called_with('ssm', 'eu-west-1', aws_access_key_id='notakey',
|
|
aws_secret_access_key="notasecret", aws_session_token=None)
|
|
|
|
|
|
def test_path_lookup_variable(mocker):
|
|
lookup = aws_ssm.LookupModule()
|
|
lookup._load_name = "aws_ssm"
|
|
|
|
boto3_double = mocker.MagicMock()
|
|
get_path_fn = boto3_double.Session.return_value.client.return_value.get_parameters_by_path
|
|
get_path_fn.return_value = path_success_response
|
|
boto3_client_double = boto3_double.Session.return_value.client
|
|
|
|
with mocker.patch.object(boto3, 'session', boto3_double):
|
|
args = copy(dummy_credentials)
|
|
args["bypath"] = 'true'
|
|
retval = lookup.run(["/testpath"], {}, **args)
|
|
assert(retval[0]["/testpath/won"] == "simple_value_won")
|
|
assert(retval[0]["/testpath/too"] == "simple_value_too")
|
|
boto3_client_double.assert_called_with('ssm', 'eu-west-1', aws_access_key_id='notakey',
|
|
aws_secret_access_key="notasecret", aws_session_token=None)
|
|
get_path_fn.assert_called_with(Path="/testpath", Recursive=False, WithDecryption=True)
|
|
|
|
|
|
def test_warn_missing_variable(mocker):
|
|
lookup = aws_ssm.LookupModule()
|
|
lookup._load_name = "aws_ssm"
|
|
|
|
boto3_double = mocker.MagicMock()
|
|
boto3_double.Session.return_value.client.return_value.get_parameters.return_value = missing_variable_fail_response
|
|
|
|
with pytest.raises(AnsibleError):
|
|
with mocker.patch.object(boto3, 'session', boto3_double):
|
|
lookup.run(["missing_variable"], {}, **dummy_credentials)
|
|
|
|
|
|
error_response = {'Error': {'Code': 'ResourceNotFoundException', 'Message': 'Fake Testing Error'}}
|
|
operation_name = 'FakeOperation'
|
|
|
|
|
|
def test_warn_denied_variable(mocker):
|
|
lookup = aws_ssm.LookupModule()
|
|
lookup._load_name = "aws_ssm"
|
|
|
|
boto3_double = mocker.MagicMock()
|
|
boto3_double.Session.return_value.client.return_value.get_parameters.side_effect = ClientError(error_response, operation_name)
|
|
|
|
with pytest.raises(AnsibleError):
|
|
with mocker.patch.object(boto3, 'session', boto3_double):
|
|
lookup.run(["denied_variable"], {}, **dummy_credentials)
|