diff options
Diffstat (limited to 'ansible_collections/amazon/aws/tests/unit/utils')
| -rw-r--r-- | ansible_collections/amazon/aws/tests/unit/utils/amazon_placebo_fixtures.py | 213 |
1 files changed, 213 insertions, 0 deletions
diff --git a/ansible_collections/amazon/aws/tests/unit/utils/amazon_placebo_fixtures.py b/ansible_collections/amazon/aws/tests/unit/utils/amazon_placebo_fixtures.py new file mode 100644 index 000000000..6912c2e32 --- /dev/null +++ b/ansible_collections/amazon/aws/tests/unit/utils/amazon_placebo_fixtures.py @@ -0,0 +1,213 @@ +from __future__ import absolute_import, division, print_function +__metaclass__ = type + +import errno +import os +import time +import mock +import pytest + +boto3 = pytest.importorskip("boto3") +botocore = pytest.importorskip("botocore") +placebo = pytest.importorskip("placebo") + +""" +Using Placebo to test modules using boto3: + +This is an example test, using the placeboify fixture to test that a module +will fail if resources it depends on don't exist. + +> from placebo_fixtures import placeboify, scratch_vpc +> +> def test_create_with_nonexistent_launch_config(placeboify): +> connection = placeboify.client('autoscaling') +> module = FakeModule('test-asg-created', None, min_size=0, max_size=0, desired_capacity=0) +> with pytest.raises(FailJSON) as excinfo: +> asg_module.create_autoscaling_group(connection, module) +> .... asserts based on module state/exceptions .... + +In more advanced cases, use unrecorded resource fixtures to fill in ARNs/IDs of +things modules depend on, such as: + +> def test_create_in_vpc(placeboify, scratch_vpc): +> connection = placeboify.client('autoscaling') +> module = FakeModule(name='test-asg-created', +> min_size=0, max_size=0, desired_capacity=0, +> availability_zones=[s['az'] for s in scratch_vpc['subnets']], +> vpc_zone_identifier=[s['id'] for s in scratch_vpc['subnets']], +> ) +> ..... so on and so forth .... +""" + + +@pytest.fixture +def placeboify(request, monkeypatch): + """This fixture puts a recording/replaying harness around `boto3_conn` + + Placeboify patches the `boto3_conn` function in ec2 module_utils to return + a boto3 session that in recording or replaying mode, depending on the + PLACEBO_RECORD environment variable. Unset PLACEBO_RECORD (the common case + for just running tests) will put placebo in replay mode, set PLACEBO_RECORD + to any value to turn off replay & operate on real AWS resources. + + The recorded sessions are stored in the test file's directory, under the + namespace `placebo_recordings/{testfile name}/{test function name}` to + distinguish them. + """ + session = boto3.Session(region_name='us-west-2') + + recordings_path = os.path.join( + request.fspath.dirname, + 'placebo_recordings', + request.fspath.basename.replace('.py', ''), + request.function.__name__ + # remove the test_ prefix from the function & file name + ).replace('test_', '') + + if not os.getenv('PLACEBO_RECORD'): + if not os.path.isdir(recordings_path): + raise NotImplementedError('Missing Placebo recordings in directory: %s' % recordings_path) + else: + try: + # make sure the directory for placebo test recordings is available + os.makedirs(recordings_path) + except OSError as e: + if e.errno != errno.EEXIST: + raise + + pill = placebo.attach(session, data_path=recordings_path) + if os.getenv('PLACEBO_RECORD'): + pill.record() + else: + pill.playback() + + def boto3_middleman_connection(module, conn_type, resource, region='us-west-2', **kwargs): + if conn_type != 'client': + # TODO support resource-based connections + raise ValueError('Mocker only supports client, not %s' % conn_type) + return session.client(resource, region_name=region) + + import ansible_collections.amazon.aws.plugins.module_utils.ec2 + monkeypatch.setattr( + ansible_collections.amazon.aws.plugins.module_utils.ec2, + 'boto3_conn', + boto3_middleman_connection, + ) + yield session + + # tear down + pill.stop() + + +@pytest.fixture(scope='module') +def basic_launch_config(): + """Create an EC2 launch config whose creation *is not* recorded and return its name + + This fixture is module-scoped, since launch configs are immutable and this + can be reused for many tests. + """ + if not os.getenv('PLACEBO_RECORD'): + yield 'pytest_basic_lc' + return + + # use a *non recording* session to make the launch config + # since that's a prereq of the ec2_asg module, and isn't what + # we're testing. + asg = boto3.client('autoscaling') + asg.create_launch_configuration( + LaunchConfigurationName='pytest_basic_lc', + ImageId='ami-9be6f38c', # Amazon Linux 2016.09 us-east-1 AMI, can be any valid AMI + SecurityGroups=[], + UserData='#!/bin/bash\necho hello world', + InstanceType='t2.micro', + InstanceMonitoring={'Enabled': False}, + AssociatePublicIpAddress=True + ) + + yield 'pytest_basic_lc' + + try: + asg.delete_launch_configuration(LaunchConfigurationName='pytest_basic_lc') + except botocore.exceptions.ClientError as e: + if 'not found' in e.message: + return + raise + + +@pytest.fixture(scope='module') +def scratch_vpc(): + if not os.getenv('PLACEBO_RECORD'): + yield { + 'vpc_id': 'vpc-123456', + 'cidr_range': '10.0.0.0/16', + 'subnets': [ + { + 'id': 'subnet-123456', + 'az': 'us-east-1d', + }, + { + 'id': 'subnet-654321', + 'az': 'us-east-1e', + }, + ] + } + return + + # use a *non recording* session to make the base VPC and subnets + ec2 = boto3.client('ec2') + vpc_resp = ec2.create_vpc( + CidrBlock='10.0.0.0/16', + AmazonProvidedIpv6CidrBlock=False, + ) + subnets = ( + ec2.create_subnet( + VpcId=vpc_resp['Vpc']['VpcId'], + CidrBlock='10.0.0.0/24', + ), + ec2.create_subnet( + VpcId=vpc_resp['Vpc']['VpcId'], + CidrBlock='10.0.1.0/24', + ) + ) + time.sleep(3) + + yield { + 'vpc_id': vpc_resp['Vpc']['VpcId'], + 'cidr_range': '10.0.0.0/16', + 'subnets': [ + { + 'id': s['Subnet']['SubnetId'], + 'az': s['Subnet']['AvailabilityZone'], + } for s in subnets + ] + } + + try: + for s in subnets: + try: + ec2.delete_subnet(SubnetId=s['Subnet']['SubnetId']) + except botocore.exceptions.ClientError as e: + if 'not found' in e.message: + continue + raise + ec2.delete_vpc(VpcId=vpc_resp['Vpc']['VpcId']) + except botocore.exceptions.ClientError as e: + if 'not found' in e.message: + return + raise + + +@pytest.fixture(scope='module') +def maybe_sleep(): + """If placebo is reading saved sessions, make sleep always take 0 seconds. + + AWS modules often perform polling or retries, but when using recorded + sessions there's no reason to wait. We can still exercise retry and other + code paths without waiting for wall-clock time to pass.""" + if not os.getenv('PLACEBO_RECORD'): + p = mock.patch('time.sleep', return_value=None) + p.start() + yield + p.stop() + else: + yield |