Properly handle permissions

.github/workflows/pytest.yml

@@ -36,7 +36,7 @@ jobs:
           chmod a+rwx .rabbitmq/*
       - name: Run RabbitMQ server
         run: |
-          docker run -d -v ./.rabbitmq:/.rabbitmq -e RABBITMQ_CONFIG_FILE=/.rabbitmq/rabbitmq.conf -p 5671:5671 -p 5672:5672 rabbitmq
+          docker run -d -v ./.rabbitmq:/.rabbitmq -e RABBITMQ_CONFIG_FILE=/.rabbitmq/rabbitmq.conf -p 5671:5671 -p 5672:5672 rabbitmq:4.2.1
       - name: Install dependencies
         run: |
           python -m pip install --upgrade pip
@@ -47,7 +47,7 @@ jobs:
         run: |
           # sometimes the scripts hang because the channels can't be closed,
           # so run under a short timeout
-          timeout 60s pytest -v --cov-report=term-missing --cov=varys
+          timeout 120s pytest -v --cov-report=term-missing --cov=varys
       - name: "Upload Logfile"
         uses: actions/upload-artifact@v4
         with:

.gitignore

@@ -3,4 +3,5 @@ varys/__pycache__*
 tests/__pycache__*
 tests/test.log
 varys.egg-info/*
+varys_client.egg-info/*
 .pytest_cache*

.rabbitmq/definitions.json

@@ -0,0 +1,61 @@
+{
+    "permissions": [
+        {
+            "configure": ".*",
+            "read": ".*",
+            "user": "guest",
+            "vhost": "/",
+            "write": ".*"
+        },
+        {
+            "configure": "test-exchange\\.\\w*|test-exchange-3\\.\\w*|test-exchange-3",
+            "read": "test-exchange|test-exchange\\.\\w.*|test-exchange-3|test-exchange-3\\.\\w.*",
+            "user": "guest2",
+            "vhost": "/",
+            "write": "test-exchange|test-exchange\\.\\w.*|test-exchange-3|test-exchange-3\\.\\w.*"
+        }
+    ],
+    "bindings": [],
+    "queues": [],
+    "parameters": [],
+    "policies": [],
+    "rabbitmq_version": "4.2.1",
+    "rabbit_version": "4.2.1",
+    "exchanges": [],
+    "vhosts": [
+        {
+            "limits": [],
+            "metadata": {
+                "description": "Default virtual host",
+                "tags": [],
+                "default_queue_type": "classic"
+            },
+            "name": "/"
+        }
+    ],
+    "users": [
+        {
+            "hashing_algorithm": "rabbit_password_hashing_sha256",
+            "limits": {},
+            "name": "guest",
+            "password_hash": "LzA8uEF2TYeMtmxczECa40VAIOdZmg08i6R+L4XEeg9b62Br",
+            "tags": [
+                "administrator"
+            ]
+        },
+        {
+            "hashing_algorithm": "rabbit_password_hashing_sha256",
+            "limits": {},
+            "name": "guest2",
+            "password_hash": "nV8dlnyFAnteKsCYrrHUPmuNo7rmxLEgOkKXf8WHNW4Nt0GS",
+            "tags": []
+        }
+    ],
+    "global_parameters": [
+        {
+            "name": "cluster_tags",
+            "value": []
+        }
+    ],
+    "topic_permissions": []
+}

.rabbitmq/rabbitmq.conf

@@ -2,3 +2,5 @@ listeners.ssl.default = 5671
 ssl_options.cacertfile           = /.rabbitmq/ca_certificate.pem
 ssl_options.certfile             = /.rabbitmq/server_localhost_certificate.pem
 ssl_options.keyfile              = /.rabbitmq/server_localhost_key.pem
+definitions.import_backend = local_filesystem
+definitions.local.path = /.rabbitmq/definitions.json

setup.cfg

@@ -1,9 +1,9 @@
 [metadata]
 name = varys-client
-version = 1.1.1
+version = 1.2.0
 author = Sam Wilkinson
 author_email = s.a.j.wilkinson@bham.ac.uk
-description = A pika-based python RabbitMQ client for use in the CLIMB-tree project
+description = A pika-based python RabbitMQ client for use in the CLIMB-GRE project
 long_description = file: README.md
 long_description_content_type = text/markdown
 url = https://github.com/climb-tre/varys

tests/test_varys.py

@@ -118,7 +118,10 @@ def quick_turnaround(self):
         self.setUp()
         # timeout seems to need to be at least 0.01s
         received_messages = [
-            message.body.decode()[1:-1] for message in self.v.receive_batch('test_varys', queue_suffix='q', timeout=0.1)
+            message.body.decode()[1:-1]
+            for message in self.v.receive_batch(
+                "test_varys", queue_suffix="q", timeout=0.1
+            )
         ]
 
         self.assertEqual(received_messages, sent_messages)
@@ -226,6 +229,118 @@ def test_quick_turnaround(self):
         self.quick_turnaround()
 
 
+class TestVarysPermissions(unittest.TestCase):
+
+    def setUp(self):
+        config = {
+            "version": "0.1",
+            "profiles": {
+                "test": {
+                    "username": "guest2",
+                    "password": "guest",
+                    "amqp_url": "localhost",
+                    "port": 5672,
+                    "use_tls": False,
+                },
+                "admin": {
+                    "username": "guest",
+                    "password": "guest",
+                    "amqp_url": "localhost",
+                    "port": 5672,
+                    "use_tls": False,
+                },
+            },
+        }
+
+        with open(TMP_FILENAME, "w") as f:
+            json.dump(config, f, ensure_ascii=False)
+
+        # Setup exchange
+        admin_varys = Varys("admin", LOG_FILENAME, config_path=TMP_FILENAME)
+        admin_varys.send("setup message", "test-exchange", queue_suffix="test_queue")
+        admin_varys.close()
+
+        credentials = pika.PlainCredentials("guest", "guest")
+
+        connection = pika.BlockingConnection(
+            pika.ConnectionParameters("localhost", credentials=credentials)
+        )
+        channel = connection.channel()
+
+        channel.queue_purge(queue="test-exchange.test_queue")
+
+        with open(TMP_FILENAME, "w") as f:
+            json.dump(config, f, ensure_ascii=False)
+
+        self.v = Varys("test", LOG_FILENAME, config_path=TMP_FILENAME)
+
+    def tearDown(self):
+        # this seems to prevent some hanging
+        # or errors related to closing connections that haven't opened yet
+        # I presume because some operations are so fast
+        # that we try to close the connections before they've opened
+        # 0.01s seems to be sufficient; 0.1s is just a bit conservative
+        time.sleep(0.1)
+
+        self.v.close()
+        os.remove(TMP_FILENAME)
+        time.sleep(0.1)
+
+        credentials = pika.PlainCredentials("guest", "guest")
+
+        connection = pika.BlockingConnection(
+            pika.ConnectionParameters("localhost", credentials=credentials)
+        )
+        channel = connection.channel()
+
+        channel.queue_purge(queue="test-exchange.test_queue")
+
+        connection.close()
+        time.sleep(0.5)
+
+        # check that all file handles were dropped for relevant loggers
+        for logger_name in ["test-exchange", "test-exchange-2", "test-exchange-3"]:
+            logger = logging.getLogger(logger_name)
+            self.assertEqual(len(logger.handlers), 0)
+
+    def test_not_permitted_declare_fail(self):
+        self.v.send(TEXT, "test-exchange-2", queue_suffix="test_queue")
+        time.sleep(0.5)
+        with open(LOG_FILENAME, "r") as f:
+            loglines = f.readlines()
+
+        self.assertTrue(
+            any(
+                "pika.exceptions.ChannelClosedByBroker: (403, " in message
+                for message in loglines
+            )
+        )
+
+    def test_send_receive_extant_queue(self):
+        self.v.send(TEXT, "test-exchange", queue_suffix="test_queue")
+        message = self.v.receive("test-exchange", queue_suffix="test_queue")
+        self.assertEqual(TEXT, json.loads(message.body))
+
+        logger = logging.getLogger("test-exchange")
+        self.assertEqual(len(logger.handlers), 1)
+
+    def test_send_nonexistant_queue(self):
+        self.v.send(TEXT, "test-exchange", queue_suffix="test_queue_2")
+        message = self.v.receive("test-exchange", queue_suffix="test_queue_2")
+        self.assertEqual(TEXT, json.loads(message.body))
+
+        logger = logging.getLogger("test-exchange")
+        self.assertEqual(len(logger.handlers), 1)
+
+    def test_send_nonexistant_exchange(self):
+        self.v.send(TEXT, "test-exchange-3", queue_suffix="test_queue")
+        message = self.v.receive("test-exchange-3", queue_suffix="test_queue")
+        self.assertEqual(TEXT, json.loads(message.body))
+
+        logger = logging.getLogger("test-exchange-3")
+        self.assertEqual(len(logger.handlers), 1)
+
+
 class TestVarysConfig(unittest.TestCase):
     def tearDown(self):
         os.remove(TMP_FILENAME)

varys/consumer.py

@@ -1,5 +1,6 @@
 import functools
 import pika
+from pika import exceptions as pika_exceptions
 import time
 
 from varys.utils import varys_message
@@ -71,12 +72,40 @@ def run(self):
 
                 self._connection = pika.BlockingConnection(self._parameters)
                 self._channel = self._connection.channel()
-                self._channel.exchange_declare(
-                    exchange=self._exchange,
-                    exchange_type=self._exchange_type,
-                    durable=True,
-                )
-                self._channel.queue_declare(queue=self._queue, durable=True)
+                try:
+                    self._channel.exchange_declare(
+                        exchange=self._exchange,
+                        exchange_type=self._exchange_type,
+                        durable=True,
+                        passive=True,
+                    )
+                except pika_exceptions.ChannelClosed as e:
+                    if e.reply_code != 404:
+                        raise
+
+                    self._log.info(
+                        f"Exchange {self._exchange} does not exist, creating it..."
+                    )
+                    self._channel = self._connection.channel()
+                    self._channel.exchange_declare(
+                        exchange=self._exchange,
+                        exchange_type=self._exchange_type,
+                        durable=True,
+                    )
+                try:
+                    self._channel.queue_declare(
+                        queue=self._queue, durable=True, passive=True
+                    )
+                except pika_exceptions.ChannelClosed as e:
+                    if e.reply_code != 404:
+                        raise
+
+                    self._log.info(
+                        f"Queue {self._queue} does not exist, creating it..."
+                    )
+                    self._channel = self._connection.channel()
+                    self._channel.queue_declare(queue=self._queue, durable=True)
+
                 self._channel.queue_bind(
                     queue=self._queue,
                     exchange=self._exchange,

varys/controller.py

@@ -75,7 +75,7 @@ def send(
         self,
         message,
         exchange,
-        queue_suffix=False,
+        queue_suffix="",
         exchange_type="fanout",
         max_attempts=1,
         reconnect_wait=10,
@@ -109,7 +109,7 @@ def send(
     def receive(
         self,
         exchange,
-        queue_suffix=False,
+        queue_suffix="",
         timeout=None,
         exchange_type="fanout",
         prefetch_count=5,

varys/producer.py

@@ -1,5 +1,6 @@
 import functools
 import pika
+from pika import exceptions as pika_exceptions
 import time
 import json
 
@@ -46,8 +47,17 @@ def publish_message(self, message, max_attempts=3):
 
         attempt = 0
         while attempt < max_attempts:
+            attempt += 1
+
+            if self._connection is None or self._connection.is_closed:
+                self._log.warning(
+                    "Connection is closed, cannot publish message, attempting to reconnect..."
+                )
+                if self._reconnect_wait > 0:
+                    time.sleep(self._reconnect_wait)
+                continue
+
             try:
-                attempt += 1
                 self._log.info(
                     f"Sending message (attempt {attempt}): {json.dumps(message)}"
                 )
@@ -85,12 +95,41 @@ def run(self):
             try:
                 self._connection = pika.BlockingConnection(self._parameters)
                 self._channel = self._connection.channel()
-                self._channel.exchange_declare(
-                    exchange=self._exchange,
-                    exchange_type=self._exchange_type,
-                    durable=True,
-                )
-                self._channel.queue_declare(queue=self._queue, durable=True)
+                try:
+                    self._channel.exchange_declare(
+                        exchange=self._exchange,
+                        exchange_type=self._exchange_type,
+                        durable=True,
+                        passive=True,
+                    )
+                except pika_exceptions.ChannelClosed as e:
+                    if e.reply_code != 404:
+                        raise
+
+                    self._log.info(
+                        f"Exchange {self._exchange} does not exist, creating it..."
+                    )
+                    self._channel = self._connection.channel()
+                    self._channel.exchange_declare(
+                        exchange=self._exchange,
+                        exchange_type=self._exchange_type,
+                        durable=True,
+                    )
+
+                try:
+                    self._channel.queue_declare(
+                        queue=self._queue, durable=True, passive=True
+                    )
+                except pika_exceptions.ChannelClosed as e:
+                    if e.reply_code != 404:
+                        raise
+
+                    self._log.info(
+                        f"Queue {self._queue} does not exist, creating it..."
+                    )
+                    self._channel = self._connection.channel()
+                    self._channel.queue_declare(queue=self._queue, durable=True)
+
                 self._channel.queue_bind(
                     queue=self._queue,
                     exchange=self._exchange,