summaryrefslogtreecommitdiff
path: root/watcher_tempest_plugin/tests/scenario/manager.py
diff options
context:
space:
mode:
Diffstat (limited to 'watcher_tempest_plugin/tests/scenario/manager.py')
-rw-r--r--watcher_tempest_plugin/tests/scenario/manager.py206
1 files changed, 0 insertions, 206 deletions
diff --git a/watcher_tempest_plugin/tests/scenario/manager.py b/watcher_tempest_plugin/tests/scenario/manager.py
deleted file mode 100644
index 5364525..0000000
--- a/watcher_tempest_plugin/tests/scenario/manager.py
+++ /dev/null
@@ -1,206 +0,0 @@
1# Copyright 2012 OpenStack Foundation
2# Copyright 2013 IBM Corp.
3# All Rights Reserved.
4#
5# Licensed under the Apache License, Version 2.0 (the "License"); you may
6# not use this file except in compliance with the License. You may obtain
7# a copy of the License at
8#
9# http://www.apache.org/licenses/LICENSE-2.0
10#
11# Unless required by applicable law or agreed to in writing, software
12# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
13# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
14# License for the specific language governing permissions and limitations
15# under the License.
16
17from oslo_log import log
18
19from tempest.common import compute
20from tempest.common import waiters
21from tempest import config
22from tempest.lib.common.utils import data_utils
23from tempest.lib.common.utils import test_utils
24from tempest.lib import exceptions as lib_exc
25import tempest.test
26
27CONF = config.CONF
28
29LOG = log.getLogger(__name__)
30
31
32class ScenarioTest(tempest.test.BaseTestCase):
33 """Base class for scenario tests. Uses tempest own clients. """
34
35 credentials = ['primary']
36
37 @classmethod
38 def setup_clients(cls):
39 super(ScenarioTest, cls).setup_clients()
40 # Clients (in alphabetical order)
41 cls.flavors_client = cls.os_primary.flavors_client
42 cls.compute_floating_ips_client = (
43 cls.os_primary.compute_floating_ips_client)
44 if CONF.service_available.glance:
45 # Check if glance v1 is available to determine which client to use.
46 if CONF.image_feature_enabled.api_v1:
47 cls.image_client = cls.os_primary.image_client
48 elif CONF.image_feature_enabled.api_v2:
49 cls.image_client = cls.os_primary.image_client_v2
50 else:
51 raise lib_exc.InvalidConfiguration(
52 'Either api_v1 or api_v2 must be True in '
53 '[image-feature-enabled].')
54 # Compute image client
55 cls.compute_images_client = cls.os_primary.compute_images_client
56 cls.keypairs_client = cls.os_primary.keypairs_client
57 # Nova security groups client
58 cls.compute_security_groups_client = (
59 cls.os_primary.compute_security_groups_client)
60 cls.compute_security_group_rules_client = (
61 cls.os_primary.compute_security_group_rules_client)
62 cls.servers_client = cls.os_primary.servers_client
63 cls.interface_client = cls.os_primary.interfaces_client
64 # Neutron network client
65 cls.networks_client = cls.os_primary.networks_client
66 cls.ports_client = cls.os_primary.ports_client
67 cls.routers_client = cls.os_primary.routers_client
68 cls.subnets_client = cls.os_primary.subnets_client
69 cls.floating_ips_client = cls.os_primary.floating_ips_client
70 cls.security_groups_client = cls.os_primary.security_groups_client
71 cls.security_group_rules_client = (
72 cls.os_primary.security_group_rules_client)
73
74 if CONF.volume_feature_enabled.api_v2:
75 cls.volumes_client = cls.os_primary.volumes_v2_client
76 cls.snapshots_client = cls.os_primary.snapshots_v2_client
77 else:
78 cls.volumes_client = cls.os_primary.volumes_client
79 cls.snapshots_client = cls.os_primary.snapshots_client
80
81 # ## Test functions library
82 #
83 # The create_[resource] functions only return body and discard the
84 # resp part which is not used in scenario tests
85
86 def _create_port(self, network_id, client=None, namestart='port-quotatest',
87 **kwargs):
88 if not client:
89 client = self.ports_client
90 name = data_utils.rand_name(namestart)
91 result = client.create_port(
92 name=name,
93 network_id=network_id,
94 **kwargs)
95 self.assertIsNotNone(result, 'Unable to allocate port')
96 port = result['port']
97 self.addCleanup(test_utils.call_and_ignore_notfound_exc,
98 client.delete_port, port['id'])
99 return port
100
101 def create_keypair(self, client=None):
102 if not client:
103 client = self.keypairs_client
104 name = data_utils.rand_name(self.__class__.__name__)
105 # We don't need to create a keypair by pubkey in scenario
106 body = client.create_keypair(name=name)
107 self.addCleanup(client.delete_keypair, name)
108 return body['keypair']
109
110 def create_server(self, name=None, image_id=None, flavor=None,
111 validatable=False, wait_until='ACTIVE',
112 clients=None, **kwargs):
113 """Wrapper utility that returns a test server.
114
115 This wrapper utility calls the common create test server and
116 returns a test server. The purpose of this wrapper is to minimize
117 the impact on the code of the tests already using this
118 function.
119 """
120
121 # NOTE(jlanoux): As a first step, ssh checks in the scenario
122 # tests need to be run regardless of the run_validation and
123 # validatable parameters and thus until the ssh validation job
124 # becomes voting in CI. The test resources management and IP
125 # association are taken care of in the scenario tests.
126 # Therefore, the validatable parameter is set to false in all
127 # those tests. In this way create_server just return a standard
128 # server and the scenario tests always perform ssh checks.
129
130 # Needed for the cross_tenant_traffic test:
131 if clients is None:
132 clients = self.os_primary
133
134 if name is None:
135 name = data_utils.rand_name(self.__class__.__name__ + "-server")
136
137 vnic_type = CONF.network.port_vnic_type
138
139 # If vnic_type is configured create port for
140 # every network
141 if vnic_type:
142 ports = []
143
144 create_port_body = {'binding:vnic_type': vnic_type,
145 'namestart': 'port-smoke'}
146 if kwargs:
147 # Convert security group names to security group ids
148 # to pass to create_port
149 if 'security_groups' in kwargs:
150 security_groups = \
151 clients.security_groups_client.list_security_groups(
152 ).get('security_groups')
153 sec_dict = dict([(s['name'], s['id'])
154 for s in security_groups])
155
156 sec_groups_names = [s['name'] for s in kwargs.pop(
157 'security_groups')]
158 security_groups_ids = [sec_dict[s]
159 for s in sec_groups_names]
160
161 if security_groups_ids:
162 create_port_body[
163 'security_groups'] = security_groups_ids
164 networks = kwargs.pop('networks', [])
165 else:
166 networks = []
167
168 # If there are no networks passed to us we look up
169 # for the project's private networks and create a port.
170 # The same behaviour as we would expect when passing
171 # the call to the clients with no networks
172 if not networks:
173 networks = clients.networks_client.list_networks(
174 **{'router:external': False, 'fields': 'id'})['networks']
175
176 # It's net['uuid'] if networks come from kwargs
177 # and net['id'] if they come from
178 # clients.networks_client.list_networks
179 for net in networks:
180 net_id = net.get('uuid', net.get('id'))
181 if 'port' not in net:
182 port = self._create_port(network_id=net_id,
183 client=clients.ports_client,
184 **create_port_body)
185 ports.append({'port': port['id']})
186 else:
187 ports.append({'port': net['port']})
188 if ports:
189 kwargs['networks'] = ports
190 self.ports = ports
191
192 tenant_network = self.get_tenant_network()
193
194 body, servers = compute.create_test_server(
195 clients,
196 tenant_network=tenant_network,
197 wait_until=wait_until,
198 name=name, flavor=flavor,
199 image_id=image_id, **kwargs)
200
201 self.addCleanup(waiters.wait_for_server_termination,
202 clients.servers_client, body['id'])
203 self.addCleanup(test_utils.call_and_ignore_notfound_exc,
204 clients.servers_client.delete_server, body['id'])
205 server = clients.servers_client.show_server(body['id'])['server']
206 return server