diff --git a/zaza/openstack/charm_tests/rabbitmq_server/tests_k8s.py b/zaza/openstack/charm_tests/rabbitmq_server/tests_k8s.py index b8776e3..aa4e771 100644 --- a/zaza/openstack/charm_tests/rabbitmq_server/tests_k8s.py +++ b/zaza/openstack/charm_tests/rabbitmq_server/tests_k8s.py @@ -14,9 +14,13 @@ """RabbitMQ K8S Testing.""" +import logging +import pika import tenacity +import urllib import rabbitmq_admin +import zaza.charm_lifecycle.utils as lifecycle_utils import zaza.model import zaza.openstack.charm_tests.test_utils as test_utils import zaza.utilities.juju as juju_utils @@ -25,21 +29,60 @@ import zaza.utilities.juju as juju_utils class RabbitMQK8STest(test_utils.BaseCharmTest): """Tests for rabbit on K8s.""" + TEST_USER = "zazauser" + TEST_PASSWORD = "password" + TEST_VHOST = "/" + TEST_QUEUE = "zaza" + APPLICATION_NAME = "rabbitmq" + + @tenacity.retry( + reraise=True, + wait=tenacity.wait_fixed(30), + stop=tenacity.stop_after_attempt(2), + ) + def check_mgr_api(self): + """Test querying managment api.""" + conn = self.get_mgr_connection() + self.assertEqual(conn.overview()["product_name"], "RabbitMQ") + + def setUp(self): + """Run class setup for running tests.""" + super(RabbitMQK8STest, self).setUp() + self.check_mgr_api() + mgr_conn = self.get_mgr_connection() + mgr_conn.create_user(self.TEST_USER, self.TEST_PASSWORD) + mgr_conn.create_user_permission( + self.TEST_USER, + self.TEST_VHOST, + configure=".*", + write=".*", + read=".*", + ) + + @property + def expected_statuses(self): + """Collect expected statuses from config.""" + test_config = lifecycle_utils.get_charm_config(fatal=False) + return test_config.get("target_deploy_status", {}) + def get_service_account(self): """Get service account details.""" result = zaza.model.run_action_on_leader( - 'rabbitmq', - 'get-operator-info', - raise_on_failure=False).data['results'] - return (result['operator-user'], result['operator-password']) + self.APPLICATION_NAME, "get-operator-info", raise_on_failure=False + ).data["results"] + return (result["operator-user"], result["operator-password"]) def get_mgr_endpoint(self): """Get url for management api.""" - public_address = juju_utils.get_application_status( - 'rabbitmq').public_address + public_address = juju_utils.get_k8s_ingress_ip(self.APPLICATION_NAME) return "http://{}:{}".format(public_address, 15672) - def get_connection(self): + def get_client_endpoint(self): + """Get url for management api.""" + public_address = juju_utils.get_k8s_ingress_ip(self.APPLICATION_NAME) + return (public_address, 5672) + + def get_mgr_connection(self): """Return an api connection.""" username, password = self.get_service_account() mgmt_url = self.get_mgr_endpoint() @@ -48,16 +91,118 @@ class RabbitMQK8STest(test_utils.BaseCharmTest): ) return connection - def test_get_operator_info_action(self): + def get_client_connection(self): + """Return an api connection.""" + credentials = pika.PlainCredentials(self.TEST_USER, self.TEST_PASSWORD) + ip, port = self.get_client_endpoint() + parameters = pika.ConnectionParameters(ip, port, "/", credentials) + connection = pika.BlockingConnection(parameters) + return connection + + def get_queue(self, conn, vhost, queue): + """Return a list of nodes in the RabbitMQ cluster.""" + return conn._api_get( + "/api/queues/{}/{}".format( + urllib.parse.quote_plus(vhost), urllib.parse.quote_plus(queue) + ) + ) + + def _test_010_get_operator_info_action(self): """Test running get-operator-info action.""" user, password = self.get_service_account() self.assertEqual(user, "operator") - @tenacity.retry( - reraise=True, - wait=tenacity.wait_fixed(30), - stop=tenacity.stop_after_attempt(2)) - def test_query_api(self): - """Test querying managment api.""" - conn = self.get_connection() - self.assertEqual(conn.overview()['product_name'], "RabbitMQ") + def get_test_queue_members(self, mgr_conn): + """Return list of members of the test queue.""" + return self.get_queue(mgr_conn, self.TEST_VHOST, self.TEST_QUEUE)[ + "members" + ] + + def create_test_queue(self): + """Create test queue.""" + client_conn = self.get_client_connection() + channel = client_conn.channel() + channel.queue_declare( + queue=self.TEST_QUEUE, + durable=True, + arguments={"x-queue-type": "quorum"}, + ) + + def test_quorum_queue_management(self): + """Test charm management for quorum queues.""" + if len(zaza.model.get_units("rabbitmq")) != 1: + self.skipTest( + "Unexpected number of units. There most be only one unit for " + "this test to run." + ) + logging.info("Creating test queue") + self.create_test_queue() + + logging.info("Ensuring test queue has 1 member.") + mgr_conn = self.get_mgr_connection() + self.assertEqual(len(self.get_test_queue_members(mgr_conn)), 1) + + logging.info("Adding 3 units") + # Add 3 additional units. The queue should automatically have the new + # units added as members + zaza.model.scale(self.APPLICATION_NAME, scale_change=3, wait=True) + logging.info("Waiting till model is idle") + zaza.model.block_until_all_units_idle() + logging.info("Wait for status ready") + zaza.model.wait_for_application_states(states=self.expected_statuses) + + logging.info("Checking queue has replicas on 3 of the 4 units.") + members = self.get_test_queue_members(mgr_conn) + self.assertEqual(len(members), 3) + + # target_member is the member going to be removed from queue and then + # added back + target_member = members[0] + target_unit = ( + target_member.split("@")[-1].split(".")[0].replace("-", "/") + ) + logging.info( + "Running action to remove unit {} from queue {}".format( + target_unit, self.TEST_QUEUE + ) + ) + zaza.model.run_action_on_leader( + "rabbitmq", + "delete-member", + action_params={ + "queue-name": self.TEST_QUEUE, + "unit-name": target_unit, + "vhost": self.TEST_VHOST, + }, + ) + logging.info( + "Waiting for leader to warn a queue exists with insufficent " + "members" + ) + zaza.model.block_until_unit_wl_message_match( + zaza.model.get_lead_unit(self.APPLICATION_NAME).name, + "WARNING.*insufficient members", + ) + self.assertNotIn(target_member, self.get_test_queue_members(mgr_conn)) + + logging.info( + "Running action to add unit {} to queue {}".format( + target_unit, self.TEST_QUEUE + ) + ) + zaza.model.run_action_on_leader( + self.APPLICATION_NAME, + "add-member", + action_params={ + "queue-name": self.TEST_QUEUE, + "unit-name": target_unit, + "vhost": self.TEST_VHOST, + }, + ) + logging.info( + "Waiting for leader to remove insufficent members warning" + ) + zaza.model.block_until_unit_wl_message_match( + zaza.model.get_lead_unit(self.APPLICATION_NAME).name, "^$" + ) + self.assertIn(target_member, self.get_test_queue_members(mgr_conn))