diff --git a/setup.py b/setup.py
index 74ff5e14..c353b9df 100644
--- a/setup.py
+++ b/setup.py
@@ -76,4 +76,12 @@ def read_version():
       packages=find_packages(),
       install_requires=install_requires,
       extras_require=extras_require,
-      include_package_data=True)
+      include_package_data=True,
+     
+      # the following makes a plugin available to pytest
+      entry_points = {
+        'pytest11': [
+            'aiomoto = tests.aws.aio.aiomoto_fixtures.py',  # or something
+        ]
+      },
+     )
diff --git a/tests/aws/__init__.py b/tests/aws/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/tests/aws/aio/__init__.py b/tests/aws/aio/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/tests/aws/aio/aiomoto_fixtures.py b/tests/aws/aio/aiomoto_fixtures.py
new file mode 100644
index 00000000..32d64997
--- /dev/null
+++ b/tests/aws/aio/aiomoto_fixtures.py
@@ -0,0 +1,166 @@
+"""
+AWS asyncio test fixtures
+"""
+
+import aiobotocore.client
+import aiobotocore.config
+import pytest
+
+from tests.aws.aio.aiomoto_services import MotoService
+from tests.aws.utils import AWS_ACCESS_KEY_ID
+from tests.aws.utils import AWS_SECRET_ACCESS_KEY
+
+
+#
+# Asyncio AWS Services
+#
+
+
+@pytest.fixture
+async def aio_aws_batch_server():
+    async with MotoService("batch") as svc:
+        yield svc.endpoint_url
+
+
+@pytest.fixture
+async def aio_aws_cloudformation_server():
+    async with MotoService("cloudformation") as svc:
+        yield svc.endpoint_url
+
+
+@pytest.fixture
+async def aio_aws_ec2_server():
+    async with MotoService("ec2") as svc:
+        yield svc.endpoint_url
+
+
+@pytest.fixture
+async def aio_aws_ecs_server():
+    async with MotoService("ecs") as svc:
+        yield svc.endpoint_url
+
+
+@pytest.fixture
+async def aio_aws_iam_server():
+    async with MotoService("iam") as svc:
+        yield svc.endpoint_url
+
+
+@pytest.fixture
+async def aio_aws_dynamodb2_server():
+    async with MotoService("dynamodb2") as svc:
+        yield svc.endpoint_url
+
+
+@pytest.fixture
+async def aio_aws_logs_server():
+    # cloud watch logs
+    async with MotoService("logs") as svc:
+        yield svc.endpoint_url
+
+
+@pytest.fixture
+async def aio_aws_s3_server():
+    async with MotoService("s3") as svc:
+        yield svc.endpoint_url
+
+
+@pytest.fixture
+async def aio_aws_sns_server():
+    async with MotoService("sns") as svc:
+        yield svc.endpoint_url
+
+
+@pytest.fixture
+async def aio_aws_sqs_server():
+    async with MotoService("sqs") as svc:
+        yield svc.endpoint_url
+
+
+#
+# Asyncio AWS Clients
+#
+
+
+@pytest.fixture
+def aio_aws_session(aws_credentials, aws_region, event_loop):
+    # pytest-asyncio provides and manages the `event_loop`
+
+    session = aiobotocore.get_session(loop=event_loop)
+    session.user_agent_name = "aiomoto"
+
+    assert session.get_default_client_config() is None
+    aioconfig = aiobotocore.config.AioConfig(max_pool_connections=1, region_name=aws_region)
+
+    # Note: tried to use proxies for the aiobotocore.endpoint, to replace
+    #      'https://batch.us-west-2.amazonaws.com/v1/describejobqueues', but
+    #      the moto.server does not behave as a proxy server.  Leaving this
+    #      here for the record to avoid trying to do it again sometime later.
+    # proxies = {
+    #     'http': os.getenv("HTTP_PROXY", "http://127.0.0.1:5000/moto-api/"),
+    #     'https': os.getenv("HTTPS_PROXY", "http://127.0.0.1:5000/moto-api/"),
+    # }
+    # assert aioconfig.proxies is None
+    # aioconfig.proxies = proxies
+
+    session.set_default_client_config(aioconfig)
+    assert session.get_default_client_config() == aioconfig
+
+    session.set_credentials(AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY)
+    session.set_debug_logger(logger_name="aiomoto")
+
+    yield session
+
+
+@pytest.fixture
+async def aio_aws_client(aio_aws_session):
+    async def _get_client(service_name):
+        async with MotoService(service_name) as srv:
+            async with aio_aws_session.create_client(
+                service_name, endpoint_url=srv.endpoint_url
+            ) as client:
+                yield client
+
+    return _get_client
+
+
+@pytest.fixture
+async def aio_aws_batch_client(aio_aws_session, aio_aws_batch_server):
+    async with aio_aws_session.create_client(
+        "batch", endpoint_url=aio_aws_batch_server
+    ) as client:
+        yield client
+
+
+@pytest.fixture
+async def aio_aws_ec2_client(aio_aws_session, aio_aws_ec2_server):
+    async with aio_aws_session.create_client("ec2", endpoint_url=aio_aws_ec2_server) as client:
+        yield client
+
+
+@pytest.fixture
+async def aio_aws_ecs_client(aio_aws_session, aio_aws_ecs_server):
+    async with aio_aws_session.create_client("ecs", endpoint_url=aio_aws_ecs_server) as client:
+        yield client
+
+
+@pytest.fixture
+async def aio_aws_iam_client(aio_aws_session, aio_aws_iam_server):
+    async with aio_aws_session.create_client("iam", endpoint_url=aio_aws_iam_server) as client:
+        client.meta.config.region_name = "aws-global"  # not AWS_REGION
+        yield client
+
+
+@pytest.fixture
+async def aio_aws_logs_client(aio_aws_session, aio_aws_logs_server):
+    async with aio_aws_session.create_client(
+        "logs", endpoint_url=aio_aws_logs_server
+    ) as client:
+        yield client
+
+
+@pytest.fixture
+async def aio_aws_s3_client(aio_aws_session, aio_aws_s3_server):
+    async with aio_aws_session.create_client("s3", endpoint_url=aio_aws_s3_server) as client:
+        yield client
+
diff --git a/tests/aws/aio/aiomoto_services.py b/tests/aws/aio/aiomoto_services.py
new file mode 100644
index 00000000..e4057bf7
--- /dev/null
+++ b/tests/aws/aio/aiomoto_services.py
@@ -0,0 +1,136 @@
+import asyncio
+import functools
+import logging
+import socket
+import threading
+import time
+import os
+
+# Third Party
+import aiohttp
+import moto.server
+import werkzeug.serving
+
+
+HOST = "127.0.0.1"
+
+_PYCHARM_HOSTED = os.environ.get("PYCHARM_HOSTED") == "1"
+CONNECT_TIMEOUT = 90 if _PYCHARM_HOSTED else 10
+
+
+def get_free_tcp_port(release_socket: bool = False):
+    sckt = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    sckt.bind(("", 0))
+    addr, port = sckt.getsockname()
+    if release_socket:
+        sckt.close()
+        return port
+
+    return sckt, port
+
+
+class MotoService:
+    """ Will Create MotoService.
+    Service is ref-counted so there will only be one per process. Real Service will
+    be returned by `__aenter__`."""
+
+    _services = dict()  # {name: instance}
+
+    def __init__(self, service_name: str, port: int = None):
+        self._service_name = service_name
+
+        if port:
+            self._socket = None
+            self._port = port
+        else:
+            self._socket, self._port = get_free_tcp_port()
+
+        self._thread = None
+        self._logger = logging.getLogger("MotoService")
+        self._refcount = None
+        self._ip_address = HOST
+        self._server = None
+
+    @property
+    def endpoint_url(self):
+        return "http://{}:{}".format(self._ip_address, self._port)
+
+    def __call__(self, func):
+        async def wrapper(*args, **kwargs):
+            await self._start()
+            try:
+                result = await func(*args, **kwargs)
+            finally:
+                await self._stop()
+            return result
+
+        functools.update_wrapper(wrapper, func)
+        wrapper.__wrapped__ = func
+        return wrapper
+
+    async def __aenter__(self):
+        svc = self._services.get(self._service_name)
+        if svc is None:
+            self._services[self._service_name] = self
+            self._refcount = 1
+            await self._start()
+            return self
+        else:
+            svc._refcount += 1
+            return svc
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        self._refcount -= 1
+
+        if self._socket:
+            self._socket.close()
+            self._socket = None
+
+        if self._refcount == 0:
+            del self._services[self._service_name]
+            await self._stop()
+
+    def _server_entry(self):
+        self._main_app = moto.server.DomainDispatcherApplication(
+            moto.server.create_backend_app, service=self._service_name
+        )
+        self._main_app.debug = True
+
+        if self._socket:
+            self._socket.close()  # release right before we use it
+            self._socket = None
+
+        self._server = werkzeug.serving.make_server(
+            self._ip_address, self._port, self._main_app, True
+        )
+        self._server.serve_forever()
+
+    async def _start(self):
+        self._thread = threading.Thread(target=self._server_entry, daemon=True)
+        self._thread.start()
+
+        async with aiohttp.ClientSession() as session:
+            start = time.time()
+
+            while time.time() - start < 10:
+                if not self._thread.is_alive():
+                    break
+
+                try:
+                    # we need to bypass the proxies due to monkeypatches
+                    async with session.get(
+                        self.endpoint_url + "/static", timeout=CONNECT_TIMEOUT
+                    ):
+                        pass
+                    break
+                except (asyncio.TimeoutError, aiohttp.ClientConnectionError):
+                    await asyncio.sleep(0.5)
+            else:
+                await self._stop()  # pytest.fail doesn't call stop_process
+                raise Exception("Cannot start MotoService: {}".format(self._service_name))
+
+    async def _stop(self):
+        if self._server:
+            self._server.shutdown()
+
+        self._thread.join()
diff --git a/tests/aws/aio/conftest.py b/tests/aws/aio/conftest.py
new file mode 100644
index 00000000..03f2ee72
--- /dev/null
+++ b/tests/aws/aio/conftest.py
@@ -0,0 +1,6 @@
+"""
+AWS asyncio test fixtures
+
+Test fixtures are loaded by ``pytest_plugins`` in tests/conftest.py
+"""
+
diff --git a/tests/aws/aio/test_aiomoto_clients.py b/tests/aws/aio/test_aiomoto_clients.py
new file mode 100644
index 00000000..5f3bacc3
--- /dev/null
+++ b/tests/aws/aio/test_aiomoto_clients.py
@@ -0,0 +1,147 @@
+"""
+Test Asyncio AWS Client Fixtures
+
+This test suite checks fixtures for aiobotocore clients.
+
+Do _not_ use default moto mock decorators, which incur:
+AttributeError: 'AWSResponse' object has no attribute 'raw_headers'
+.. seealso:: https://github.com/aio-libs/aiobotocore/issues/755
+"""
+
+import os
+
+import pytest
+from aiobotocore.client import AioBaseClient
+from aiobotocore.session import AioSession
+
+from tests.aws.utils import AWS_REGION
+from tests.aws.utils import AWS_ACCESS_KEY_ID
+from tests.aws.utils import AWS_SECRET_ACCESS_KEY
+from tests.aws.utils import has_moto_mocks
+from tests.aws.utils import response_success
+
+
+def test_aio_aws_session_credentials(aio_aws_session):
+    assert isinstance(aio_aws_session, AioSession)
+    credentials = aio_aws_session.get_credentials()
+    assert credentials.access_key == AWS_ACCESS_KEY_ID
+    assert credentials.secret_key == AWS_SECRET_ACCESS_KEY
+    assert os.getenv("AWS_ACCESS_KEY_ID")
+    assert os.getenv("AWS_SECRET_ACCESS_KEY")
+    assert os.getenv("AWS_ACCESS_KEY_ID") == AWS_ACCESS_KEY_ID
+    assert os.getenv("AWS_SECRET_ACCESS_KEY") == AWS_SECRET_ACCESS_KEY
+
+
+@pytest.mark.asyncio
+async def test_aio_aws_batch_client(aio_aws_batch_client):
+    client = aio_aws_batch_client
+    assert isinstance(client, AioBaseClient)
+
+    assert client.meta.config.region_name == AWS_REGION
+    assert client.meta.region_name == AWS_REGION
+
+    resp = await client.describe_job_queues()
+    assert response_success(resp)
+    assert resp.get("jobQueues") == []
+
+    # the event-name mocks are dynamically generated after calling the method;
+    # for aio-clients, they should be disabled for aiohttp to hit moto.server.
+    assert not has_moto_mocks(client, "before-send.batch.DescribeJobQueues")
+
+
+@pytest.mark.asyncio
+async def test_aio_aws_ec2_client(aio_aws_ec2_client):
+    client = aio_aws_ec2_client
+    assert isinstance(client, AioBaseClient)
+    assert client.meta.config.region_name == AWS_REGION
+    assert client.meta.region_name == AWS_REGION
+
+    resp = await client.describe_instances()
+    assert response_success(resp)
+    assert resp.get("Reservations") == []
+
+    # the event-name mocks are dynamically generated after calling the method;
+    # for aio-clients, they should be disabled for aiohttp to hit moto.server.
+    assert not has_moto_mocks(client, "before-send.ec2.DescribeInstances")
+
+
+@pytest.mark.asyncio
+async def test_aio_aws_ecs_client(aio_aws_ecs_client):
+    client = aio_aws_ecs_client
+    assert isinstance(client, AioBaseClient)
+    assert client.meta.config.region_name == AWS_REGION
+    assert client.meta.region_name == AWS_REGION
+
+    resp = await client.list_task_definitions()
+    assert response_success(resp)
+    assert resp.get("taskDefinitionArns") == []
+
+    # the event-name mocks are dynamically generated after calling the method;
+    # for aio-clients, they should be disabled for aiohttp to hit moto.server.
+    assert not has_moto_mocks(client, "before-send.ecs.ListTaskDefinitions")
+
+
+@pytest.mark.asyncio
+async def test_aio_aws_iam_client(aio_aws_iam_client):
+    client = aio_aws_iam_client
+    assert isinstance(client, AioBaseClient)
+    assert client.meta.config.region_name == "aws-global"  # not AWS_REGION
+    assert client.meta.region_name == "aws-global"  # not AWS_REGION
+
+    resp = await client.list_roles()
+    assert response_success(resp)
+    assert resp.get("Roles") == []
+
+    # the event-name mocks are dynamically generated after calling the method;
+    # for aio-clients, they should be disabled for aiohttp to hit moto.server.
+    assert not has_moto_mocks(client, "before-send.iam.ListRoles")
+
+
+@pytest.mark.asyncio
+async def test_aio_aws_logs_client(aio_aws_logs_client):
+    client = aio_aws_logs_client
+    assert isinstance(client, AioBaseClient)
+    assert client.meta.config.region_name == AWS_REGION
+    assert client.meta.region_name == AWS_REGION
+
+    resp = await client.describe_log_groups()
+    assert response_success(resp)
+    assert resp.get("logGroups") == []
+
+    # the event-name mocks are dynamically generated after calling the method;
+    # for aio-clients, they should be disabled for aiohttp to hit moto.server.
+    assert not has_moto_mocks(client, "before-send.cloudwatch-logs.DescribeLogGroups")
+
+
+@pytest.mark.asyncio
+async def test_aio_aws_s3_client(aio_aws_s3_client):
+    client = aio_aws_s3_client
+    assert isinstance(client, AioBaseClient)
+    assert client.meta.config.region_name == AWS_REGION
+    assert client.meta.region_name == AWS_REGION
+
+    resp = await client.list_buckets()
+    assert response_success(resp)
+    assert resp.get("Buckets") == []
+
+    # the event-name mocks are dynamically generated after calling the method;
+    # for aio-clients, they should be disabled for aiohttp to hit moto.server.
+    assert not has_moto_mocks(client, "before-send.s3.ListBuckets")
+
+
+@pytest.mark.asyncio
+async def test_aio_aws_client(aio_aws_client):
+    # aio_aws_client is an async generator
+    # aio_aws_client(service_name) yields a client
+    async for client in aio_aws_client("s3"):
+        assert isinstance(client, AioBaseClient)
+        assert client.meta.config.region_name == AWS_REGION
+        assert client.meta.region_name == AWS_REGION
+
+        resp = await client.list_buckets()
+        assert response_success(resp)
+        assert resp.get("Buckets") == []
+
+        # the event-name mocks are dynamically generated after calling the method;
+        # for aio-clients, they should be disabled for aiohttp to hit moto.server.
+        assert not has_moto_mocks(client, "before-send.s3.ListBuckets")
diff --git a/tests/aws/aio/test_aiomoto_service.py b/tests/aws/aio/test_aiomoto_service.py
new file mode 100644
index 00000000..0106fb7b
--- /dev/null
+++ b/tests/aws/aio/test_aiomoto_service.py
@@ -0,0 +1,66 @@
+"""
+Test MotoService
+
+Test the aiohttp wrappers on moto.server, which run moto.server in a
+thread for each service (batch, s3, etc), using async/await wrappers
+to start and stop each server.
+"""
+import json
+
+import aiohttp
+import pytest
+
+from tests.aws.aio.aiomoto_services import HOST
+from tests.aws.aio.aiomoto_services import MotoService
+
+
+def test_moto_service():
+    # this instantiates a MotoService but does not start a server
+    service = MotoService("s3")
+    assert HOST in service.endpoint_url
+    assert service._server is None
+
+
+@pytest.mark.asyncio
+async def test_moto_batch_service():
+    async with MotoService("batch") as batch_service:
+        assert batch_service._server  # __aenter__ starts a moto.server
+
+        url = batch_service.endpoint_url + "/v1/describejobqueues"
+        batch_query = {"jobQueues": [], "maxResults": 10}
+        async with aiohttp.ClientSession() as session:
+            async with session.post(url, data=batch_query, timeout=5) as resp:
+                assert resp.status == 200
+                job_queues = await resp.text()
+                job_queues = json.loads(job_queues)
+                assert job_queues["jobQueues"] == []
+
+
+@pytest.mark.asyncio
+async def test_moto_s3_service():
+    async with MotoService("s3") as s3_service:
+        assert s3_service._server  # __aenter__ starts a moto.server
+
+        url = s3_service.endpoint_url
+        s3_xmlns = "http://s3.amazonaws.com/doc/2006-03-01"
+        async with aiohttp.ClientSession() as session:
+            # https://docs.aws.amazon.com/AmazonS3/latest/API/API_ListBuckets.html
+            async with session.get(url, timeout=5) as resp:
+                assert resp.status == 200
+                content = await resp.text()  # ListAllMyBucketsResult XML
+                assert s3_xmlns in content
+
+
+# This test is not necessary to run every time, but might be useful later.
+# @pytest.mark.asyncio
+# async def test_moto_api_service():
+#     # The moto-api is a flask UI to view moto backends
+#     async with MotoService("moto_api") as moto_api_service:
+#         assert moto_api_service._server  # __aenter__ starts a moto.server
+#
+#         url = moto_api_service.endpoint_url + "/moto-api"
+#         async with aiohttp.ClientSession() as session:
+#             async with session.get(url, timeout=5) as resp:
+#                 assert resp.status == 200
+#                 content = await resp.text()
+#                 assert content
diff --git a/tests/aws/aws_fixtures.py b/tests/aws/aws_fixtures.py
new file mode 100644
index 00000000..3c27646e
--- /dev/null
+++ b/tests/aws/aws_fixtures.py
@@ -0,0 +1,258 @@
+"""
+AWS test fixtures
+
+This test suite uses a large suite of moto mocks for the AWS batch
+infrastructure. These infrastructure mocks are derived from the moto test
+suite for testing the batch client. The test infrastructure should be used
+according to the moto license. That license overrides any global license
+applied to my notes project.
+
+.. seealso::
+
+    - https://github.com/spulec/moto/pull/1197/files
+    - https://github.com/spulec/moto/blob/master/tests/test_batch/test_batch.py
+"""
+import os
+from typing import NamedTuple
+from typing import Optional
+
+import boto3
+import botocore.waiter
+import pytest
+from moto import mock_batch
+from moto import mock_ec2
+from moto import mock_ecs
+from moto import mock_iam
+from moto import mock_logs
+
+# import moto.settings
+# moto.settings.TEST_SERVER_MODE = True
+
+# Use dummy AWS credentials
+from moto import mock_s3
+
+from tests.aws.utils import AWS_REGION
+from tests.aws.utils import AWS_ACCESS_KEY_ID
+from tests.aws.utils import AWS_SECRET_ACCESS_KEY
+
+AWS_HOST = "127.0.0.1"
+AWS_PORT = "5000"
+
+
+@pytest.fixture
+def aws_host():
+    return os.getenv("AWS_HOST", AWS_HOST)
+
+
+@pytest.fixture
+def aws_port():
+    return os.getenv("AWS_PORT", AWS_PORT)
+
+
+@pytest.fixture
+def aws_proxy(aws_host, aws_port, monkeypatch):
+    # only required if using a moto stand-alone server or similar local stack
+    monkeypatch.setenv("HTTP_PROXY", f"http://{aws_host}:{aws_port}")
+    monkeypatch.setenv("HTTPS_PROXY", f"http://{aws_host}:{aws_port}")
+
+
+@pytest.fixture
+def aws_credentials(monkeypatch):
+    monkeypatch.setenv("AWS_ACCESS_KEY_ID", AWS_ACCESS_KEY_ID)
+    monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", AWS_SECRET_ACCESS_KEY)
+    monkeypatch.setenv("AWS_SECURITY_TOKEN", "test")
+    monkeypatch.setenv("AWS_SESSION_TOKEN", "test")
+
+
+@pytest.fixture
+def aws_region():
+    return AWS_REGION
+
+
+@pytest.fixture(scope="session")
+def job_queue_name():
+    return "moto_test_job_queue"
+
+
+@pytest.fixture(scope="session")
+def job_definition_name():
+    return "moto_test_job_definition"
+
+
+#
+# AWS Clients
+#
+
+
+class AwsClients(NamedTuple):
+    batch: "botocore.client.Batch"
+    ec2: "botocore.client.EC2"
+    ecs: "botocore.client.ECS"
+    iam: "botocore.client.IAM"
+    logs: "botocore.client.CloudWatchLogs"
+    s3: "botocore.client.S3"
+    region: str
+
+
+@pytest.fixture
+def aws_batch_client(aws_region):
+    with mock_batch():
+        yield boto3.client("batch", region_name=aws_region)
+
+
+@pytest.fixture
+def aws_ec2_client(aws_region):
+    with mock_ec2():
+        yield boto3.client("ec2", region_name=aws_region)
+
+
+@pytest.fixture
+def aws_ecs_client(aws_region):
+    with mock_ecs():
+        yield boto3.client("ecs", region_name=aws_region)
+
+
+@pytest.fixture
+def aws_iam_client(aws_region):
+    with mock_iam():
+        yield boto3.client("iam", region_name=aws_region)
+
+
+@pytest.fixture
+def aws_logs_client(aws_region):
+    with mock_logs():
+        yield boto3.client("logs", region_name=aws_region)
+
+
+@pytest.fixture
+def aws_s3_client(aws_region):
+    with mock_s3():
+        yield boto3.client("s3", region_name=aws_region)
+
+
+@pytest.fixture
+def aws_clients(
+    aws_batch_client,
+    aws_ec2_client,
+    aws_ecs_client,
+    aws_iam_client,
+    aws_logs_client,
+    aws_s3_client,
+    aws_region,
+):
+    return AwsClients(
+        batch=aws_batch_client,
+        ec2=aws_ec2_client,
+        ecs=aws_ecs_client,
+        iam=aws_iam_client,
+        logs=aws_logs_client,
+        s3=aws_s3_client,
+        region=aws_region,
+    )
+
+
+#
+# Batch Infrastructure
+#
+
+
+class BatchInfrastructure:
+    aws_region: str
+    aws_clients: AwsClients
+    vpc_id: Optional[str] = None
+    subnet_id: Optional[str] = None
+    security_group_id: Optional[str] = None
+    iam_arn: Optional[str] = None
+    compute_env_name: Optional[str] = None
+    compute_env_arn: Optional[str] = None
+    job_queue_name: Optional[str] = None
+    job_queue_arn: Optional[str] = None
+    job_definition_name: Optional[str] = None
+    job_definition_arn: Optional[str] = None
+
+
+def batch_infrastructure(
+    aws_clients: AwsClients, job_queue_name: str, job_definition_name: str
+) -> BatchInfrastructure:
+    """
+    Create AWS Batch infrastructure, including:
+    - VPC with subnet
+    - Security group and IAM role
+    - Batch compute environment and job queue
+    - Batch job definition
+
+    This function is not a fixture so that tests can pass the AWS clients to it and then
+    continue to use the infrastructure created by it while the client fixtures are in-tact for
+    the duration of a test.
+    """
+
+    infrastructure = BatchInfrastructure()
+    infrastructure.aws_region = aws_clients.region
+    infrastructure.aws_clients = aws_clients
+
+    resp = aws_clients.ec2.create_vpc(CidrBlock="172.30.0.0/24")
+    vpc_id = resp["Vpc"]["VpcId"]
+
+    resp = aws_clients.ec2.create_subnet(
+        AvailabilityZone=f"{aws_clients.region}a", CidrBlock="172.30.0.0/25", VpcId=vpc_id
+    )
+    subnet_id = resp["Subnet"]["SubnetId"]
+
+    resp = aws_clients.ec2.create_security_group(
+        Description="moto_test_sg_desc", GroupName="moto_test_sg", VpcId=vpc_id
+    )
+    sg_id = resp["GroupId"]
+
+    resp = aws_clients.iam.create_role(
+        RoleName="MotoTestRole", AssumeRolePolicyDocument="moto_test_policy"
+    )
+    iam_arn = resp["Role"]["Arn"]
+
+    compute_env_name = "moto_test_compute_env"
+    resp = aws_clients.batch.create_compute_environment(
+        computeEnvironmentName=compute_env_name,
+        type="UNMANAGED",
+        state="ENABLED",
+        serviceRole=iam_arn,
+    )
+    compute_env_arn = resp["computeEnvironmentArn"]
+
+    resp = aws_clients.batch.create_job_queue(
+        jobQueueName=job_queue_name,
+        state="ENABLED",
+        priority=123,
+        computeEnvironmentOrder=[{"order": 123, "computeEnvironment": compute_env_arn}],
+    )
+    assert resp["jobQueueName"] == job_queue_name
+    assert resp["jobQueueArn"]
+    job_queue_arn = resp["jobQueueArn"]
+
+    resp = aws_clients.batch.register_job_definition(
+        jobDefinitionName=job_definition_name,
+        type="container",
+        containerProperties={
+            "image": "busybox",
+            "vcpus": 2,
+            "memory": 8,
+            "command": ["sleep", "10"],  # NOTE: job runs for 10 sec without overrides
+        },
+    )
+    assert resp["jobDefinitionName"] == job_definition_name
+    assert resp["jobDefinitionArn"]
+    job_definition_arn = resp["jobDefinitionArn"]
+    assert resp["revision"]
+    assert resp["jobDefinitionArn"].endswith(
+        "{0}:{1}".format(resp["jobDefinitionName"], resp["revision"])
+    )
+
+    infrastructure.vpc_id = vpc_id
+    infrastructure.subnet_id = subnet_id
+    infrastructure.security_group_id = sg_id
+    infrastructure.iam_arn = iam_arn
+    infrastructure.compute_env_name = compute_env_name
+    infrastructure.compute_env_arn = compute_env_arn
+    infrastructure.job_queue_name = job_queue_name
+    infrastructure.job_queue_arn = job_queue_arn
+    infrastructure.job_definition_name = job_definition_name
+    infrastructure.job_definition_arn = job_definition_arn
+    return infrastructure
diff --git a/tests/aws/conftest.py b/tests/aws/conftest.py
new file mode 100644
index 00000000..12a4e57d
--- /dev/null
+++ b/tests/aws/conftest.py
@@ -0,0 +1,5 @@
+"""
+AWS test fixtures
+
+Test fixtures are loaded by ``pytest_plugins`` in tests/conftest.py
+"""
diff --git a/tests/aws/test_aws_fixtures.py b/tests/aws/test_aws_fixtures.py
new file mode 100644
index 00000000..d2b18a6d
--- /dev/null
+++ b/tests/aws/test_aws_fixtures.py
@@ -0,0 +1,120 @@
+"""
+Test AWS Fixtures
+
+This test suite checks fixtures for moto clients.
+
+"""
+
+import os
+
+from botocore.client import BaseClient
+
+from tests.aws.aws_fixtures import AwsClients
+from tests.aws.utils import AWS_REGION
+from tests.aws.utils import AWS_ACCESS_KEY_ID
+from tests.aws.utils import AWS_SECRET_ACCESS_KEY
+from tests.aws.utils import has_moto_mocks
+from tests.aws.utils import response_success
+
+
+def test_aws_credentials(aws_credentials):
+    assert os.getenv("AWS_ACCESS_KEY_ID")
+    assert os.getenv("AWS_SECRET_ACCESS_KEY")
+    assert os.getenv("AWS_ACCESS_KEY_ID") == AWS_ACCESS_KEY_ID
+    assert os.getenv("AWS_SECRET_ACCESS_KEY") == AWS_SECRET_ACCESS_KEY
+
+
+def test_aws_clients(aws_clients):
+    assert isinstance(aws_clients, AwsClients)
+    assert isinstance(aws_clients.batch, BaseClient)
+    assert isinstance(aws_clients.ec2, BaseClient)
+    assert isinstance(aws_clients.ecs, BaseClient)
+    assert isinstance(aws_clients.iam, BaseClient)
+    assert isinstance(aws_clients.logs, BaseClient)
+    assert isinstance(aws_clients.s3, BaseClient)
+    assert isinstance(aws_clients.region, str)
+    assert aws_clients.region == AWS_REGION
+
+
+def test_aws_batch_client(aws_batch_client):
+    client = aws_batch_client
+    assert isinstance(client, BaseClient)
+    assert client.meta.config.region_name == AWS_REGION
+    assert client.meta.region_name == AWS_REGION
+
+    resp = client.describe_job_queues()
+    assert response_success(resp)
+    assert resp.get("jobQueues") == []
+
+    # the event-name mocks are dynamically generated after calling the method
+    assert has_moto_mocks(client, "before-send.batch.DescribeJobQueues")
+
+
+def test_aws_ec2_client(aws_ec2_client):
+    client = aws_ec2_client
+    assert isinstance(client, BaseClient)
+    assert client.meta.config.region_name == AWS_REGION
+    assert client.meta.region_name == AWS_REGION
+
+    resp = client.describe_instances()
+    assert response_success(resp)
+    assert resp.get("Reservations") == []
+
+    # the event-name mocks are dynamically generated after calling the method
+    assert has_moto_mocks(client, "before-send.ec2.DescribeInstances")
+
+
+def test_aws_ecs_client(aws_ecs_client):
+    client = aws_ecs_client
+    assert isinstance(client, BaseClient)
+    assert client.meta.config.region_name == AWS_REGION
+    assert client.meta.region_name == AWS_REGION
+
+    resp = client.list_task_definitions()
+    assert response_success(resp)
+    assert resp.get("taskDefinitionArns") == []
+
+    # the event-name mocks are dynamically generated after calling the method
+    assert has_moto_mocks(client, "before-send.ecs.ListTaskDefinitions")
+
+
+def test_aws_iam_client(aws_iam_client):
+    client = aws_iam_client
+    assert isinstance(client, BaseClient)
+    assert client.meta.config.region_name == "aws-global"  # not AWS_REGION
+    assert client.meta.region_name == "aws-global"  # not AWS_REGION
+
+    resp = client.list_roles()
+    assert response_success(resp)
+    assert resp.get("Roles") == []
+
+    # the event-name mocks are dynamically generated after calling the method
+    assert has_moto_mocks(client, "before-send.iam.ListRoles")
+
+
+def test_aws_logs_client(aws_logs_client):
+    client = aws_logs_client
+    assert isinstance(client, BaseClient)
+    assert client.meta.config.region_name == AWS_REGION
+    assert client.meta.region_name == AWS_REGION
+
+    resp = client.describe_log_groups()
+    assert response_success(resp)
+    assert resp.get("logGroups") == []
+
+    # the event-name mocks are dynamically generated after calling the method
+    assert has_moto_mocks(client, "before-send.cloudwatch-logs.DescribeLogGroups")
+
+
+def test_aws_s3_client(aws_s3_client):
+    client = aws_s3_client
+    assert isinstance(client, BaseClient)
+    assert client.meta.config.region_name == AWS_REGION
+    assert client.meta.region_name == AWS_REGION
+
+    resp = client.list_buckets()
+    assert response_success(resp)
+    assert resp.get("Buckets") == []
+
+    # the event-name mocks are dynamically generated after calling the method
+    assert has_moto_mocks(client, "before-send.s3.ListBuckets")
diff --git a/tests/aws/utils.py b/tests/aws/utils.py
new file mode 100644
index 00000000..53be1c08
--- /dev/null
+++ b/tests/aws/utils.py
@@ -0,0 +1,26 @@
+from moto.core.models import BotocoreStubber
+
+AWS_REGION = "us-west-2"
+AWS_ACCESS_KEY_ID = "test_AWS_ACCESS_KEY_ID"
+AWS_SECRET_ACCESS_KEY = "test_AWS_SECRET_ACCESS_KEY"
+
+
+def assert_status_code(response, status_code):
+    assert response.get("ResponseMetadata", {}).get("HTTPStatusCode") == status_code
+
+
+def response_success(response):
+    return response.get("ResponseMetadata", {}).get("HTTPStatusCode") == 200
+
+
+def has_moto_mocks(client, event_name):
+    # moto registers mock callbacks with the `before-send` event-name, using
+    # specific callbacks for the methods that are generated dynamically. By
+    # checking that the first callback is a BotocoreStubber, this verifies
+    # that moto mocks are intercepting client requests.
+    callbacks = client.meta.events._emitter._lookup_cache[event_name]
+    if len(callbacks) > 0:
+        stub = callbacks[0]
+        assert isinstance(stub, BotocoreStubber)
+        return stub.enabled
+    return False
diff --git a/tests/conftest.py b/tests/conftest.py
index 71130734..ab874857 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -470,4 +470,9 @@ async def delete_sqs_queue(sqs_client, queue_url):
     assert_status_code(response, 200)
 
 
-pytest_plugins = ['mock_server']
+pytest_plugins = [
+    "mock_server",
+    "tests.aws.aws_fixtures",
+    "tests.aws.aio.aiomoto_fixtures",
+]
+