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 ressources required by a
13 testcase (including all Functest ones):
14 - TenantNetwork1 selects the user and the project set as env vars
15 - TenantNetwork2 creates a user and project to isolate the same ressources
17 This classes could be reused by more complexed scenarios (Single VM)
25 import os_client_config
27 from xtesting.core import testcase
29 from functest.utils import config
30 from functest.utils import env
33 class NewProject(object):
34 """Ease creating new projects/users"""
35 # pylint: disable=too-many-instance-attributes
37 __logger = logging.getLogger(__name__)
38 default_member = "Member"
40 def __init__(self, cloud, case_name, guid):
42 self.orig_cloud = cloud
43 self.case_name = case_name
52 """Create projects/users"""
53 assert self.orig_cloud
55 self.password = str(uuid.uuid4())
56 self.domain = self.orig_cloud.get_domain(
57 name_or_id=self.orig_cloud.auth.get(
58 "project_domain_name", "Default"))
59 self.project = self.orig_cloud.create_project(
60 name='{}-project_{}'.format(self.case_name, self.guid),
61 description="Created by OPNFV Functest: {}".format(
63 domain_id=self.domain.id)
64 self.__logger.debug("project: %s", self.project)
65 self.user = self.orig_cloud.create_user(
66 name='{}-user_{}'.format(self.case_name, self.guid),
67 password=self.password,
68 domain_id=self.domain.id)
69 self.__logger.debug("user: %s", self.user)
71 if not self.orig_cloud.get_role(
72 self.default_member) and not self.orig_cloud.get_role(
73 self.default_member.lower()):
74 raise Exception("Cannot detect {}".format(self.default_member))
75 except Exception: # pylint: disable=broad-except
76 self.__logger.info("Creating default role %s", self.default_member)
77 self.role = self.orig_cloud.create_role(self.default_member)
78 self.__logger.debug("role: %s", self.role)
79 self.orig_cloud.grant_role(
80 self.default_member, user=self.user.id, project=self.project.id,
81 domain=self.domain.id)
82 osconfig = os_client_config.config.OpenStackConfig()
83 osconfig.cloud_config[
84 'clouds']['envvars']['project_name'] = self.project.name
85 osconfig.cloud_config['clouds']['envvars']['username'] = self.user.name
86 osconfig.cloud_config['clouds']['envvars']['password'] = self.password
87 self.__logger.debug("cloud_config %s", osconfig.cloud_config)
88 self.cloud = shade.OpenStackCloud(
89 cloud_config=osconfig.get_one_cloud())
90 self.__logger.debug("new cloud %s", self.cloud.auth)
93 """Remove projects/users"""
95 assert self.orig_cloud
97 assert self.project.id
98 self.orig_cloud.delete_user(self.user.id)
99 self.orig_cloud.delete_project(self.project.id)
101 self.orig_cloud.delete_role(self.role.id)
102 except Exception: # pylint: disable=broad-except
103 self.__logger.exception("Cannot clean all ressources")
106 class TenantNetwork1(testcase.TestCase):
107 # pylint: disable=too-many-instance-attributes
108 """Create a tenant network (scenario1)
110 It creates and configures all tenant network ressources required by
111 advanced testcases (subnet, network and router).
113 It ensures that all testcases inheriting from TenantNetwork1 could work
114 without network specific configurations (or at least read the same config
118 __logger = logging.getLogger(__name__)
119 cidr = '192.168.0.0/24'
120 shared_network = False
122 def __init__(self, **kwargs):
123 if "case_name" not in kwargs:
124 kwargs["case_name"] = 'tenantnetwork1'
125 super(TenantNetwork1, self).__init__(**kwargs)
126 self.res_dir = os.path.join(
127 getattr(config.CONF, 'dir_results'), self.case_name)
129 cloud_config = os_client_config.get_config()
130 self.cloud = shade.OpenStackCloud(cloud_config=cloud_config)
131 except Exception: # pylint: disable=broad-except
134 self.__logger.exception("Cannot connect to Cloud")
136 self.ext_net = self.get_external_network(self.cloud)
137 except Exception: # pylint: disable=broad-except
138 self.__logger.exception("Cannot get the external network")
139 self.guid = str(uuid.uuid4())
145 def get_external_network(cloud):
147 Returns the configured external network name or
148 the first retrieved external network name
151 if env.get("EXTERNAL_NETWORK"):
152 network = cloud.get_network(
153 env.get("EXTERNAL_NETWORK"), {"router:external": True})
156 networks = cloud.list_networks({"router:external": True})
161 def _create_network_ressources(self):
165 if hasattr(config.CONF, '{}_network_type'.format(self.case_name)):
166 provider["network_type"] = getattr(
167 config.CONF, '{}_network_type'.format(self.case_name))
168 if hasattr(config.CONF, '{}_physical_network'.format(self.case_name)):
169 provider["physical_network"] = getattr(
170 config.CONF, '{}_physical_network'.format(self.case_name))
171 if hasattr(config.CONF, '{}_segmentation_id'.format(self.case_name)):
172 provider["segmentation_id"] = getattr(
173 config.CONF, '{}_segmentation_id'.format(self.case_name))
174 self.network = self.cloud.create_network(
175 '{}-net_{}'.format(self.case_name, self.guid),
177 shared=self.shared_network)
178 self.__logger.debug("network: %s", self.network)
180 self.subnet = self.cloud.create_subnet(
182 subnet_name='{}-subnet_{}'.format(self.case_name, self.guid),
184 config.CONF, '{}_private_subnet_cidr'.format(self.case_name),
187 dns_nameservers=[env.get('NAMESERVER')])
188 self.__logger.debug("subnet: %s", self.subnet)
190 self.router = self.cloud.create_router(
191 name='{}-router_{}'.format(self.case_name, self.guid),
192 ext_gateway_net_id=self.ext_net.id)
193 self.__logger.debug("router: %s", self.router)
194 self.cloud.add_router_interface(self.router, subnet_id=self.subnet.id)
196 def run(self, **kwargs):
197 status = testcase.TestCase.EX_RUN_ERROR
200 self.start_time = time.time()
201 self._create_network_ressources()
203 status = testcase.TestCase.EX_OK
204 except Exception: # pylint: disable=broad-except
205 self.__logger.exception('Cannot run %s', self.case_name)
207 self.stop_time = time.time()
213 self.cloud.remove_router_interface(self.router, self.subnet.id)
214 self.cloud.delete_router(self.router.id)
215 self.cloud.delete_subnet(self.subnet.id)
216 self.cloud.delete_network(self.network.id)
217 except Exception: # pylint: disable=broad-except
218 self.__logger.exception("cannot clean all ressources")
221 class TenantNetwork2(TenantNetwork1):
222 """Create a tenant network (scenario2)
224 It creates new user/project before creating and configuring all tenant
225 network ressources required by a testcase (subnet, network and router).
227 It ensures that all testcases inheriting from TenantNetwork2 could work
228 without network specific configurations (or at least read the same config
232 __logger = logging.getLogger(__name__)
234 def __init__(self, **kwargs):
235 if "case_name" not in kwargs:
236 kwargs["case_name"] = 'tenantnetwork2'
237 super(TenantNetwork2, self).__init__(**kwargs)
240 self.project = NewProject(
241 self.cloud, self.case_name, self.guid)
242 self.project.create()
243 self.cloud = self.project.cloud
244 except Exception: # pylint: disable=broad-except
245 self.__logger.exception("Cannot create user or project")
251 super(TenantNetwork2, self).clean()
254 except Exception: # pylint: disable=broad-except
255 self.__logger.exception("Cannot clean all ressources")