3 # Copyright (c) 2018 Orange and others.
5 # All rights reserved. This program and the accompanying materials
6 # are made available under the terms of the Apache License, Version 2.0
7 # which accompanies this distribution, and is available at
8 # http://www.apache.org/licenses/LICENSE-2.0
10 """Ease deploying tenant networks
12 It offers a simple way to create all tenant network resources required by a
13 testcase (including all Functest ones):
15 - TenantNetwork1 selects the user and the project set as env vars
16 - TenantNetwork2 creates a user and project to isolate the same resources
18 This classes could be reused by more complexed scenarios (Single VM)
26 import os_client_config
28 from tempest.lib.common.utils import data_utils
29 from xtesting.core import testcase
31 from functest.utils import config
32 from functest.utils import env
33 from functest.utils import functest_utils
37 """Ease creating new projects/users"""
38 # pylint: disable=too-many-instance-attributes
40 __logger = logging.getLogger(__name__)
42 def __init__(self, cloud, case_name, guid):
44 self.orig_cloud = cloud
45 self.case_name = case_name
52 self.default_member = env.get('NEW_USER_ROLE')
55 """Create projects/users"""
56 assert self.orig_cloud
58 self.password = data_utils.rand_password().replace('%', '!')
59 self.__logger.debug("password: %s", self.password)
60 self.domain = self.orig_cloud.get_domain(
61 name_or_id=self.orig_cloud.auth.get(
62 "project_domain_name", "Default"))
63 self.project = self.orig_cloud.create_project(
64 name=f'{self.case_name[:18]}-project_{self.guid}',
65 description=f"Created by OPNFV Functest: {self.case_name}",
66 domain_id=self.domain.id)
67 self.__logger.debug("project: %s", self.project)
68 self.user = self.orig_cloud.create_user(
69 name=f'{self.case_name}-user_{self.guid}',
70 password=self.password,
71 domain_id=self.domain.id)
72 self.__logger.debug("user: %s", self.user)
74 if self.orig_cloud.get_role(self.default_member):
75 self.role_name = self.default_member
76 elif self.orig_cloud.get_role(self.default_member.lower()):
77 self.role_name = self.default_member.lower()
79 raise Exception(f"Cannot detect {self.default_member}")
80 except Exception: # pylint: disable=broad-except
81 self.__logger.info("Creating default role %s", self.default_member)
82 role = self.orig_cloud.create_role(self.default_member)
83 self.role_name = role.name
84 self.__logger.debug("role: %s", role)
85 self.orig_cloud.grant_role(
86 self.role_name, user=self.user.id, project=self.project.id,
87 domain=self.domain.id)
88 osconfig = os_client_config.config.OpenStackConfig()
89 osconfig.cloud_config[
90 'clouds']['envvars']['project_name'] = self.project.name
91 osconfig.cloud_config[
92 'clouds']['envvars']['project_id'] = self.project.id
93 osconfig.cloud_config['clouds']['envvars']['username'] = self.user.name
94 osconfig.cloud_config['clouds']['envvars']['password'] = self.password
95 self.__logger.debug("cloud_config %s", osconfig.cloud_config)
96 self.cloud = shade.OpenStackCloud(
97 cloud_config=osconfig.get_one_cloud())
98 self.__logger.debug("new cloud %s", self.cloud.auth)
100 def get_environ(self):
104 OS_USERNAME=self.user.name,
105 OS_PROJECT_NAME=self.project.name,
106 OS_PROJECT_ID=self.project.id,
107 OS_PASSWORD=self.password)
109 del environ['OS_TENANT_NAME']
110 del environ['OS_TENANT_ID']
111 except Exception: # pylint: disable=broad-except
116 """Remove projects/users"""
118 assert self.orig_cloud
120 self.orig_cloud.delete_user(self.user.id)
122 self.orig_cloud.delete_project(self.project.id)
123 secgroups = self.orig_cloud.list_security_groups(
124 filters={'name': 'default',
125 'project_id': self.project.id})
127 sec_id = secgroups[0].id
128 self.orig_cloud.delete_security_group(sec_id)
129 except Exception: # pylint: disable=broad-except
130 self.__logger.exception("Cannot clean all resources")
133 class TenantNetwork1(testcase.TestCase):
134 # pylint: disable=too-many-instance-attributes
135 """Create a tenant network (scenario1)
137 It creates and configures all tenant network resources required by
138 advanced testcases (subnet, network and router).
140 It ensures that all testcases inheriting from TenantNetwork1 could work
141 without network specific configurations (or at least read the same config
145 __logger = logging.getLogger(__name__)
146 cidr = '192.168.120.0/24'
147 shared_network = False
149 def __init__(self, **kwargs):
150 if "case_name" not in kwargs:
151 kwargs["case_name"] = 'tenantnetwork1'
152 super().__init__(**kwargs)
153 self.dir_results = os.path.join(getattr(config.CONF, 'dir_results'))
154 self.res_dir = os.path.join(self.dir_results, self.case_name)
155 self.output_log_name = 'functest.log'
156 self.output_debug_log_name = 'functest.debug.log'
159 cloud_config = os_client_config.get_config()
160 self.cloud = self.orig_cloud = shade.OpenStackCloud(
161 cloud_config=cloud_config)
162 except Exception: # pylint: disable=broad-except
163 self.cloud = self.orig_cloud = None
164 self.__logger.exception("Cannot connect to Cloud")
165 if env.get('NO_TENANT_NETWORK').lower() != 'true':
167 self.ext_net = self.get_external_network(self.cloud)
168 except Exception: # pylint: disable=broad-except
169 self.__logger.exception("Cannot get the external network")
170 self.guid = str(uuid.uuid4())
176 def get_external_network(cloud):
178 Return the configured external network name or
179 the first retrieved external network name
182 if env.get("EXTERNAL_NETWORK"):
183 network = cloud.get_network(
184 env.get("EXTERNAL_NETWORK"), {"router:external": True})
187 networks = cloud.list_networks({"router:external": True})
193 def get_default_role(cloud, member="Member"):
194 """Get the default role
196 It also tests the role in lowercase to avoid possible conflicts.
198 role = cloud.get_role(member)
200 role = cloud.get_role(member.lower())
204 def get_public_auth_url(cloud):
205 """Get Keystone public endpoint"""
206 keystone_id = functest_utils.search_services(cloud, 'keystone')[0].id
207 endpoint = cloud.search_endpoints(
208 filters={'interface': 'public',
209 'service_id': keystone_id})[0].url
212 def create_network_resources(self):
213 """Create all tenant network resources
215 It creates a router which gateway is the external network detected.
216 The new subnet is attached to that router.
218 Raises: expection on error
221 if env.get('NO_TENANT_NETWORK').lower() != 'true':
224 if hasattr(config.CONF, f'{self.case_name}_network_type'):
225 provider["network_type"] = getattr(
226 config.CONF, f'{self.case_name}_network_type')
227 if hasattr(config.CONF, f'{self.case_name}_physical_network'):
228 provider["physical_network"] = getattr(
229 config.CONF, f'{self.case_name}_physical_network')
230 if hasattr(config.CONF, f'{self.case_name}_segmentation_id'):
231 provider["segmentation_id"] = getattr(
232 config.CONF, f'{self.case_name}_segmentation_id')
233 domain = self.orig_cloud.get_domain(
234 name_or_id=self.orig_cloud.auth.get(
235 "project_domain_name", "Default"))
236 project = self.orig_cloud.get_project(
237 self.cloud.auth['project_name'],
239 self.network = self.orig_cloud.create_network(
240 f'{self.case_name}-net_{self.guid}',
241 provider=provider, project_id=project.id,
242 shared=self.shared_network)
243 self.__logger.debug("network: %s", self.network)
245 self.subnet = self.cloud.create_subnet(
247 subnet_name=f'{self.case_name}-subnet_{self.guid}',
249 config.CONF, f'{self.case_name}_private_subnet_cidr',
252 dns_nameservers=[env.get('NAMESERVER')])
253 self.__logger.debug("subnet: %s", self.subnet)
255 self.router = self.cloud.create_router(
256 name=f'{self.case_name}-router_{self.guid}',
257 ext_gateway_net_id=self.ext_net.id if self.ext_net else None)
258 self.__logger.debug("router: %s", self.router)
259 self.cloud.add_router_interface(self.router, subnet_id=self.subnet.id)
261 def run(self, **kwargs):
262 status = testcase.TestCase.EX_RUN_ERROR
265 self.start_time = time.time()
266 if env.get('NO_TENANT_NETWORK').lower() != 'true':
267 self.create_network_resources()
269 status = testcase.TestCase.EX_OK
270 except Exception: # pylint: disable=broad-except
271 self.__logger.exception('Cannot run %s', self.case_name)
273 self.stop_time = time.time()
281 self.cloud.remove_router_interface(
282 self.router, self.subnet.id)
283 self.cloud.delete_router(self.router.id)
285 self.cloud.delete_subnet(self.subnet.id)
287 self.cloud.delete_network(self.network.id)
288 except Exception: # pylint: disable=broad-except
289 self.__logger.exception("cannot clean all resources")
292 class TenantNetwork2(TenantNetwork1):
293 """Create a tenant network (scenario2)
295 It creates new user/project before creating and configuring all tenant
296 network resources required by a testcase (subnet, network and router).
298 It ensures that all testcases inheriting from TenantNetwork2 could work
299 without network specific configurations (or at least read the same config
303 __logger = logging.getLogger(__name__)
305 def __init__(self, **kwargs):
306 if "case_name" not in kwargs:
307 kwargs["case_name"] = 'tenantnetwork2'
308 super().__init__(**kwargs)
311 self.project = NewProject(
312 self.cloud, self.case_name, self.guid)
313 self.project.create()
314 self.cloud = self.project.cloud
315 except Exception: # pylint: disable=broad-except
316 self.__logger.exception("Cannot create user or project")
325 except Exception: # pylint: disable=broad-except
326 self.__logger.exception("Cannot clean all resources")