Skip to content

Commit

Permalink
Update tests for Pytest support
Browse files Browse the repository at this point in the history
  • Loading branch information
nateprewitt committed Sep 23, 2021
1 parent 3a91c14 commit 85364f0
Show file tree
Hide file tree
Showing 5 changed files with 122 additions and 70 deletions.
7 changes: 3 additions & 4 deletions scripts/ci/run-tests
Original file line number Diff line number Diff line change
Expand Up @@ -34,8 +34,7 @@ def process_args(args):
test_args = ""
if args.with_cov:
test_args += (
f"--with-xunit --cover-erase --with-coverage "
f"--cover-package {PACKAGE} --cover-xml -v "
f"--cov={PACKAGE} --cov-report xml -v "
)
dirs = " ".join(args.test_dirs)

Expand All @@ -53,8 +52,8 @@ if __name__ == "__main__":
parser.add_argument(
"-r",
"--test-runner",
default="nosetests",
help="Test runner to execute tests. Defaults to nose.",
default="pytest",
help="Test runner to execute tests. Defaults to pytest.",
)
parser.add_argument(
"-c",
Expand Down
90 changes: 58 additions & 32 deletions tests/functional/docs/test_smoke.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,45 +19,71 @@
from boto3.docs.service import ServiceDocumenter


def test_docs_generated():
"""Verify we can generate the appropriate docs for all services"""
@pytest.fixture
def botocore_session():
return botocore.session.get_session()

@pytest.fixture
def boto3_session():
return boto3.Session(region_name='us-east-1')

def all_services():
botocore_session = botocore.session.get_session()
session = boto3.Session(region_name='us-east-1')
for service_name in session.get_available_services():
generated_docs = ServiceDocumenter(
service_name, session=session).document_service()
generated_docs = generated_docs.decode('utf-8')
client = boto3.client(service_name, 'us-east-1')
yield service_name


@pytest.fixture
def available_resources():
session = boto3.Session(region_name='us-east-1')
return session.get_available_resources()


@pytest.mark.parametrize('service_name', all_services())
def test_documentation(
boto3_session, botocore_session, available_resources, service_name
):
generated_docs = ServiceDocumenter(
service_name, session=boto3_session).document_service()
generated_docs = generated_docs.decode('utf-8')
client = boto3.client(service_name, 'us-east-1')

# Check that all of the services have the appropriate title
_assert_has_title(generated_docs, client)


# Check that all services have the client documented.
_assert_has_client_documentation(generated_docs, service_name, client)

# Check that all of the services have the appropriate title
yield (_assert_has_title, generated_docs, client)

# Check that all services have the client documented.
yield (_assert_has_client_documentation, generated_docs, service_name,
client)
#If the service has resources, make sure the service resource
#is at least documented.
if service_name in available_resources:

resource = boto3.resource(service_name, 'us-east-1')
_assert_has_resource_documentation(
generated_docs, service_name, resource
)

# If the client can paginate, make sure the paginators are documented.
try:
paginator_model = botocore_session.get_paginator_model(
# If the client can paginate, make sure the paginators are documented.
try:
paginator_model = botocore_session.get_paginator_model(
service_name)
yield (_assert_has_paginator_documentation, generated_docs,
service_name, client,
sorted(paginator_model._paginator_config))
except DataNotFoundError:
pass

# If the client has waiters, make sure the waiters are documented
if client.waiter_names:
waiter_model = botocore_session.get_waiter_model(service_name)
yield (_assert_has_waiter_documentation, generated_docs,
service_name, client, waiter_model)

# If the service has resources, make sure the service resource
# is at least documented.
if service_name in session.get_available_resources():
resource = boto3.resource(service_name, 'us-east-1')
yield (_assert_has_resource_documentation, generated_docs,
service_name, resource)
_assert_has_paginator_documentation(
generated_docs, service_name, client,
sorted(paginator_model._paginator_config)
)
except DataNotFoundError:
pass


# If the client has waiters, make sure the waiters are documented.
if client.waiter_names:
waiter_model = botocore_session.get_waiter_model(service_name)
_assert_has_waiter_documentation(
generated_docs, service_name, client, waiter_model
)


def _assert_contains_lines_in_order(lines, contents):
Expand Down
42 changes: 29 additions & 13 deletions tests/functional/test_smoke.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,11 +23,29 @@ def create_session():
return session


def test_can_create_all_resources():
"""Verify we can create all existing resources."""
def _all_resources():
session = create_session()
for service_name in session.get_available_resources():
yield _test_create_resource, session, service_name
yield session, service_name


def _all_clients():
session = create_session()
for service_name in session.get_available_services():
yield session, service_name


def _all_api_version_args():
botocore_session = botocore.session.get_session()
boto3_session = create_session()
for service_name in boto3_session.get_available_resources():
yield (service_name, botocore_session, boto3_session)


@pytest.mark.parametrize('resource_args', _all_resources())
def test_can_create_all_resources(all_resources):
"""Verify we can create all existing resources."""
_test_create_resource(*resource_args)


def _test_create_resource(session, service_name):
Expand All @@ -37,23 +55,21 @@ def _test_create_resource(session, service_name):
assert hasattr(resource, 'meta')


def test_can_create_all_clients():
session = create_session()
for service_name in session.get_available_services():
yield _test_create_client, session, service_name
@pytest.mark.parametrize('client_args', _all_clients())
def test_can_create_all_resources(client_args):
"""Verify we can create all existing clients."""
_test_create_client(*client_args)


def _test_create_client(session, service_name):
client = session.client(service_name)
assert hasattr(client, 'meta')


def test_api_versions_synced_with_botocore():
botocore_session = botocore.session.get_session()
boto3_session = create_session()
for service_name in boto3_session.get_available_resources():
yield (_assert_same_api_versions, service_name,
botocore_session, boto3_session)
@pytest.mark.parametrize('api_version_args', _all_api_version_args())
def test_api_versions_synced_with_botocore(api_version_args):
"""Verify both boto3 and botocore clients stay in sync."""
_assert_same_api_versions(*api_version_args)


def _assert_same_api_versions(service_name, botocore_session, boto3_session):
Expand Down
26 changes: 16 additions & 10 deletions tests/integration/test_collections.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,9 +11,11 @@
# ANY KIND, either express or implied. See the License for the specific
# language governing permissions and limitations under the License.

import boto3.session
import pytest

import boto3.session
from boto3.resources.collection import CollectionManager
from boto3.resources.base import ServiceResource


# A map of services to regions that cannot use us-west-2
Expand All @@ -24,33 +26,37 @@

# A list of collections to ignore. They require parameters
# or are very slow to run.
BLACKLIST = {
BLOCKLIST = {
'ec2': ['images'],
'iam': ['signing_certificates'],
'sqs': ['dead_letter_source_queues']
}


def test_all_collections():
# This generator yields test functions for every collection
# on every available resource, except those which have
# been blacklisted.
def all_collections():
# This generator yields every collection on every available resource,
# except those which have been blocklisted.
session = boto3.session.Session()
for service_name in session.get_available_resources():
resource = session.resource(
service_name,
region_name=REGION_MAP.get(service_name, 'us-west-2'))

for key in dir(resource):
if key in BLACKLIST.get(service_name, []):
if key in BLOCKLIST.get(service_name, []):
continue

value = getattr(resource, key)
if isinstance(value, CollectionManager):
yield _test_collection, service_name, key, value
yield value


def _test_collection(service_name, collection_name, collection):
@pytest.mark.parametrize("collection", all_collections())
def test_all_collections(collection):
"""Test all collections work on every available resource."""
# Create a list of the first page of items. This tests that
# a remote request can be made, the response parsed, and that
# resources are successfully created.
list(collection.limit(1))
collection_list = list(collection.limit(1))
assert len(collection_list) < 2
assert all([isinstance(res, ServiceResource) for res in collection_list])
27 changes: 16 additions & 11 deletions tests/unit/resources/test_collection_smoke.py
Original file line number Diff line number Diff line change
Expand Up @@ -63,13 +63,7 @@ def _shape_has_pagination_param(shape):
return False


def test_all_collections_have_paginators_if_needed():
# If a collection relies on an operation that is paginated, it
# will require a paginator to iterate through all of the resources
# with the all() method. If there is no paginator, it will only
# make it through the first page of results. So we need to make sure
# if a collection looks like it uses a paginated operation then there
# should be a paginator applied to it.
def _collection_test_args():
botocore_session = botocore.session.get_session()
session = Session(botocore_session=botocore_session)
loader = botocore_session.get_component('data_loader')
Expand All @@ -91,13 +85,24 @@ def test_all_collections_have_paginators_if_needed():
# Iterate over all of the collections for each resource model
# and ensure that the collection has a paginator if it needs one.
for collection_model in resource_model.collections:
yield (
_assert_collection_has_paginator_if_needed, client,
service_name, resource_name, collection_model)
yield (client, service_name, resource_name, collection_model)

@pytest.mark.parametrize(
'collection_args',
_collection_test_args()
)
def test_all_collections_have_paginators_if_needed(collection_args):
# If a collection relies on an operation that is paginated, it
# will require a paginator to iterate through all of the resources
# with the all() method. If there is no paginator, it will only
# make it through the first page of results. So we need to make sure
# if a collection looks like it uses a paginated operation then there
# should be a paginator applied to it.
_assert_collection_has_paginator_if_needed(*collection_args)

def _assert_collection_has_paginator_if_needed(
client, service_name, resource_name, collection_model):
client, service_name, resource_name, collection_model
):
underlying_operation_name = collection_model.request.operation
# See if the operation can be paginated from the client.
can_paginate_operation = client.can_paginate(
Expand Down

0 comments on commit 85364f0

Please sign in to comment.