You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
ansible/test/units/modules/cloud/amazon/placebo_fixtures.py

207 lines
6.5 KiB
Python

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_', '')
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.module_utils.ec2
monkeypatch.setattr(
ansible.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