Test mem cleanup

tests/conftest.py

@@ -19,6 +19,14 @@
 from prompt_toolkit.output import DummyOutput
 
 import awscli.logger
+from awscli.clidriver import create_clidriver
+
+
+@pytest.fixture
+def cli_driver():
+    # Each model-validation command gets its own driver so previously checked
+    # command tables and models do not accumulate in a module-wide cache.
+    return create_clidriver()
 
 
 @pytest.fixture(autouse=True)
@@ -33,7 +41,11 @@ def clear_loggers():
     for logger_name in loggers:
         logger = logging.getLogger(logger_name)
         logger.handlers = []
-        logger.setLevel(logging.NOTSET)
+        if logger.level != logging.NOTSET:
+            logger.setLevel(logging.NOTSET)
+    # setLevel clears every logger's effective-level cache. Do this once even
+    # when no levels changed, rather than once per already-reset logger.
+    logging.root.setLevel(logging.root.level)
     awscli.logger.disable_crt_logging()
 
 

tests/functional/autocomplete/test_completion_files.py

@@ -90,65 +90,60 @@ class ServiceTestData:
     service_name: str
 
 
-def get_models_with_completions():
+def get_services_with_completions():
     session = clidriver.create_clidriver().session
     loader = session.get_component('data_loader')
-    services_with_completions = loader.list_available_services('completions-1')
-    models = []
-    for service_name in services_with_completions:
-        service_model = loader.load_service_model(service_name, 'service-2')
-        api_version = service_model['metadata']['apiVersion']
-        completions = loader.load_service_model(
-            service_name, 'completions-1', api_version
-        )
-        models.append(
-            ServiceTestData(service_model, completions, service_name)
-        )
-    return models
-
-
-def _get_invalid_completion_operations():
+    return loader.list_available_services('completions-1')
+
+
+@pytest.fixture(scope='module', params=get_services_with_completions())
+def service_test_data(request):
+    # Only the current service's documents live in this fixture; collection
+    # retains service names, not another copy of every service model.
+    service_name = request.param
+    session = clidriver.create_clidriver().session
+    loader = session.get_component('data_loader')
+    service_model = loader.load_service_model(service_name, 'service-2')
+    api_version = service_model['metadata']['apiVersion']
+    completions = loader.load_service_model(
+        service_name, 'completions-1', api_version
+    )
+    return ServiceTestData(service_model, completions, service_name)
+
+
+def _get_invalid_completion_operations(service_test_data):
     cases = []
-    for test_data in get_models_with_completions():
-        known_ops = set(test_data.service_model['operations'])
-        for op_name in test_data.completions['operations']:
-            if op_name not in known_ops:
-                cases.append((test_data.service_name, op_name))
-        for resource in test_data.completions.get('resources', {}).values():
-            op_name = resource.get('operation')
-            if op_name and op_name not in known_ops:
-                cases.append((test_data.service_name, op_name))
+    known_ops = set(service_test_data.service_model['operations'])
+    for op_name in service_test_data.completions['operations']:
+        if op_name not in known_ops:
+            cases.append(op_name)
+    resources = service_test_data.completions.get('resources', {})
+    for resource in resources.values():
+        op_name = resource.get('operation')
+        if op_name and op_name not in known_ops:
+            cases.append(op_name)
     return cases
 
 
 @pytest.mark.validates_models
-@pytest.mark.parametrize(
-    "service_name, operation_name",
-    _get_invalid_completion_operations(),
-    ids=lambda val: str(val),
-)
 def test_completions_operations_exist_in_model(
-    service_name, operation_name, record_property
+    service_test_data, record_property
 ):
-    record_property('aws_service', service_name)
-    record_property('aws_operation', operation_name)
-    pytest.fail(
-        f"Completions file for '{service_name}' references operation "
-        f"'{operation_name}' which does not exist in the service model. "
-        f"The completions-1.json file must be updated to remove or "
-        f"update references to this operation."
+    invalid_operations = _get_invalid_completion_operations(service_test_data)
+    record_property('aws_service', service_test_data.service_name)
+    for operation_name in invalid_operations:
+        record_property('aws_operation', operation_name)
+    assert not invalid_operations, (
+        f"Completions file for '{service_test_data.service_name}' references "
+        f"operations absent from the service model: {invalid_operations}. "
+        "The completions-1.json file must remove or update these references."
     )
 
 
 @pytest.mark.validates_models
-@pytest.mark.parametrize(
-    "test_data",
-    get_models_with_completions(),
-    ids=lambda test_data: test_data.service_name,
-)
-def test_verify_generated_completions_are_valid(test_data):
-    completions = test_data.completions
-    service_model = test_data.service_model
+def test_verify_generated_completions_are_valid(service_test_data):
+    completions = service_test_data.completions
+    service_model = service_test_data.service_model
     _validate_schema(completions)
     # Validate that every operation named in the completions
     # file references a known operation.
@@ -256,11 +251,11 @@ def _validate_jmespath_expression(
     arg_gen = ArgumentGenerator(use_member_names=True)
     sample_output = arg_gen.generate_skeleton(output_shape)
     search_result = jmespath.search(jmespath_expr, sample_output)
-    assert (
-        search_result is not None
-    ), "Expression is blob or another unsupported type"
+    assert search_result is not None, (
+        "Expression is blob or another unsupported type"
+    )
     assert search_result, f"Expression is broken: {jmespath_expr}"
     sample_arg = search_result[0]
-    assert isinstance(
-        sample_arg, str
-    ), f"Expression not a string: {jmespath_expr}"
+    assert isinstance(sample_arg, str), (
+        f"Expression not a string: {jmespath_expr}"
+    )

tests/functional/botocore/configured_endpoint_urls/test_configured_endpoint_url.py

@@ -21,7 +21,7 @@
 from botocore.compat import urlsplit
 from botocore.config import Config
 
-from tests import ClientHTTPStubber
+from tests import ALL_SERVICES, ClientHTTPStubber, create_session
 
 ENDPOINT_TESTDATA_FILE = Path(__file__).parent / "profile-tests.json"
 
@@ -124,7 +124,7 @@ def _do_create_client(
             f.write(config_file_contents)
             f.flush()
 
-        return botocore.session.Session(profile=profile).create_client(
+        return create_session(profile=profile).create_client(
             service, **client_args
         )
 
@@ -159,15 +159,9 @@ def assert_endpoint_url_used_for_operation(
 
 
 def _known_service_names_and_ids():
-    my_session = botocore.session.get_session()
-    loader = my_session.get_component('data_loader')
-    available_services = loader.list_available_services('service-2')
-
-    result = []
-    for service_name in available_services:
-        model = my_session.get_service_model(service_name)
-        result.append((model.service_name, model.service_id))
-    return sorted(result)
+    return sorted(
+        (model.service_name, model.service_id) for model in ALL_SERVICES
+    )
 
 
 SERVICE_TO_OPERATION = {'s3': 'list_buckets', 'dynamodb': 'list_tables'}

tests/functional/botocore/docs/test_streaming_body.py

@@ -13,30 +13,36 @@
 from botocore import xform_name
 from botocore.docs.service import ServiceDocumenter
 
+from tests import ALL_SERVICES
 from tests.functional.botocore.docs import BaseDocsFunctionalTest
 
 
 class TestStreamingBodyDocumentation(BaseDocsFunctionalTest):
     def test_all_streaming_body_are_properly_documented(self):
-        for service in self._session.get_available_services():
-            client = self._session.create_client(
-                service,
-                region_name='us-east-1',
-                aws_access_key_id='foo',
-                aws_secret_access_key='bar',
-            )
-            service_model = client.meta.service_model
-            for operation in service_model.operation_names:
-                operation_model = service_model.operation_model(operation)
-                if operation_model.has_streaming_output:
-                    self.assert_streaming_body_is_properly_documented(
-                        service, xform_name(operation)
-                    )
+        for service_model in ALL_SERVICES:
+            operations = [
+                xform_name(operation)
+                for operation in service_model.operation_names
+                if service_model.operation_model(
+                    operation
+                ).has_streaming_output
+            ]
+            if not operations:
+                continue
+            service = service_model.service_name
+            # Generate the complete service once, not once per streaming
+            # operation. Each operation's documentation is still checked.
+            ServiceDocumenter(
+                service, self._session, self.root_services_path
+            ).document_service()
+            for operation in operations:
+                self.assert_streaming_body_is_properly_documented(
+                    service, operation
+                )
 
-    def assert_streaming_body_is_properly_documented(self, service, operation):
-        ServiceDocumenter(
-            service, self._session, self.root_services_path
-        ).document_service()
+    def assert_streaming_body_is_properly_documented(
+        self, service, operation
+    ):
         contents = self.get_client_method_contents(service, operation)
         method_docs = self.get_method_document_block(operation, contents)
         self.assert_contains_line('StreamingBody', method_docs)

tests/functional/botocore/test_auth_config.py

@@ -14,7 +14,7 @@
 from botocore.config import Config
 from botocore.handlers import get_bearer_auth_supported_services
 
-from tests import create_session, mock
+from tests import ALL_SERVICES, create_session, mock
 
 # In the future, a service may have a list of credentials requirements where one
 # signature may fail and others may succeed. e.g. a service may want to use bearer
@@ -46,42 +46,33 @@
 
 
 def _all_test_cases():
-    session = create_session()
-    loader = session.get_component('data_loader')
-
-    services = loader.list_available_services('service-2')
     auth_services = []
     auth_operations = []
 
-    for service in services:
-        service_model = session.get_service_model(service)
+    for service_model in ALL_SERVICES:
+        service = service_model.service_name
         signing_name = service_model.signing_name
         auth_config = service_model.metadata.get('auth', {})
         if signing_name in KNOWN_MIXED_AUTH_SERVICES:
             if set(auth_config) == KNOWN_MIXED_AUTH_SCHEMES:
                 # Skip service due to known mixed auth configurations.
                 continue
         if auth_config:
-            auth_services.append([service, auth_config])
+            auth_services.append((service, auth_config))
         for operation in service_model.operation_names:
             operation_model = service_model.operation_model(operation)
             if operation_model.auth:
-                auth_operations.append([service, operation_model])
+                auth_operations.append((service, operation_model))
     return auth_services, auth_operations
 
 
 AUTH_SERVICES, AUTH_OPERATIONS = _all_test_cases()
 
 
 @pytest.mark.validates_models
-@pytest.mark.parametrize(
-    "auth_service, auth_config",
-    AUTH_SERVICES
-)
+@pytest.mark.parametrize("auth_service, auth_config", AUTH_SERVICES)
 def test_all_requirements_match_for_service(
-        auth_service,
-        auth_config,
-        record_property
+    auth_service, auth_config, record_property
 ):
     record_property("aws_service", auth_service)
     # Validates that all service-level signature types have the same requirements
@@ -91,7 +82,9 @@ def test_all_requirements_match_for_service(
 
 @pytest.mark.validates_models
 @pytest.mark.parametrize("auth_service, operation_model", AUTH_OPERATIONS)
-def test_all_requirements_match_for_operation(auth_service, operation_model, record_property):
+def test_all_requirements_match_for_operation(
+    auth_service, operation_model, record_property
+):
     record_property("aws_service", operation_model.service_model.service_name)
     record_property("aws_operation", operation_model.name)
     # Validates that all operation-level signature types have the same requirements

tests/functional/botocore/test_h2_required.py

@@ -11,7 +11,8 @@
 # ANY KIND, either express or implied. See the License for the specific
 # language governing permissions and limitations under the License.
 import pytest
-from botocore.session import get_session
+
+from tests import ALL_SERVICES
 
 _H2_REQUIRED = object()
 # Service names to list of known HTTP 2 operations
@@ -27,15 +28,11 @@
 
 
 def _all_test_cases():
-    session = get_session()
-    loader = session.get_component('data_loader')
-
-    services = loader.list_available_services('service-2')
     h2_services = []
     h2_operations = []
 
-    for service in services:
-        service_model = session.get_service_model(service)
+    for service_model in ALL_SERVICES:
+        service = service_model.service_name
         h2_config = service_model.metadata.get('protocolSettings', {}).get(
             'h2'
         )

tests/functional/botocore/test_paginator_config.py

@@ -17,6 +17,8 @@
 import pytest
 from jmespath.exceptions import JMESPathError
 
+from tests import ALL_SERVICES
+
 KNOWN_PAGE_KEYS = set(
     [
         'input_token',
@@ -142,8 +144,9 @@ def _pagination_configs():
     session = botocore.session.get_session()
     loader = session.get_component('data_loader')
     services = loader.list_available_services('paginators-1')
+    service_models = {model.service_name: model for model in ALL_SERVICES}
     for service_name in services:
-        service_model = session.get_service_model(service_name)
+        service_model = service_models[service_name]
         page_config = loader.load_service_model(service_name, 'paginators-1')
         for op_name, single_config in page_config['pagination'].items():
             yield (op_name, single_config, service_model)