cleanup and simplify

pyproject.toml

@@ -82,7 +82,6 @@ dev = [
     "requests ~= 2.32.3", # Used in test_version.py
     "time-machine ~= 2.15.0", # Used in test_signers.py
     "tomli; python_version<'3.11'", # Used in test_version.py
-    "werkzeug ~= 3.0.6", # Used in moto_server.py
 ]
 
 [tool.uv]

tests/boto_tests/unit/test_signers.py

@@ -25,6 +25,7 @@
 DATE = datetime.datetime(2024, 11, 7, 17, 39, 33, tzinfo=timezone.utc)
 
 
+@pytest.mark.parametrize('aws_auth', [{'aws_secret_access_key': 'skid', 'aws_access_key_id': 'akid'}])
 async def test_signers_generate_db_auth_token(rds_client):
     hostname = 'prod-instance.us-east-1.rds.amazonaws.com'
     port = 3306
@@ -41,15 +42,19 @@ async def test_signers_generate_db_auth_token(rds_client):
             hostname, port, username
         )
 
-    # A scheme needs to be appended to the beginning or urlsplit may fail
-    # on certain systems.
-    assert result.startswith(
-        'prod-instance.us-east-1.rds.amazonaws.com:3306/?AWSAccessKeyId=xxx&'
-    )
-    assert result2.startswith(
-        'prod-instance.us-east-1.rds.amazonaws.com:3306/?AWSAccessKeyId=xxx&'
+    expected_result = (
+        'prod-instance.us-east-1.rds.amazonaws.com:3306/?Action=connect'
+        '&DBUser=someusername&X-Amz-Algorithm=AWS4-HMAC-SHA256'
+        '&X-Amz-Date=20161107T173933Z&X-Amz-SignedHeaders=host'
+        '&X-Amz-Expires=900&X-Amz-Credential=akid%2F20161107%2F'
+        'us-east-1%2Frds-db%2Faws4_request&X-Amz-Signature'
+        '=d1138cdbc0ca63eec012ec0fc6c2267e03642168f5884a7795320d4c18374c61'
     )
 
+    assert_url_equal('http://' + result, 'http://' + expected_result)
+
+    assert result2 == result
+
 
 class TestDSQLGenerateDBAuthToken:
     @pytest.fixture(scope="session")

tests/conftest.py

@@ -126,7 +126,7 @@ def alternative_region():
 
 @pytest.fixture
 def signature_version():
-    return 's3'
+    return 'v4'
 
 
 @pytest.fixture
@@ -160,23 +160,18 @@ def config(request, region, signature_version):
     )
 
 
+@pytest.fixture
+def aws_auth():
+    return {'aws_secret_access_key': 'xxx', 'aws_access_key_id': 'xxx'}
+
+
 @pytest.fixture
 def mocking_test():
     # change this flag for test with real aws
     # TODO: this should be merged with pytest.mark.moto
     return True
 
 
-def moto_config(endpoint_url):
-    kw = dict(
-        endpoint_url=endpoint_url,
-        aws_secret_access_key="xxx",
-        aws_access_key_id="xxx",
-    )
-
-    return kw
-
-
 @pytest.fixture
 def patch_attributes(request):
     """Call unittest.mock.patch on arguments passed through a pytest mark.
@@ -231,10 +226,11 @@ async def s3_client(
     mocking_test,
     s3_verify,
     patch_attributes,
+    aws_auth
 ):
     # This depends on mock_attributes because we may want to test event listeners.
     # See the documentation of `mock_attributes` for details.
-    kw = moto_config(moto_server) if mocking_test else {}
+    kw = {'endpoint_url': moto_server, **aws_auth} if mocking_test else {}
 
     async with session.create_client(
         's3', region_name=region, config=config, verify=s3_verify, **kw
@@ -244,9 +240,9 @@ async def s3_client(
 
 @pytest.fixture
 async def alternative_s3_client(
-    session, alternative_region, signature_version, moto_server, mocking_test
+    session, alternative_region, signature_version, moto_server, mocking_test, aws_auth
 ):
-    kw = moto_config(moto_server) if mocking_test else {}
+    kw = {'endpoint_url': moto_server, **aws_auth} if mocking_test else {}
 
     config = AioConfig(
         region_name=alternative_region,
@@ -262,8 +258,8 @@ async def alternative_s3_client(
 
 
 @pytest.fixture
-async def dynamodb_client(session, region, config, moto_server, mocking_test):
-    kw = moto_config(moto_server) if mocking_test else {}
+async def dynamodb_client(session, region, config, moto_server, mocking_test, aws_auth):
+    kw = {'endpoint_url': moto_server, **aws_auth} if mocking_test else {}
     async with session.create_client(
         'dynamodb', region_name=region, config=config, **kw
     ) as client:
@@ -272,81 +268,81 @@ async def dynamodb_client(session, region, config, moto_server, mocking_test):
 
 @pytest.fixture
 async def cloudformation_client(
-    session, region, config, moto_server, mocking_test
+    session, region, config, moto_server, mocking_test, aws_auth
 ):
-    kw = moto_config(moto_server) if mocking_test else {}
+    kw = {'endpoint_url': moto_server, **aws_auth} if mocking_test else {}
     async with session.create_client(
         'cloudformation', region_name=region, config=config, **kw
     ) as client:
         yield client
 
 
 @pytest.fixture
-async def sns_client(session, region, config, moto_server, mocking_test):
-    kw = moto_config(moto_server) if mocking_test else {}
+async def sns_client(session, region, config, moto_server, mocking_test, aws_auth):
+    kw = {'endpoint_url': moto_server, **aws_auth} if mocking_test else {}
     async with session.create_client(
         'sns', region_name=region, config=config, **kw
     ) as client:
         yield client
 
 
 @pytest.fixture
-async def sqs_client(session, region, config, moto_server, mocking_test):
-    kw = moto_config(moto_server) if mocking_test else {}
+async def sqs_client(session, region, config, moto_server, mocking_test, aws_auth):
+    kw = {'endpoint_url': moto_server, **aws_auth} if mocking_test else {}
     async with session.create_client(
         'sqs', region_name=region, config=config, **kw
     ) as client:
         yield client
 
 
 @pytest.fixture
-async def batch_client(session, region, config, moto_server, mocking_test):
-    kw = moto_config(moto_server) if mocking_test else {}
+async def batch_client(session, region, config, moto_server, mocking_test, aws_auth):
+    kw = {'endpoint_url': moto_server, **aws_auth} if mocking_test else {}
     async with session.create_client(
         'batch', region_name=region, config=config, **kw
     ) as client:
         yield client
 
 
 @pytest.fixture
-async def lambda_client(session, region, config, moto_server, mocking_test):
-    kw = moto_config(moto_server) if mocking_test else {}
+async def lambda_client(session, region, config, moto_server, mocking_test, aws_auth):
+    kw = {'endpoint_url': moto_server, **aws_auth} if mocking_test else {}
     async with session.create_client(
         'lambda', region_name=region, config=config, **kw
     ) as client:
         yield client
 
 
 @pytest.fixture
-async def iam_client(session, region, config, moto_server, mocking_test):
-    kw = moto_config(moto_server) if mocking_test else {}
+async def iam_client(session, region, config, moto_server, mocking_test, aws_auth):
+    kw = {'endpoint_url': moto_server, **aws_auth} if mocking_test else {}
     async with session.create_client(
         'iam', region_name=region, config=config, **kw
     ) as client:
         yield client
 
 
 @pytest.fixture
-async def rds_client(session, region, config, moto_server, mocking_test):
-    kw = moto_config(moto_server) if mocking_test else {}
+async def rds_client(session, region, config, moto_server, mocking_test, aws_auth):
+    kw = {'endpoint_url': moto_server, **aws_auth} if mocking_test else {}
     async with session.create_client(
         'rds', region_name=region, config=config, **kw
     ) as client:
         yield client
 
 
 @pytest.fixture
-async def ec2_client(session, region, config, moto_server, mocking_test):
-    kw = moto_config(moto_server) if mocking_test else {}
+async def ec2_client(session, region, config, moto_server, mocking_test, aws_auth):
+    kw = {'endpoint_url': moto_server, **aws_auth} if mocking_test else {}
     async with session.create_client(
         'ec2', region_name=region, config=config, **kw
     ) as client:
         yield client
 
 
 @pytest.fixture
-async def kinesis_client(session, region, config, moto_server, mocking_test):
-    kw = moto_config(moto_server) if mocking_test else {}
+async def kinesis_client(session, region, config, moto_server, mocking_test, aws_auth):
+    kw = {'endpoint_url': moto_server, **aws_auth} if mocking_test else {}
     async with session.create_client(
         'kinesis', region_name=region, config=config, **kw
     ) as client:

tests/mock_server.py

@@ -1,20 +1,33 @@
 import asyncio
 import multiprocessing
+import socket
 
 # Third Party
 import aiohttp
 import aiohttp.web
 import pytest
 from aiohttp.web import StreamResponse
+from moto.server import ThreadedMotoServer
 
-# aiobotocore
-from tests.moto_server import MotoService, get_free_tcp_port, host
 
 _proxy_bypass = {
     "http": None,
     "https": None,
 }
 
+host = '127.0.0.1'
+
+
+def get_free_tcp_port(release_socket: bool = False):
+    sckt = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    sckt.bind((host, 0))
+    addr, port = sckt.getsockname()
+    if release_socket:
+        sckt.close()
+        return port
+
+    return sckt, port
+
 
 # This runs in a subprocess for a variety of reasons
 # 1) early versions of python 3.5 did not correctly set one thread per run loop
@@ -98,5 +111,10 @@ async def _wait_until_up(self):
 
 @pytest.fixture
 async def moto_server(server_scheme):
-    async with MotoService(ssl=server_scheme == 'https') as svc:
-        yield svc.endpoint_url
+    server = ThreadedMotoServer(port=0)
+    try:
+        server.start()
+        host, port = server.get_host_and_port()
+        yield f'http://{host}:{port}'
+    finally:
+        server.stop()