add amqp 091 test

Author: Gsantomaggio

Makefile

@@ -7,5 +7,13 @@ rabbitmq-server:
 rabbitmq-server-stop:
 	 ./.ci/ubuntu/gha-setup.sh stop
 
+format:
+	poetry run isort --skip rabbitmq_amqp_python_client/qpid  .
+	poetry run black rabbitmq_amqp_python_client/
+	poetry run black tests/
+	poetry run flake8 --exclude=venv,local_tests,docs/examples,rabbitmq_amqp_python_client/qpid --max-line-length=120 --ignore=E203,W503
+
+test: format
+	poetry run pytest .
 help:
 	cat Makefile

poetry.lock

@@ -9,9 +9,12 @@ readme = "README.md"
 [tool.poetry.dependencies]
 python = "^3.9"
 python-qpid-proton = "^0.39.0"
-pyjwt = "^2.10.1"
 typing-extensions = "^4.13.0"
 
+[tool.poetry.group.test.dependencies]
+pyjwt = "^2.10.1"
+pika = "^1.3.2"
+
 [tool.poetry.group.dev.dependencies]
 flake8 = "^7.1.1"
 isort = "^5.9.3"

pyproject.toml

@@ -0,0 +1,68 @@
+import functools
+
+import pika
+
+from rabbitmq_amqp_python_client import (
+    AddressHelper,
+    Connection,
+    Converter,
+    OutcomeState,
+    QuorumQueueSpecification,
+)
+from rabbitmq_amqp_python_client.qpid.proton import Message
+
+
+def test_publish_queue(connection: Connection) -> None:
+    queue_name = "amqp091-queue"
+    management = connection.management()
+
+    management.declare_queue(QuorumQueueSpecification(name=queue_name))
+
+    raised = False
+
+    publisher = None
+    accepted = False
+
+    try:
+        publisher = connection.publisher(
+            destination=AddressHelper.queue_address(queue_name)
+        )
+        status = publisher.publish(
+            Message(body=Converter.string_to_bytes("my_test_string_for_amqp"))
+        )
+        if status.remote_state == OutcomeState.ACCEPTED:
+            accepted = True
+    except Exception:
+        raised = True
+
+    if publisher is not None:
+        publisher.close()
+
+    assert accepted is True
+    assert raised is False
+
+    credentials = pika.PlainCredentials("guest", "guest")
+    parameters = pika.ConnectionParameters("localhost", credentials=credentials)
+    connection = pika.BlockingConnection(parameters)
+    channel = connection.channel()
+
+    def on_message(chan, method_frame, header_frame, body, userdata=None):
+        """Called when a message is received. Log message and ack it."""
+        chan.basic_ack(delivery_tag=method_frame.delivery_tag)
+        assert body is not None
+        body_text = Converter.bytes_to_string(body)
+        assert body_text is not None
+        assert body_text == "my_test_string_for_amqp"
+        channel.stop_consuming()
+
+    on_message_callback = functools.partial(on_message, userdata="on_message_userdata")
+    channel.basic_qos(
+        prefetch_count=1,
+    )
+    channel.basic_consume(queue_name, on_message_callback)
+
+    channel.start_consuming()
+    connection.close()
+
+    management.delete_queue(queue_name)
+    management.close()