Merge pull request #603 from mosquito/feature/python3.12-tests

add 3.12

.github/workflows/tests.yml

@@ -51,13 +51,6 @@ jobs:
   tests:
     runs-on: ubuntu-latest
 
-    services:
-      rabbitmq:
-        image: docker://mosquito/aiormq-rabbitmq
-        ports:
-          - 5672:5672
-          - 5671:5671
-
     strategy:
       fail-fast: false
 
@@ -67,6 +60,7 @@ jobs:
           - '3.9'
           - '3.10'
           - '3.11'
+          - '3.12'
     steps:
       - uses: actions/checkout@v2
       - name: Setup python${{ matrix.python }}

aio_pika/message.py

@@ -10,9 +10,9 @@
 from pamqp.common import FieldValue
 
 from .abc import (
-    MILLISECONDS, AbstractChannel, AbstractIncomingMessage,
-    AbstractMessage, AbstractProcessContext, DateType, DeliveryMode,
-    HeadersType, MessageInfo, NoneType,
+    MILLISECONDS, AbstractChannel, AbstractIncomingMessage, AbstractMessage,
+    AbstractProcessContext, DateType, DeliveryMode, HeadersType, MessageInfo,
+    NoneType,
 )
 from .exceptions import ChannelInvalidStateError, MessageProcessError
 from .log import get_logger

aio_pika/patterns/rpc.py

@@ -105,7 +105,7 @@ def __init__(
         self.host_exceptions = host_exceptions
 
     def __remove_future(
-        self, correlation_id: str
+        self, correlation_id: str,
     ) -> Callable[[asyncio.Future], None]:
         def do_remove(future: asyncio.Future) -> None:
             log.debug("Remove done future %r", future)
@@ -122,7 +122,7 @@ def create_future(self) -> Tuple[asyncio.Future, str]:
 
     def _format_routing_key(self, method_name: str) -> str:
         return (
-            f'{self.rpc_exchange.name}::{method_name}'
+            f"{self.rpc_exchange.name}::{method_name}"
             if self.rpc_exchange
             else method_name
         )
@@ -159,7 +159,7 @@ async def close(self) -> None:
 
     async def initialize(
         self, auto_delete: bool = True,
-        durable: bool = False, exchange: str = '', **kwargs: Any,
+        durable: bool = False, exchange: str = "", **kwargs: Any,
     ) -> None:
         if hasattr(self, "result_queue"):
             return
@@ -168,7 +168,8 @@ async def initialize(
             exchange,
             type=ExchangeType.DIRECT,
             auto_delete=True,
-            durable=durable) if exchange else None
+            durable=durable,
+        ) if exchange else None
 
         self.result_queue = await self.channel.declare_queue(
             None, auto_delete=auto_delete, durable=durable, **kwargs,
@@ -403,8 +404,10 @@ async def call(
 
         log.debug("Publishing calls for %s(%r)", routing_key, kwargs)
         exchange = self.rpc_exchange or self.channel.default_exchange
-        await exchange.publish(message, routing_key=routing_key,
-                               mandatory=True)
+        await exchange.publish(
+            message, routing_key=routing_key,
+            mandatory=True,
+        )
 
         log.debug("Waiting RPC result for %s(%r)", routing_key, kwargs)
         return await future
@@ -437,7 +440,7 @@ async def register(
         if self.rpc_exchange:
             await queue.bind(
                 self.rpc_exchange,
-                routing_key
+                routing_key,
             )
 
         if func in self.consumer_tags:

pyproject.toml

@@ -19,6 +19,7 @@ classifiers = [
     "Programming Language :: Python :: 3.9",
     "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
+    "Programming Language :: Python :: 3.12",
     "Programming Language :: Python :: Implementation :: CPython",
     "Programming Language :: Python :: Implementation :: PyPy",
     "Programming Language :: Python",
@@ -60,9 +61,10 @@ sphinx-autobuild = "^2021.3.14"
 timeout-decorator = "^0.5.0"
 types-setuptools = "^65.6.0.2"
 setuptools = "^69.0.3"
+testcontainers = "^3.7.1"
 
 [tool.poetry.group.uvloop.dependencies]
-uvloop = "^0.17.0"
+uvloop = "^0.19"
 
 [build-system]
 requires = ["poetry-core"]
@@ -103,6 +105,10 @@ disallow_untyped_decorators = false
 disallow_untyped_defs = false
 warn_unused_ignores = false
 
+[[tool.mypy.overrides]]
+module = ["testcontainers.*"]
+ignore_missing_imports = true
+
 [tool.pytest.ini_options]
 log_cli = true
 addopts = "-p no:asyncio"

tests/conftest.py

@@ -1,15 +1,17 @@
 import asyncio
 import gc
-import os
+import socket
 import tracemalloc
 from contextlib import suppress
 from functools import partial
+from time import sleep
+from typing import Any, Generator
 
 import aiormq
 import pamqp
 import pytest
 from aiomisc import awaitable
-from aiormq.connection import DEFAULT_PORTS
+from testcontainers.core.container import DockerContainer
 from yarl import URL
 
 import aio_pika
@@ -60,18 +62,61 @@ def payload(coroutine):
                 raise result
 
 
-@pytest.fixture
-def amqp_direct_url(request) -> URL:
-    url = URL(
-        os.getenv("AMQP_URL", "amqp://guest:guest@localhost"),
-    ).update_query(name=request.node.nodeid)
+class RabbitmqContainer(DockerContainer):       # type: ignore
+    _amqp_port: int
+    _amqps_port: int
 
-    default_port = DEFAULT_PORTS[url.scheme]
+    def get_amqp_url(self) -> URL:
+        return URL.build(
+            scheme="amqp", user="guest", password="guest", path="//",
+            host=self.get_container_host_ip(),
+            port=self._amqp_port,
+        )
 
-    if not url.port:
-        url = url.with_port(default_port)
+    def get_amqps_url(self) -> URL:
+        return URL.build(
+            scheme="amqps", user="guest", password="guest", path="//",
+            host=self.get_container_host_ip(),
+            port=self._amqps_port,
+        )
 
-    return url
+    def readiness_probe(self) -> None:
+        host = self.get_container_host_ip()
+        port = int(self.get_exposed_port(5672))
+        while True:
+            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+                try:
+                    sock.connect((host, port))
+                    sock.send(b"AMQP\0x0\0x0\0x9\0x1")
+                    data = sock.recv(4)
+                    if len(data) != 4:
+                        sleep(0.3)
+                        continue
+                except ConnectionError:
+                    sleep(0.3)
+                    continue
+                return
+
+    def start(self) -> "RabbitmqContainer":
+        self.with_exposed_ports(5672, 5671)
+        super().start()
+        self.readiness_probe()
+        self._amqp_port = int(self.get_exposed_port(5672))
+        self._amqps_port = int(self.get_exposed_port(5671))
+        return self
+
+
+@pytest.fixture(scope="module")
+def rabbitmq_container() -> Generator[RabbitmqContainer, Any, Any]:
+    with RabbitmqContainer("mosquito/aiormq-rabbitmq") as container:
+        yield container
+
+
+@pytest.fixture(scope="module")
+def amqp_direct_url(request, rabbitmq_container: RabbitmqContainer) -> URL:
+    return rabbitmq_container.get_amqp_url().update_query(
+        name=request.node.nodeid
+    )
 
 
 @pytest.fixture

tests/test_amqps.py

@@ -15,14 +15,14 @@ def connection_fabric(request):
 
 
 @pytest.fixture
-def create_connection(connection_fabric, event_loop, amqp_url):
+def create_connection(connection_fabric, event_loop, rabbitmq_container):
     ssl_context = ssl.create_default_context()
     ssl_context.check_hostname = False
     ssl_context.verify_mode = ssl.VerifyMode.CERT_NONE
 
     return partial(
         connection_fabric,
-        amqp_url.with_scheme("amqps").with_port(5671),
+        rabbitmq_container.get_amqps_url(),
         loop=event_loop,
         ssl_context=ssl_context,
     )

tests/test_rpc.py

@@ -247,8 +247,8 @@ async def inner():
             await rpc.call("test.not-serializable")
 
     async def test_custom_exchange(self, channel: aio_pika.Channel):
-        rpc_ex1 = await RPC.create(channel, auto_delete=True, exchange='ex1')
-        rpc_ex2 = await RPC.create(channel, auto_delete=True, exchange='ex2')
+        rpc_ex1 = await RPC.create(channel, auto_delete=True, exchange="ex1")
+        rpc_ex2 = await RPC.create(channel, auto_delete=True, exchange="ex2")
         rpc_default = await RPC.create(channel, auto_delete=True)
 
         await rpc_ex1.register("test.rpc", rpc_func, auto_delete=True)