Add amphora failover API test

This patch adds the amphora failover API test.

This patch also improves the load balancer failover API test to validate
that the amphora IDs have changed. This was a TODO in the code waiting
for the amphora show service client to be completed.

Change-Id: Ibffeb959d88b36be26355ff1b6a96dde8cfcd8c8
Story: 2004852
Task: 29077

octavia_tempest_plugin/tests/api/v2/test_amphora.py

@@ -30,11 +30,9 @@ class AmphoraAPITest(test_base.LoadBalancerBaseTest):
     @classmethod
     def skip_checks(cls):
         super(AmphoraAPITest, cls).skip_checks()
-
         if CONF.load_balancer.provider not in ['amphora', 'octavia']:
-            raise cls.skipException("Amphora tests require provider 'amphora' "
-                                    "or 'octavia' (alias to 'amphora', "
-                                    " deprecated) set.")
+            raise cls.skipException('Amphora tests only run with the amphora '
+                                    'provider enabled.')
 
     @classmethod
     def resource_setup(cls):
@@ -96,3 +94,41 @@ class AmphoraAPITest(test_base.LoadBalancerBaseTest):
         amp = self.lb_admin_amphora_client.show_amphora(amphora_1[const.ID])
 
         self.assertEqual(const.STATUS_ALLOCATED, amp[const.STATUS])
+
+    @decorators.idempotent_id('fb772680-b2ba-4fc3-989b-95ad8492ccaf')
+    def test_amphora_failover(self):
+        """Tests the amphora failover API.
+
+        * Validates that non-admin accounts cannot failover amphora
+        * Fails over an amphora
+        * Validates that a new amphora is built
+        """
+        amphorae = self.lb_admin_amphora_client.list_amphorae(
+            query_params='{loadbalancer_id}={lb_id}'.format(
+                loadbalancer_id=const.LOADBALANCER_ID, lb_id=self.lb_id))
+        amphora_1 = amphorae[0]
+
+        # Test RBAC not authorized for non-admin role
+        if not CONF.load_balancer.RBAC_test_type == const.NONE:
+            self.assertRaises(exceptions.Forbidden,
+                              self.os_primary.amphora_client.amphora_failover,
+                              amphora_1[const.ID])
+            self.assertRaises(
+                exceptions.Forbidden,
+                self.os_roles_lb_member.amphora_client.amphora_failover,
+                amphora_1[const.ID])
+
+        self.lb_admin_amphora_client.amphora_failover(amphora_1[const.ID])
+
+        waiters.wait_for_status(self.mem_lb_client.show_loadbalancer,
+                                self.lb_id, const.PROVISIONING_STATUS,
+                                const.ACTIVE,
+                                CONF.load_balancer.lb_build_interval,
+                                CONF.load_balancer.lb_build_timeout)
+
+        after_amphorae = self.lb_admin_amphora_client.list_amphorae(
+            query_params='{loadbalancer_id}={lb_id}'.format(
+                loadbalancer_id=const.LOADBALANCER_ID, lb_id=self.lb_id))
+
+        for new_amp in after_amphorae:
+            self.assertNotEqual(amphora_1[const.ID], new_amp[const.ID])

octavia_tempest_plugin/tests/api/v2/test_load_balancer.py

@@ -826,6 +826,11 @@ class LoadBalancerAPITest(test_base.LoadBalancerBaseTest):
         lb = self.mem_lb_client.show_loadbalancer(lb[const.ID])
         self.assertEqual(const.ACTIVE, lb[const.PROVISIONING_STATUS])
 
+        if CONF.load_balancer.provider in ['amphora', 'octavia']:
+            before_amphorae = self.lb_admin_amphora_client.list_amphorae(
+                query_params='{loadbalancer_id}={lb_id}'.format(
+                    loadbalancer_id=const.LOADBALANCER_ID, lb_id=lb[const.ID]))
+
         self.os_roles_lb_admin.loadbalancer_client.failover_loadbalancer(
             lb[const.ID])
 
@@ -834,8 +839,17 @@ class LoadBalancerAPITest(test_base.LoadBalancerBaseTest):
                                      const.ACTIVE,
                                      CONF.load_balancer.lb_build_interval,
                                      CONF.load_balancer.lb_build_timeout)
-        # TODO(johnsom) Assert the amphora ID has changed when amp client
-        #               is available.
+
+        if CONF.load_balancer.provider in ['amphora', 'octavia']:
+            after_amphorae = self.lb_admin_amphora_client.list_amphorae(
+                query_params='{loadbalancer_id}={lb_id}'.format(
+                    loadbalancer_id=const.LOADBALANCER_ID, lb_id=lb[const.ID]))
+
+            # Make sure all of the amphora on the load balancer have
+            # failed over
+            for amphora in before_amphorae:
+                for new_amp in after_amphorae:
+                    self.assertNotEqual(amphora[const.ID], new_amp[const.ID])
 
         # Attempt to clean up so that one full test run doesn't start 10+
         # amps before the cleanup phase fires