mirror of
https://github.com/kevin1024/vcrpy.git
synced 2025-12-09 01:03:24 +00:00
refactor boto3 integration test using fixtures instead of global variables
This commit is contained in:
@@ -7,18 +7,6 @@ import boto3 # NOQA
|
|||||||
import botocore # NOQA
|
import botocore # NOQA
|
||||||
import vcr # NOQA
|
import vcr # NOQA
|
||||||
|
|
||||||
ses = boto3.Session(
|
|
||||||
aws_access_key_id=os.environ.get('AWS_ACCESS_KEY_ID', "default"),
|
|
||||||
aws_secret_access_key=os.environ.get('AWS_SECRET_ACCESS_KEY', "default"),
|
|
||||||
aws_session_token=None,
|
|
||||||
region_name=os.environ.get('AWS_DEFAULT_REGION', "default"),
|
|
||||||
# botocore_session=None,
|
|
||||||
# profile_name=None
|
|
||||||
)
|
|
||||||
|
|
||||||
IAM_CLIENT = ses.client('iam')
|
|
||||||
IAM_USER_NAME = 'vcrpy'
|
|
||||||
|
|
||||||
try:
|
try:
|
||||||
from botocore import awsrequest # NOQA
|
from botocore import awsrequest # NOQA
|
||||||
|
|
||||||
@@ -26,7 +14,6 @@ try:
|
|||||||
except ImportError:
|
except ImportError:
|
||||||
botocore_awsrequest = False
|
botocore_awsrequest = False
|
||||||
|
|
||||||
|
|
||||||
# skip tests if boto does not use vendored requests anymore
|
# skip tests if boto does not use vendored requests anymore
|
||||||
# https://github.com/boto/botocore/pull/1495
|
# https://github.com/boto/botocore/pull/1495
|
||||||
boto3_skip_vendored_requests = pytest.mark.skipif(
|
boto3_skip_vendored_requests = pytest.mark.skipif(
|
||||||
@@ -40,6 +27,33 @@ boto3_skip_awsrequest = pytest.mark.skipif(
|
|||||||
ver=botocore.__version__))
|
ver=botocore.__version__))
|
||||||
|
|
||||||
|
|
||||||
|
IAM_USER_NAME = "vcrpy"
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def iam_client():
|
||||||
|
def _iam_client(boto3_session=None):
|
||||||
|
if boto3_session is None:
|
||||||
|
boto3_session = boto3.Session(
|
||||||
|
aws_access_key_id=os.environ.get('AWS_ACCESS_KEY_ID', "default"),
|
||||||
|
aws_secret_access_key=os.environ.get('AWS_SECRET_ACCESS_KEY', "default"),
|
||||||
|
aws_session_token=None,
|
||||||
|
region_name=os.environ.get('AWS_DEFAULT_REGION', "default"),
|
||||||
|
)
|
||||||
|
return boto3_session.client('iam')
|
||||||
|
return _iam_client
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def get_user(iam_client):
|
||||||
|
def _get_user(client=None, user_name=IAM_USER_NAME):
|
||||||
|
if client is None:
|
||||||
|
# Default client set with fixture `iam_client`
|
||||||
|
client = iam_client()
|
||||||
|
return client.get_user(UserName=user_name)
|
||||||
|
return _get_user
|
||||||
|
|
||||||
|
|
||||||
@boto3_skip_vendored_requests
|
@boto3_skip_vendored_requests
|
||||||
def test_boto_vendored_stubs(tmpdir):
|
def test_boto_vendored_stubs(tmpdir):
|
||||||
with vcr.use_cassette(str(tmpdir.join('boto3-stubs.yml'))):
|
with vcr.use_cassette(str(tmpdir.join('boto3-stubs.yml'))):
|
||||||
@@ -55,45 +69,32 @@ def test_boto_vendored_stubs(tmpdir):
|
|||||||
VerifiedHTTPSConnection('hostname.does.not.matter')
|
VerifiedHTTPSConnection('hostname.does.not.matter')
|
||||||
|
|
||||||
|
|
||||||
@boto3_skip_awsrequest
|
def test_boto3_without_vcr(get_user):
|
||||||
def test_boto3_awsrequest_stubs(tmpdir):
|
response = get_user()
|
||||||
with vcr.use_cassette(str(tmpdir.join('boto3-stubs.yml'))):
|
|
||||||
from botocore.awsrequest import AWSHTTPConnection, AWSHTTPSConnection
|
|
||||||
from vcr.stubs.boto3_stubs import VCRRequestsHTTPConnection, VCRRequestsHTTPSConnection
|
|
||||||
assert issubclass(VCRRequestsHTTPConnection, AWSHTTPConnection)
|
|
||||||
assert issubclass(VCRRequestsHTTPSConnection, AWSHTTPSConnection)
|
|
||||||
AWSHTTPConnection('hostname.does.not.matter')
|
|
||||||
AWSHTTPSConnection('hostname.does.not.matter')
|
|
||||||
|
|
||||||
|
|
||||||
def test_boto3_without_vcr():
|
|
||||||
response = IAM_CLIENT.get_user(UserName=IAM_USER_NAME)
|
|
||||||
|
|
||||||
assert response['User']['UserName'] == IAM_USER_NAME
|
assert response['User']['UserName'] == IAM_USER_NAME
|
||||||
|
|
||||||
|
|
||||||
def test_boto_medium_difficulty(tmpdir):
|
def test_boto_medium_difficulty(tmpdir, get_user):
|
||||||
|
|
||||||
with vcr.use_cassette(str(tmpdir.join('boto3-medium.yml'))):
|
with vcr.use_cassette(str(tmpdir.join('boto3-medium.yml'))):
|
||||||
response = IAM_CLIENT.get_user(UserName=IAM_USER_NAME)
|
response = get_user()
|
||||||
assert response['User']['UserName'] == IAM_USER_NAME
|
assert response['User']['UserName'] == IAM_USER_NAME
|
||||||
|
|
||||||
with vcr.use_cassette(str(tmpdir.join('boto3-medium.yml'))) as cass:
|
with vcr.use_cassette(str(tmpdir.join('boto3-medium.yml'))) as cass:
|
||||||
response = IAM_CLIENT.get_user(UserName=IAM_USER_NAME)
|
response = get_user()
|
||||||
assert response['User']['UserName'] == IAM_USER_NAME
|
assert response['User']['UserName'] == IAM_USER_NAME
|
||||||
assert cass.all_played
|
assert cass.all_played
|
||||||
|
|
||||||
|
|
||||||
def test_boto_hardcore_mode(tmpdir):
|
def test_boto_hardcore_mode(tmpdir, iam_client, get_user):
|
||||||
with vcr.use_cassette(str(tmpdir.join('boto3-hardcore.yml'))):
|
with vcr.use_cassette(str(tmpdir.join('boto3-hardcore.yml'))):
|
||||||
ses = boto3.Session(
|
ses = boto3.Session(
|
||||||
aws_access_key_id=os.environ.get('AWS_ACCESS_KEY_ID'),
|
aws_access_key_id=os.environ.get('AWS_ACCESS_KEY_ID'),
|
||||||
aws_secret_access_key=os.environ.get('AWS_SECRET_ACCESS_KEY'),
|
aws_secret_access_key=os.environ.get('AWS_SECRET_ACCESS_KEY'),
|
||||||
region_name=os.environ.get('AWS_DEFAULT_REGION'),
|
region_name=os.environ.get('AWS_DEFAULT_REGION'),
|
||||||
)
|
)
|
||||||
|
client = iam_client(ses)
|
||||||
iam_client = ses.client('iam')
|
response = get_user(client=client)
|
||||||
response = iam_client.get_user(UserName=IAM_USER_NAME)
|
|
||||||
assert response['User']['UserName'] == IAM_USER_NAME
|
assert response['User']['UserName'] == IAM_USER_NAME
|
||||||
|
|
||||||
with vcr.use_cassette(str(tmpdir.join('boto3-hardcore.yml'))) as cass:
|
with vcr.use_cassette(str(tmpdir.join('boto3-hardcore.yml'))) as cass:
|
||||||
@@ -104,7 +105,7 @@ def test_boto_hardcore_mode(tmpdir):
|
|||||||
region_name=os.environ.get('AWS_DEFAULT_REGION'),
|
region_name=os.environ.get('AWS_DEFAULT_REGION'),
|
||||||
)
|
)
|
||||||
|
|
||||||
iam_client = ses.client('iam')
|
client = iam_client(ses)
|
||||||
response = iam_client.get_user(UserName=IAM_USER_NAME)
|
response = get_user(client=client)
|
||||||
assert response['User']['UserName'] == IAM_USER_NAME
|
assert response['User']['UserName'] == IAM_USER_NAME
|
||||||
assert cass.all_played
|
assert cass.all_played
|
||||||
|
|||||||
@@ -41,4 +41,3 @@ class VCRRequestsHTTPSConnection(VCRHTTPSConnection, VerifiedHTTPSConnection):
|
|||||||
self.real_connection.cert_reqs = kwargs.get("cert_reqs", 'CERT_NONE')
|
self.real_connection.cert_reqs = kwargs.get("cert_reqs", 'CERT_NONE')
|
||||||
|
|
||||||
self._sock = None
|
self._sock = None
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user