Add amphora update service client and API test
This patch adds a service client for the amphora agent configuration update
API and adds an API test for the amphora configuration update API.
It also adds the service client for amphora failover and cleans up some
client credintials in the amphora scenario tests.
Change-Id: I4b1a1f48d2f619b883619811539ddb262d6b5f45
diff --git a/octavia_tempest_plugin/tests/api/v2/test_amphora.py b/octavia_tempest_plugin/tests/api/v2/test_amphora.py
new file mode 100644
index 0000000..8b4bb35
--- /dev/null
+++ b/octavia_tempest_plugin/tests/api/v2/test_amphora.py
@@ -0,0 +1,98 @@
+# Copyright 2019 Rackspace US Inc. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+from tempest import config
+from tempest.lib.common.utils import data_utils
+from tempest.lib import decorators
+from tempest.lib import exceptions
+
+from octavia_tempest_plugin.common import constants as const
+from octavia_tempest_plugin.tests import test_base
+from octavia_tempest_plugin.tests import waiters
+
+CONF = config.CONF
+
+
+class AmphoraAPITest(test_base.LoadBalancerBaseTest):
+ """Test the amphora object API."""
+
+ @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.")
+
+ @classmethod
+ def resource_setup(cls):
+ """Setup resources needed by the tests."""
+ super(AmphoraAPITest, cls).resource_setup()
+
+ lb_name = data_utils.rand_name("lb_member_lb1-amphora-api")
+ lb_kwargs = {const.PROVIDER: CONF.load_balancer.provider,
+ const.NAME: lb_name}
+
+ cls._setup_lb_network_kwargs(lb_kwargs)
+
+ lb = cls.mem_lb_client.create_loadbalancer(**lb_kwargs)
+ cls.lb_id = lb[const.ID]
+ cls.addClassResourceCleanup(
+ cls.mem_lb_client.cleanup_loadbalancer,
+ cls.lb_id)
+
+ waiters.wait_for_status(cls.mem_lb_client.show_loadbalancer,
+ cls.lb_id, const.PROVISIONING_STATUS,
+ const.ACTIVE,
+ CONF.load_balancer.lb_build_interval,
+ CONF.load_balancer.lb_build_timeout)
+
+ @decorators.idempotent_id('b7fc231b-dcfa-47a5-99f3-ec5ddcc48f30')
+ def test_amphora_update(self):
+ """Tests the amphora agent configuration update API
+
+ * Tests that users without the loadbalancer admin role cannot
+ update an amphora.
+ * Update the amphora.
+ """
+
+ # We have to do this here as the api_version and clients are not
+ # setup in time to use a decorator or the skip_checks mixin
+ if not self.lb_admin_amphora_client.is_version_supported(
+ self.api_version, '2.7'):
+ raise self.skipException('Amphora update is only available on '
+ 'Octavia API version 2.7 or newer.')
+
+ 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 that a user without the load balancer admin role cannot
+ # create a flavor
+ if CONF.load_balancer.RBAC_test_type == const.ADVANCED:
+ self.assertRaises(
+ exceptions.Forbidden,
+ self.os_primary.amphora_client.update_amphora_config,
+ amphora_1[const.ID])
+
+ self.lb_admin_amphora_client.update_amphora_config(amphora_1[const.ID])
+
+ # TODO(johnsom) Assert that an amphora config setting updated
+ # when we have a setting to check.
+
+ amp = self.lb_admin_amphora_client.show_amphora(amphora_1[const.ID])
+
+ self.assertEqual(const.STATUS_ALLOCATED, amp[const.STATUS])
diff --git a/octavia_tempest_plugin/tests/scenario/v2/test_amphora.py b/octavia_tempest_plugin/tests/scenario/v2/test_amphora.py
index b91a368..30a116c 100644
--- a/octavia_tempest_plugin/tests/scenario/v2/test_amphora.py
+++ b/octavia_tempest_plugin/tests/scenario/v2/test_amphora.py
@@ -115,11 +115,11 @@
# Test that a user with cloud admin role can list the amphorae
if not CONF.load_balancer.RBAC_test_type == const.NONE:
- adm = self.os_admin.amphora_client.list_amphorae()
+ adm = self.lb_admin_amphora_client.list_amphorae()
self.assertTrue(len(adm) >= 2 * self._expected_amp_count(adm))
# Get an actual list of the amphorae
- amphorae = self.os_admin.amphora_client.list_amphorae()
+ amphorae = self.lb_admin_amphora_client.list_amphorae()
# There should be AT LEAST 2, there may be more depending on the
# configured topology, or if there are other LBs created besides ours
@@ -127,7 +127,7 @@
len(amphorae) >= 2 * self._expected_amp_count(amphorae))
show_amphora_response_fields = const.SHOW_AMPHORA_RESPONSE_FIELDS
- if self.mem_amphora_client.is_version_supported(
+ if self.lb_admin_amphora_client.is_version_supported(
self.api_version, '2.1'):
show_amphora_response_fields.append('created_at')
show_amphora_response_fields.append('updated_at')
@@ -140,7 +140,7 @@
self.assertIn(field, amp)
amp_id = amp[const.ID]
- amp_obj = self.os_admin.amphora_client.show_amphora(
+ amp_obj = self.lb_admin_amphora_client.show_amphora(
amphora_id=amp_id)
# Make sure all of the fields exist on the amp show record
@@ -148,7 +148,7 @@
self.assertIn(field, amp_obj)
# Verify a few of the fields are the right type
- if self.mem_amphora_client.is_version_supported(
+ if self.lb_admin_amphora_client.is_version_supported(
self.api_version, '2.1'):
parser.parse(amp_obj[const.CREATED_AT])
parser.parse(amp_obj[const.UPDATED_AT])
@@ -175,13 +175,13 @@
self.assertEqual(amp[field], amp_obj[field])
# Test filtering by loadbalancer_id
- amphorae = self.os_admin.amphora_client.list_amphorae(
+ 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))
self.assertEqual(self._expected_amp_count(amphorae), len(amphorae))
self.assertEqual(self.lb_id, amphorae[0][const.LOADBALANCER_ID])
- amphorae = self.os_admin.amphora_client.list_amphorae(
+ amphorae = self.lb_admin_amphora_client.list_amphorae(
query_params='{loadbalancer_id}={lb_id}'.format(
loadbalancer_id=const.LOADBALANCER_ID, lb_id=lb_id))
self.assertEqual(self._expected_amp_count(amphorae), len(amphorae))
diff --git a/octavia_tempest_plugin/tests/test_base.py b/octavia_tempest_plugin/tests/test_base.py
index 12929f4..2e46812 100644
--- a/octavia_tempest_plugin/tests/test_base.py
+++ b/octavia_tempest_plugin/tests/test_base.py
@@ -121,7 +121,7 @@
cls.os_roles_lb_member.healthmonitor_client)
cls.mem_l7policy_client = cls.os_roles_lb_member.l7policy_client
cls.mem_l7rule_client = cls.os_roles_lb_member.l7rule_client
- cls.mem_amphora_client = cls.os_roles_lb_member.amphora_client
+ cls.lb_admin_amphora_client = cls.os_roles_lb_admin.amphora_client
cls.lb_admin_flavor_profile_client = (
cls.os_roles_lb_admin.flavor_profile_client)
cls.lb_admin_flavor_client = cls.os_roles_lb_admin.flavor_client