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)
28 import os_client_config
30 from xtesting.core import testcase
32 from functest.utils import config
33 from functest.utils import env
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 = ''.join(random.choice(
59 string.ascii_letters + string.digits) for _ in range(30))
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='{}-project_{}'.format(self.case_name[:18], self.guid),
65 description="Created by OPNFV Functest: {}".format(
67 domain_id=self.domain.id)
68 self.__logger.debug("project: %s", self.project)
69 self.user = self.orig_cloud.create_user(
70 name='{}-user_{}'.format(self.case_name, self.guid),
71 password=self.password,
72 domain_id=self.domain.id)
73 self.__logger.debug("user: %s", self.user)
75 if self.orig_cloud.get_role(self.default_member):
76 self.role_name = self.default_member
77 elif self.orig_cloud.get_role(self.default_member.lower()):
78 self.role_name = self.default_member.lower()
80 raise Exception("Cannot detect {}".format(self.default_member))
81 except Exception: # pylint: disable=broad-except
82 self.__logger.info("Creating default role %s", self.default_member)
83 role = self.orig_cloud.create_role(self.default_member)
84 self.role_name = role.name
85 self.__logger.debug("role: %s", role)
86 self.orig_cloud.grant_role(
87 self.role_name, user=self.user.id, project=self.project.id,
88 domain=self.domain.id)
89 osconfig = os_client_config.config.OpenStackConfig()
90 osconfig.cloud_config[
91 'clouds']['envvars']['project_name'] = self.project.name
92 osconfig.cloud_config[
93 'clouds']['envvars']['project_id'] = self.project.id
94 osconfig.cloud_config['clouds']['envvars']['username'] = self.user.name
95 osconfig.cloud_config['clouds']['envvars']['password'] = self.password
96 self.__logger.debug("cloud_config %s", osconfig.cloud_config)
97 self.cloud = shade.OpenStackCloud(
98 cloud_config=osconfig.get_one_cloud())
99 self.__logger.debug("new cloud %s", self.cloud.auth)
101 def get_environ(self):
105 OS_USERNAME=self.user.name,
106 OS_PROJECT_NAME=self.project.name,
107 OS_PROJECT_ID=self.project.id,
108 OS_PASSWORD=self.password)
110 del environ['OS_TENANT_NAME']
111 del environ['OS_TENANT_ID']
112 except Exception: # pylint: disable=broad-except
117 """Remove projects/users"""
119 assert self.orig_cloud
121 self.orig_cloud.delete_user(self.user.id)
123 self.orig_cloud.delete_project(self.project.id)
124 secgroups = self.orig_cloud.list_security_groups(
125 filters={'name': 'default',
126 'project_id': self.project.id})
128 sec_id = secgroups[0].id
129 self.orig_cloud.delete_security_group(sec_id)
130 except Exception: # pylint: disable=broad-except
131 self.__logger.exception("Cannot clean all resources")
134 class TenantNetwork1(testcase.TestCase):
135 # pylint: disable=too-many-instance-attributes
136 """Create a tenant network (scenario1)
138 It creates and configures all tenant network resources required by
139 advanced testcases (subnet, network and router).
141 It ensures that all testcases inheriting from TenantNetwork1 could work
142 without network specific configurations (or at least read the same config
146 __logger = logging.getLogger(__name__)
147 cidr = '192.168.120.0/24'
148 shared_network = False
151 def __init__(self, **kwargs):
152 if "case_name" not in kwargs:
153 kwargs["case_name"] = 'tenantnetwork1'
154 super(TenantNetwork1, self).__init__(**kwargs)
155 self.res_dir = os.path.join(
156 getattr(config.CONF, 'dir_results'), self.case_name)
158 cloud_config = os_client_config.get_config()
159 self.cloud = self.orig_cloud = shade.OpenStackCloud(
160 cloud_config=cloud_config)
161 except Exception: # pylint: disable=broad-except
162 self.cloud = self.orig_cloud = None
164 self.__logger.exception("Cannot connect to Cloud")
166 self.ext_net = self.get_external_network(self.cloud)
167 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 = cloud.search_services('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 not self.allow_no_fip:
224 if hasattr(config.CONF, '{}_network_type'.format(self.case_name)):
225 provider["network_type"] = getattr(
226 config.CONF, '{}_network_type'.format(self.case_name))
227 if hasattr(config.CONF, '{}_physical_network'.format(self.case_name)):
228 provider["physical_network"] = getattr(
229 config.CONF, '{}_physical_network'.format(self.case_name))
230 if hasattr(config.CONF, '{}_segmentation_id'.format(self.case_name)):
231 provider["segmentation_id"] = getattr(
232 config.CONF, '{}_segmentation_id'.format(self.case_name))
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 '{}-net_{}'.format(self.case_name, 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='{}-subnet_{}'.format(self.case_name, self.guid),
249 config.CONF, '{}_private_subnet_cidr'.format(self.case_name),
252 dns_nameservers=[env.get('NAMESERVER')])
253 self.__logger.debug("subnet: %s", self.subnet)
255 self.router = self.cloud.create_router(
256 name='{}-router_{}'.format(self.case_name, 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 self.create_network_resources()
268 status = testcase.TestCase.EX_OK
269 except Exception: # pylint: disable=broad-except
270 self.__logger.exception('Cannot run %s', self.case_name)
272 self.stop_time = time.time()
280 self.cloud.remove_router_interface(
281 self.router, self.subnet.id)
282 self.cloud.delete_router(self.router.id)
284 self.cloud.delete_subnet(self.subnet.id)
286 self.cloud.delete_network(self.network.id)
287 except Exception: # pylint: disable=broad-except
288 self.__logger.exception("cannot clean all resources")
291 class TenantNetwork2(TenantNetwork1):
292 """Create a tenant network (scenario2)
294 It creates new user/project before creating and configuring all tenant
295 network resources required by a testcase (subnet, network and router).
297 It ensures that all testcases inheriting from TenantNetwork2 could work
298 without network specific configurations (or at least read the same config
302 __logger = logging.getLogger(__name__)
304 def __init__(self, **kwargs):
305 if "case_name" not in kwargs:
306 kwargs["case_name"] = 'tenantnetwork2'
307 super(TenantNetwork2, self).__init__(**kwargs)
310 self.project = NewProject(
311 self.cloud, self.case_name, self.guid)
312 self.project.create()
313 self.cloud = self.project.cloud
314 except Exception: # pylint: disable=broad-except
315 self.__logger.exception("Cannot create user or project")
321 super(TenantNetwork2, self).clean()
324 except Exception: # pylint: disable=broad-except
325 self.__logger.exception("Cannot clean all resources")