diff --git a/octavia_tempest_plugin/tests/api/v2/test_amphora.py b/octavia_tempest_plugin/tests/api/v2/test_amphora.py index 8b4bb355..78736798 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 @@ 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]) 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 66d26cf6..10ae85d4 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 @@ 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