Neutron gateway actions

zaza/openstack/charm_tests/neutron/tests.py

@@ -21,6 +21,7 @@
 
 
 import copy
+import json
 import logging
 import tenacity
 
@@ -248,6 +249,141 @@ def _get_services(cls):
         return services
 
 
+class NeutronGatewayStatusActionsTest(test_utils.OpenStackBaseTest):
+    """Test status actions of Neutron Gateway Charm.
+
+    actions:
+      * get-status-routers
+      * get-status-dhcp
+      * get-status-lb
+    """
+
+    SKIP_LBAAS_TESTS = True
+
+    @classmethod
+    def setUpClass(cls, application_name='neutron-gateway', model_alias=None):
+        """Run class setup for running Neutron Gateway tests."""
+        super(NeutronGatewayStatusActionsTest, cls).setUpClass(
+            application_name, model_alias)
+        # set up clients
+        cls.neutron_client = (
+            openstack_utils.get_neutron_session_client(cls.keystone_session))
+
+        # Loadbalancer tests not supported on Train and above and on
+        # releases Mitaka and below
+        current_release = openstack_utils.get_os_release()
+        bionic_train = openstack_utils.get_os_release('bionic_train')
+        xenial_mitaka = openstack_utils.get_os_release('xenial_mitaka')
+        cls.SKIP_LBAAS_TESTS = (current_release <= xenial_mitaka or
+                                current_release >= bionic_train)
+
+    def _assert_result_match(self, action_result, resource_list,
+                             resource_name):
+        """Assert that action_result contains same data as resource_list."""
+        # make sure that action completed successfully
+        self.assertEqual(action_result.status, 'completed')
+
+        # extract data from juju action
+        action_data = action_result.data.get('results', {}).get(resource_name)
+        resources_from_action = json.loads(action_data)
+
+        # pull resource IDs from expected resource list and juju action data
+        expected_resource_ids = {resource['id'] for resource in resource_list}
+        result_resource_ids = {resource['id'] for resource in
+                               resources_from_action}
+
+        # assert that juju action returned expected resources
+        self.assertEqual(result_resource_ids, expected_resource_ids)
+
+    def test_get_status_routers(self):
+        """Test that get-status-routers reports correct neutron routers."""
+        # fetch neutron routers using neutron client
+        ngw_unit = zaza.model.get_units(self.application_name,
+                                        model_name=self.model_name)[0]
+        routers_from_client = self.neutron_client.list_routers().get(
+            'routers', [])
+
+        if not routers_from_client:
+            self.fail('At least one router must be configured for this test '
+                      'to pass.')
+
+        # fetch neutron routers using juju-action
+        result = zaza.model.run_action(ngw_unit.entity_id,
+                                       'get-status-routers',
+                                       model_name=self.model_name,
+                                       action_params={"format": "json"})
+
+        # assert that data from neutron client match data from juju action
+        self._assert_result_match(result, routers_from_client, 'router-list')
+
+    def test_get_status_dhcp(self):
+        """Test that get-status-dhcp reports correct DHCP networks."""
+        # fetch DHCP networks using neutron client
+        ngw_unit = zaza.model.get_units(self.application_name,
+                                        model_name=self.model_name)[0]
+        networks_from_client = self.neutron_client.list_networks().get(
+            'networks', [])
+
+        if not networks_from_client:
+            self.fail('At least one network must be configured for this test '
+                      'to pass.')
+
+        # fetch DHCP networks using juju-action
+        result = zaza.model.run_action(ngw_unit.entity_id,
+                                       'get-status-dhcp',
+                                       model_name=self.model_name,
+                                       action_params={"format": "json"})
+
+        # assert that data from neutron client match data from juju action
+        self._assert_result_match(result, networks_from_client,
+                                  'dhcp-networks')
+
+    def test_get_status_load_balancers(self):
+        """Test that get-status-lb reports correct loadbalancers."""
+        if self.SKIP_LBAAS_TESTS:
+            self.skipTest('LBaasV2 is not supported in this version.')
+
+        loadbalancer_id = None
+
+        try:
+            # create LBaasV2 for the purpose of this test
+            lbaas_name = 'test_lbaas'
+            subnet_list = self.neutron_client.list_subnets(
+                name='private_subnet').get('subnets', [])
+
+            if not subnet_list:
+                raise RuntimeError('Expected subnet "private_subnet" is not '
+                                   'configured.')
+
+            subnet = subnet_list[0]
+            loadbalancer_data = {'loadbalancer': {'name': lbaas_name,
+                                                  'vip_subnet_id': subnet['id']
+                                                  }
+                                 }
+            loadbalancer = self.neutron_client.create_loadbalancer(
+                body=loadbalancer_data)
+            loadbalancer_id = loadbalancer['loadbalancer']['id']
+
+            # test that client and action report same data
+            ngw_unit = zaza.model.get_units(self.application_name,
+                                            model_name=self.model_name)[0]
+            lbaas_from_client = self.neutron_client.list_loadbalancers().get(
+                'loadbalancers', [])
+
+            result = zaza.model.run_action(ngw_unit.entity_id,
+                                           'get-status-lb',
+                                           model_name=self.model_name,
+                                           action_params={"format": "json"})
+
+            self._assert_result_match(result, lbaas_from_client,
+                                      'load-balancers')
+        except Exception as exc:
+            raise exc
+        finally:
+            if loadbalancer_id:
+                self.neutron_client.delete_loadbalancer(loadbalancer_id)
+
+
 class NeutronCreateNetworkTest(test_utils.OpenStackBaseTest):
     """Test creating a Neutron network through the API.