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 assert self.orig_cloud.get_role(self.default_member)
72 except Exception: # pylint: disable=broad-except
73 self.__logger.info("Creating default role %s", self.default_member)
74 self.role = self.orig_cloud.create_role(self.default_member)
75 self.__logger.debug("role: %s", self.role)
76 self.orig_cloud.grant_role(
77 self.default_member, user=self.user.id, project=self.project.id,
78 domain=self.domain.id)
79 osconfig = os_client_config.config.OpenStackConfig()
80 osconfig.cloud_config[
81 'clouds']['envvars']['project_name'] = self.project.name
82 osconfig.cloud_config['clouds']['envvars']['username'] = self.user.name
83 osconfig.cloud_config['clouds']['envvars']['password'] = self.password
84 self.__logger.debug("cloud_config %s", osconfig.cloud_config)
85 self.cloud = shade.OpenStackCloud(
86 cloud_config=osconfig.get_one_cloud())
87 self.__logger.debug("new cloud %s", self.cloud.auth)
90 """Remove projects/users"""
92 assert self.orig_cloud
94 assert self.project.id
95 self.orig_cloud.delete_user(self.user.id)
96 self.orig_cloud.delete_project(self.project.id)
98 self.orig_cloud.delete_role(self.role.id)
99 except Exception: # pylint: disable=broad-except
100 self.__logger.exception("Cannot clean all ressources")
103 class TenantNetwork1(testcase.TestCase):
104 # pylint: disable=too-many-instance-attributes
105 """Create a tenant network (scenario1)
107 It creates and configures all tenant network ressources required by
108 advanced testcases (subnet, network and router).
110 It ensures that all testcases inheriting from TenantNetwork1 could work
111 without network specific configurations (or at least read the same config
115 __logger = logging.getLogger(__name__)
116 cidr = '192.168.0.0/24'
117 shared_network = False
119 def __init__(self, **kwargs):
120 if "case_name" not in kwargs:
121 kwargs["case_name"] = 'tenantnetwork1'
122 super(TenantNetwork1, self).__init__(**kwargs)
123 self.res_dir = os.path.join(
124 getattr(config.CONF, 'dir_results'), self.case_name)
126 cloud_config = os_client_config.get_config()
127 self.cloud = shade.OpenStackCloud(cloud_config=cloud_config)
128 except Exception: # pylint: disable=broad-except
131 self.__logger.exception("Cannot connect to Cloud")
133 self.ext_net = self.get_external_network(self.cloud)
134 except Exception: # pylint: disable=broad-except
135 self.__logger.exception("Cannot get the external network")
136 self.guid = str(uuid.uuid4())
142 def get_external_network(cloud):
144 Returns the configured external network name or
145 the first retrieved external network name
148 if env.get("EXTERNAL_NETWORK"):
149 network = cloud.get_network(
150 env.get("EXTERNAL_NETWORK"), {"router:external": True})
153 networks = cloud.list_networks({"router:external": True})
158 def _create_network_ressources(self):
162 if hasattr(config.CONF, '{}_network_type'.format(self.case_name)):
163 provider["network_type"] = getattr(
164 config.CONF, '{}_network_type'.format(self.case_name))
165 if hasattr(config.CONF, '{}_physical_network'.format(self.case_name)):
166 provider["physical_network"] = getattr(
167 config.CONF, '{}_physical_network'.format(self.case_name))
168 if hasattr(config.CONF, '{}_segmentation_id'.format(self.case_name)):
169 provider["segmentation_id"] = getattr(
170 config.CONF, '{}_segmentation_id'.format(self.case_name))
171 self.network = self.cloud.create_network(
172 '{}-net_{}'.format(self.case_name, self.guid),
174 shared=self.shared_network)
175 self.__logger.debug("network: %s", self.network)
177 self.subnet = self.cloud.create_subnet(
179 subnet_name='{}-subnet_{}'.format(self.case_name, self.guid),
181 config.CONF, '{}_private_subnet_cidr'.format(self.case_name),
184 dns_nameservers=[env.get('NAMESERVER')])
185 self.__logger.debug("subnet: %s", self.subnet)
187 self.router = self.cloud.create_router(
188 name='{}-router_{}'.format(self.case_name, self.guid),
189 ext_gateway_net_id=self.ext_net.id)
190 self.__logger.debug("router: %s", self.router)
191 self.cloud.add_router_interface(self.router, subnet_id=self.subnet.id)
193 def run(self, **kwargs):
194 status = testcase.TestCase.EX_RUN_ERROR
197 self.start_time = time.time()
198 self._create_network_ressources()
200 status = testcase.TestCase.EX_OK
201 except Exception: # pylint: disable=broad-except
202 self.__logger.exception('Cannot run %s', self.case_name)
204 self.stop_time = time.time()
210 self.cloud.remove_router_interface(self.router, self.subnet.id)
211 self.cloud.delete_router(self.router.id)
212 self.cloud.delete_subnet(self.subnet.id)
213 self.cloud.delete_network(self.network.id)
214 except Exception: # pylint: disable=broad-except
215 self.__logger.exception("cannot clean all ressources")
218 class TenantNetwork2(TenantNetwork1):
219 """Create a tenant network (scenario2)
221 It creates new user/project before creating and configuring all tenant
222 network ressources required by a testcase (subnet, network and router).
224 It ensures that all testcases inheriting from TenantNetwork2 could work
225 without network specific configurations (or at least read the same config
229 __logger = logging.getLogger(__name__)
231 def __init__(self, **kwargs):
232 if "case_name" not in kwargs:
233 kwargs["case_name"] = 'tenantnetwork2'
234 super(TenantNetwork2, self).__init__(**kwargs)
237 self.project = NewProject(
238 self.cloud, self.case_name, self.guid)
239 self.project.create()
240 self.cloud = self.project.cloud
241 except Exception: # pylint: disable=broad-except
242 self.__logger.exception("Cannot create user or project")
248 super(TenantNetwork2, self).clean()
251 except Exception: # pylint: disable=broad-except
252 self.__logger.exception("Cannot clean all ressources")