Merge "Add amphora failover API test"
diff --git a/octavia_tempest_plugin/tests/api/v2/test_amphora.py b/octavia_tempest_plugin/tests/api/v2/test_amphora.py
index 8b4bb35..7873679 100644
--- a/octavia_tempest_plugin/tests/api/v2/test_amphora.py
+++ b/octavia_tempest_plugin/tests/api/v2/test_amphora.py
@@ -30,11 +30,9 @@
@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 @@
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])
diff --git a/octavia_tempest_plugin/tests/api/v2/test_load_balancer.py b/octavia_tempest_plugin/tests/api/v2/test_load_balancer.py
index 66d26cf..10ae85d 100644
--- a/octavia_tempest_plugin/tests/api/v2/test_load_balancer.py
+++ b/octavia_tempest_plugin/tests/api/v2/test_load_balancer.py
@@ -826,6 +826,11 @@
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 @@
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