Ported neutron-gateway tests to zaza (400/401)

zaza/openstack/charm_tests/neutron/tests.py

@@ -17,6 +17,7 @@
 """Encapsulating `neutron-openvswitch` testing."""
 
 
+import time
 import logging
 import tenacity
 import unittest
@@ -39,6 +40,72 @@ class NeutronGatewayTest(test_utils.OpenStackBaseTest):
         cls.current_os_release = openstack_utils.get_os_release()
         cls.services = cls._get_services()
 
+        # set up clients
+        cls.neutron_client = (
+            openstack_utils.get_neutron_session_client(cls.keystone_session))
+
+        bionic_stein = openstack_utils.get_os_release('bionic_stein')
+
+        cls.pgrep_full = (True if cls.current_os_release >= bionic_stein
+                          else False)
+
+    def test_400_create_network(self):
+        """Create a network, verify that it exists, and then delete it."""
+        logging.debug('Creating neutron network...')
+        self.neutron_client.format = 'json'
+        net_name = 'test_net'
+
+        # Verify that the network doesn't exist
+        networks = self.neutron_client.list_networks(name=net_name)
+        net_count = len(networks['networks'])
+        assert net_count == 0, (
+            "Expected zero networks, found {}".format(net_count))
+
+        # Create a network and verify that it exists
+        network = {'name': net_name}
+        self.neutron_client.create_network({'network': network})
+
+        networks = self.neutron_client.list_networks(name=net_name)
+        logging.debug('Networks: {}'.format(networks))
+        net_len = len(networks['networks'])
+        assert net_len == 1, (
+            "Expected 1 network, found {}".format(net_len))
+
+        logging.debug('Confirming new neutron network...')
+        network = networks['networks'][0]
+        assert network['name'] == net_name, "network ext_net not found"
+
+        # Cleanup
+        logging.debug('Deleting neutron network...')
+        self.neutron_client.delete_network(network['id'])
+
+    def test_401_enable_qos(self):
+        """Check qos settings set via neutron-api charm."""
+        if (self.current_os_release >=
+                openstack_utils.get_os_release('trusty_mitaka')):
+            logging.info('running qos check')
+
+            with self.config_change(
+                    {'enable-qos': 'False'},
+                    {'enable-qos': 'True'},
+                    application_name="neutron-api"):
+
+                # wait for neutron-gateway to complete as well
+                time.sleep(60)
+
+                # obtain dhcp agent to identify the neutron-gateway host
+                dhcp_agent = self.neutron_client.list_agents(
+                    binary='neutron-dhcp-agent')['agents'][0]
+                neutron_gw_host = dhcp_agent['host']
+                logging.debug('neutron gw host: {}'.format(neutron_gw_host))
+
+                # check extensions on the ovs agent to validate qos
+                ovs_agent = self.neutron_client.list_agents(
+                    binary='neutron-openvswitch-agent',
+                    host=neutron_gw_host)['agents'][0]
+
+                self.assertIn('qos', ovs_agent['configurations']['extensions'])
+
     def test_900_restart_on_config_change(self):
         """Checking restart happens on config change.
 
@@ -59,15 +126,9 @@ class NeutronGatewayTest(test_utils.OpenStackBaseTest):
         # Config file affected by juju set config change
         conf_file = '/etc/neutron/neutron.conf'
 
-        bionic_stein = openstack_utils.get_os_release('bionic_stein')
-
         # Make config change, check for service restarts
         logging.info(
             'Setting verbose on neutron-api {}'.format(set_alternate))
-        if self.current_os_release >= bionic_stein:
-            pgrep_full = True
-        else:
-            pgrep_full = False
         self.restart_on_changed(
             conf_file,
             set_default,
@@ -75,7 +136,7 @@ class NeutronGatewayTest(test_utils.OpenStackBaseTest):
             default_entry,
             alternate_entry,
             self.services,
-            pgrep_full=pgrep_full)
+            pgrep_full=self.pgrep_full)
 
     def test_910_pause_and_resume(self):
         """Run pause and resume tests.
@@ -83,14 +144,9 @@ class NeutronGatewayTest(test_utils.OpenStackBaseTest):
         Pause service and check services are stopped then resume and check
         they are started
         """
-        bionic_stein = openstack_utils.get_os_release('bionic_stein')
-        if self.current_os_release >= bionic_stein:
-            pgrep_full = True
-        else:
-            pgrep_full = False
         with self.pause_resume(
                 self.services,
-                pgrep_full=pgrep_full):
+                pgrep_full=self.pgrep_full):
             logging.info("Testing pause resume")
 
     @classmethod