refresh_token

Author: DanielePalaia

rabbitmq_amqp_python_client/common.py

@@ -18,6 +18,7 @@ class CommonValues(enum.Enum):
     key = "key"
     queue = "queues"
     bindings = "bindings"
+    path_tokens = "/auth/tokens"
 
 
 class ExchangeType(enum.Enum):

rabbitmq_amqp_python_client/connection.py

@@ -383,3 +383,8 @@ def active_producers(self) -> int:
     def active_consumers(self) -> int:
         """Returns the number of active consumers"""
         return len(self._consumers)
+
+    def refresh_token(self, token: str) -> None:
+
+        management = self.management()
+        management.refresh_token(token)

rabbitmq_amqp_python_client/management.py

@@ -573,3 +573,14 @@ def queue_info(self, name: str) -> QueueInfo:
             message_count=queue_info["message_count"],
             consumer_count=queue_info["consumer_count"],
         )
+
+    def refresh_token(self, token: str) -> None:
+
+        self.request(
+            token.encode(),
+            CommonValues.path_tokens.value,
+            CommonValues.command_put.value,
+            [
+                CommonValues.response_code_204.value,
+            ],
+        )

tests/test_connection.py

@@ -44,7 +44,7 @@ def test_connection_ssl(ssl_context) -> None:
     environment.close()
 
 
-def test_connection_auth(environment_auth: Environment) -> None:
+def test_connection_oauth(environment_auth: Environment) -> None:
 
     connection = environment_auth.connection()
     connection.dial()
@@ -54,7 +54,7 @@ def test_connection_auth(environment_auth: Environment) -> None:
     connection.close()
 
 
-def test_connection_auth_with_timeout(environment_auth: Environment) -> None:
+def test_connection_oauth_with_timeout(environment_auth: Environment) -> None:
 
     connection = environment_auth.connection()
     connection.dial()
@@ -65,10 +65,32 @@ def test_connection_auth_with_timeout(environment_auth: Environment) -> None:
     try:
         management = connection.management()
         management.declare_queue(QuorumQueueSpecification(name="test-queue"))
+        management.close()
     except Exception:
         raised = True
 
     assert raised is True
+
+    connection.close()
+
+
+def test_connection_oauth_refresh_token(environment_auth: Environment) -> None:
+
+    connection = environment_auth.connection()
+    connection.dial()
+    # let the token expire
+    time.sleep(3)
+    raised = False
+    # token expired, refresh
+    connection.refresh_token("abcdefghijklmnopqrstuvwxyz0123456789ABCDEFGH")
+
+    try:
+        management = connection.management()
+        management.declare_queue(QuorumQueueSpecification(name="test-queue"))
+    except Exception:
+        raised = True
+
+    assert raised is False
     connection.close()