Implementing Healthchecks support for #39

amqpstorm/management/api.py

@@ -3,6 +3,7 @@
 from amqpstorm.management.channel import Channel
 from amqpstorm.management.connection import Connection
 from amqpstorm.management.exchange import Exchange
+from amqpstorm.management.healthchecks import HealthChecks
 from amqpstorm.management.http_client import HTTPClient
 from amqpstorm.management.queue import Queue
 from amqpstorm.management.user import User
@@ -23,6 +24,7 @@ def __init__(self, api_url, username, password, timeout=10):
         self._channel = Channel(self.http_client)
         self._connection = Connection(self.http_client)
         self._exchange = Exchange(self.http_client)
+        self._healthchecks = HealthChecks(self.http_client)
         self._queue = Queue(self.http_client)
         self._user = User(self.http_client)
         self._virtual_host = VirtualHost(self.http_client)
@@ -59,6 +61,14 @@ def exchange(self):
         """
         return self._exchange
 
+    @property
+    def healthchecks(self):
+        """RabbitMQ Healthchecks.
+
+        :rtype: amqpstorm.management.healthchecks.Healthchecks
+        """
+        return self._healthchecks
+
     @property
     def queue(self):
         """RabbitMQ Queue Operations.

amqpstorm/management/healthchecks.py

@@ -0,0 +1,25 @@
+from amqpstorm.management.base import ManagementHandler
+
+HEALTHCHECKS = 'healthchecks/node/'
+HEALTHCHECKS_NODE = 'healthchecks/node/%s'
+
+
+class HealthChecks(ManagementHandler):
+    def get(self, node=None):
+        """Run basic healthchecks against the current node, or against a given
+        node.
+
+            Example response:
+                > {"status":"ok"}
+                > {"status":"failed","reason":"string"}
+
+        :param node: Node name
+
+        :raises ApiError: Raises if the remote server encountered an error.
+        :raises ApiConnectionError: Raises if there was a connectivity issue.
+
+        :rtype: dict
+        """
+        if not node:
+            return self.http_client.get(HEALTHCHECKS)
+        return self.http_client.get(HEALTHCHECKS_NODE % node)

amqpstorm/tests/functional/management/healthcheck_tests.py

@@ -0,0 +1,26 @@
+from amqpstorm.management import ManagementApi
+from amqpstorm.tests import HTTP_URL
+from amqpstorm.tests import PASSWORD
+from amqpstorm.tests import USERNAME
+from amqpstorm.tests.utility import TestFunctionalFramework
+from amqpstorm.tests.utility import setup
+
+
+class ApiHealthchecksFunctionalTests(TestFunctionalFramework):
+    @setup()
+    def test_healthtests_get(self):
+        api = ManagementApi(HTTP_URL, USERNAME, PASSWORD)
+
+        result = api.healthchecks.get()
+        self.assertIsInstance(result, dict)
+        self.assertEqual(result['status'], 'ok')
+
+    @setup()
+    def test_healthtests_get_with_node_name(self):
+        api = ManagementApi(HTTP_URL, USERNAME, PASSWORD)
+
+        node_name = api.overview()['contexts'][0]['node']
+
+        result = api.healthchecks.get(node_name)
+        self.assertIsInstance(result, dict)
+        self.assertEqual(result['status'], 'ok')