projects / osm / NBI.git / blob
? search:


7c172d793a0a62faeac4094d7dbd6040838b5a44
[osm/NBI.git] / osm_nbi / tests / test.py
1 #! /usr/bin/python3
2 # -*- coding: utf-8 -*-
3
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
7 #
8 # http://www.apache.org/licenses/LICENSE-2.0
9 #
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
13 # implied.
14 # See the License for the specific language governing permissions and
15 # limitations under the License.
16
17 import getopt
18 import sys
19 import requests
20 import json
21 import logging
22 import yaml
23 # import json
24 # import tarfile
25 from time import sleep
26 import os
27
28 __author__ = "Alfonso Tierno, alfonso.tiernosepulveda@telefonica.com"
29 __date__ = "$2018-03-01$"
30 __version__ = "0.3"
31 version_date = "Oct 2018"
32
33
34 def usage():
35 print("Usage: ", sys.argv[0], "[options]")
36 print(" Performs system tests over running NBI. It can be used for real OSM test using option '--test-osm'")
37 print(" If this is the case env variables 'OSMNBITEST_VIM_NAME' must be suplied to create a VIM if not exist "
38 "where deployment is done")
39 print("OPTIONS")
40 print(" -h|--help: shows this help")
41 print(" --insecure: Allows non trusted https NBI server")
42 print(" --list: list available tests")
43 print(" --manual-check: Deployment tests stop after deployed to allow manual inspection. Only make sense with "
44 "'--test-osm'")
45 print(" -p|--password PASSWORD: NBI access password. 'admin' by default")
46 print(" ---project PROJECT: NBI access project. 'admin' by default")
47 print(" --test TEST[,...]: Execute only a test or a comma separated list of tests")
48 print(" --params key=val: params to the previous test. key can be vnfd-files, nsd-file, ns-name, ns-config")
49 print(" --test-osm: If missing this test is intended for NBI only, no other OSM components are expected. Use "
50 "this flag to test the system. LCM and RO components are expected to be up and running")
51 print(" --timeout TIMEOUT: General NBI timeout, by default {}s".format(timeout))
52 print(" --timeout-deploy TIMEOUT: Timeout used for getting NS deployed, by default {}s".format(timeout_deploy))
53 print(" --timeout-configure TIMEOUT: Timeout used for getting NS deployed and configured,"
54 " by default {}s".format(timeout_configure))
55 print(" -u|--user USERNAME: NBI access username. 'admin' by default")
56 print(" --url URL: complete NBI server URL. 'https//localhost:9999/osm' by default")
57 print(" -v|--verbose print debug information, can be used several times")
58 print(" --no-verbose remove verbosity")
59 print(" --version: prints current version")
60 print("ENV variables used for real deployment tests with option osm-test.")
61 print(" export OSMNBITEST_VIM_NAME=vim-name")
62 print(" export OSMNBITEST_VIM_URL=vim-url")
63 print(" export OSMNBITEST_VIM_TYPE=vim-type")
64 print(" export OSMNBITEST_VIM_TENANT=vim-tenant")
65 print(" export OSMNBITEST_VIM_USER=vim-user")
66 print(" export OSMNBITEST_VIM_PASSWORD=vim-password")
67 print(" export OSMNBITEST_VIM_CONFIG=\"vim-config\"")
68 print(" export OSMNBITEST_NS_NAME=\"vim-config\"")
69 return
70
71
72 r_header_json = {"Content-type": "application/json"}
73 headers_json = {
74 "Content-type": "application/json",
75 "Accept": "application/json",
76 }
77 r_header_yaml = {"Content-type": "application/yaml"}
78 headers_yaml = {
79 "Content-type": "application/yaml",
80 "Accept": "application/yaml",
81 }
82 r_header_text = {"Content-type": "text/plain"}
83 r_header_octect = {"Content-type": "application/octet-stream"}
84 headers_text = {
85 "Accept": "text/plain",
86 }
87 r_header_zip = {"Content-type": "application/zip"}
88 headers_zip = {
89 "Accept": "application/zip",
90 }
91 headers_zip_yaml = {
92 "Accept": "application/yaml", "Content-type": "application/zip"
93 }
94
95
96 # test ones authorized
97 test_authorized_list = (
98 ("AU1", "Invalid vnfd id", "GET", "/vnfpkgm/v1/vnf_packages/non-existing-id",
99 headers_json, None, 404, r_header_json, "json"),
100 ("AU2", "Invalid nsd id", "GET", "/nsd/v1/ns_descriptors/non-existing-id",
101 headers_yaml, None, 404, r_header_yaml, "yaml"),
102 ("AU3", "Invalid nsd id", "DELETE", "/nsd/v1/ns_descriptors_content/non-existing-id",
103 headers_yaml, None, 404, r_header_yaml, "yaml"),
104 )
105 timeout = 120 # general timeout
106 timeout_deploy = 60*10 # timeout for NS deploying without charms
107 timeout_configure = 60*20 # timeout for NS deploying and configuring
108
109
110 class TestException(Exception):
111 pass
112
113
114 class TestRest:
115 def __init__(self, url_base, header_base=None, verify=False, user="admin", password="admin", project="admin"):
116 self.url_base = url_base
117 if header_base is None:
118 self.header_base = {}
119 else:
120 self.header_base = header_base.copy()
121 self.s = requests.session()
122 self.s.headers = self.header_base
123 self.verify = verify
124 self.token = False
125 self.user = user
126 self.password = password
127 self.project = project
128 self.vim_id = None
129 # contains ID of tests obtained from Location response header. "" key contains last obtained id
130 self.test_ids = {}
131 self.old_test_description = ""
132 self.test_name = None
133 self.step = 0
134
135 def set_header(self, header):
136 self.s.headers.update(header)
137
138 def set_tet_name(self, test_name):
139 self.test_name = test_name
140
141 def unset_header(self, key):
142 if key in self.s.headers:
143 del self.s.headers[key]
144
145 def test(self, name, description, method, url, headers, payload, expected_codes, expected_headers,
146 expected_payload, store_file=None):
147 """
148 Performs an http request and check http code response. Exit if different than allowed. It get the returned id
149 that can be used by following test in the URL with {name} where name is the name of the test
150 :param name: short name of the test
151 :param description: description of the test
152 :param method: HTTP method: GET,PUT,POST,DELETE,...
153 :param url: complete URL or relative URL
154 :param headers: request headers to add to the base headers
155 :param payload: Can be a dict, transformed to json, a text or a file if starts with '@'
156 :param expected_codes: expected response codes, can be int, int tuple or int range
157 :param expected_headers: expected response headers, dict with key values
158 :param expected_payload: expected payload, 0 if empty, 'yaml', 'json', 'text', 'zip', 'octet-stream'
159 :param store_file: filename to store content
160 :return: requests response
161 """
162 r = None
163 try:
164 if not self.s:
165 self.s = requests.session()
166 # URL
167 if not url:
168 url = self.url_base
169 elif not url.startswith("http"):
170 url = self.url_base + url
171
172 var_start = url.find("<") + 1
173 while var_start:
174 var_end = url.find(">", var_start)
175 if var_end == -1:
176 break
177 var_name = url[var_start:var_end]
178 if var_name in self.test_ids:
179 url = url[:var_start-1] + self.test_ids[var_name] + url[var_end+1:]
180 var_start += len(self.test_ids[var_name])
181 var_start = url.find("<", var_start) + 1
182 if payload:
183 if isinstance(payload, str):
184 if payload.startswith("@"):
185 mode = "r"
186 file_name = payload[1:]
187 if payload.startswith("@b"):
188 mode = "rb"
189 file_name = payload[2:]
190 with open(file_name, mode) as f:
191 payload = f.read()
192 elif isinstance(payload, dict):
193 payload = json.dumps(payload)
194
195 test_description = "Test {} {} {} {}".format(name, description, method, url)
196 if self.old_test_description != test_description:
197 self.old_test_description = test_description
198 logger.warning(test_description)
199 stream = False
200 if expected_payload in ("zip", "octet-string") or store_file:
201 stream = True
202 r = getattr(self.s, method.lower())(url, data=payload, headers=headers, verify=self.verify, stream=stream)
203 if expected_payload in ("zip", "octet-string") or store_file:
204 logger.debug("RX {}".format(r.status_code))
205 else:
206 logger.debug("RX {}: {}".format(r.status_code, r.text))
207
208 # check response
209 if expected_codes:
210 if isinstance(expected_codes, int):
211 expected_codes = (expected_codes,)
212 if r.status_code not in expected_codes:
213 raise TestException(
214 "Got status {}. Expected {}. {}".format(r.status_code, expected_codes, r.text))
215
216 if expected_headers:
217 for header_key, header_val in expected_headers.items():
218 if header_key.lower() not in r.headers:
219 raise TestException("Header {} not present".format(header_key))
220 if header_val and header_val.lower() not in r.headers[header_key]:
221 raise TestException("Header {} does not contain {} but {}".format(header_key, header_val,
222 r.headers[header_key]))
223
224 if expected_payload is not None:
225 if expected_payload == 0 and len(r.content) > 0:
226 raise TestException("Expected empty payload")
227 elif expected_payload == "json":
228 try:
229 r.json()
230 except Exception as e:
231 raise TestException("Expected json response payload, but got Exception {}".format(e))
232 elif expected_payload == "yaml":
233 try:
234 yaml.safe_load(r.text)
235 except Exception as e:
236 raise TestException("Expected yaml response payload, but got Exception {}".format(e))
237 elif expected_payload in ("zip", "octet-string"):
238 if len(r.content) == 0:
239 raise TestException("Expected some response payload, but got empty")
240 # try:
241 # tar = tarfile.open(None, 'r:gz', fileobj=r.raw)
242 # for tarinfo in tar:
243 # tarname = tarinfo.name
244 # print(tarname)
245 # except Exception as e:
246 # raise TestException("Expected zip response payload, but got Exception {}".format(e))
247 elif expected_payload == "text":
248 if len(r.content) == 0:
249 raise TestException("Expected some response payload, but got empty")
250 # r.text
251 if store_file:
252 with open(store_file, 'wb') as fd:
253 for chunk in r.iter_content(chunk_size=128):
254 fd.write(chunk)
255
256 location = r.headers.get("Location")
257 if location:
258 _id = location[location.rfind("/") + 1:]
259 if _id:
260 self.test_ids[name] = str(_id)
261 self.test_ids["last_id"] = str(_id) # last id
262 self.test_ids[""] = str(_id) # last id
263 return r
264 except TestException as e:
265 r_status_code = None
266 r_text = None
267 if r:
268 r_status_code = r.status_code
269 r_text = r.text
270 logger.error("{} \nRX code{}: {}".format(e, r_status_code, r_text))
271 exit(1)
272 except IOError as e:
273 logger.error("Cannot open file {}".format(e))
274 exit(1)
275
276 def get_autorization(self): # user=None, password=None, project=None):
277 if self.token: # and self.user == user and self.password == password and self.project == project:
278 return
279 # self.user = user
280 # self.password = password
281 # self.project = project
282 r = self.test("TOKEN", "Obtain token", "POST", "/admin/v1/tokens", headers_json,
283 {"username": self.user, "password": self.password, "project_id": self.project},
284 (200, 201), r_header_json, "json")
285 response = r.json()
286 self.token = response["id"]
287 self.set_header({"Authorization": "Bearer {}".format(self.token)})
288
289 def remove_authorization(self):
290 if self.token:
291 self.test("TOKEN_DEL", "Delete token", "DELETE", "/admin/v1/tokens/{}".format(self.token), headers_json,
292 None, (200, 201, 204), None, None)
293 self.token = None
294 self.unset_header("Authorization")
295
296 def get_create_vim(self, test_osm):
297 if self.vim_id:
298 return self.vim_id
299 self.get_autorization()
300 if test_osm:
301 vim_name = os.environ.get("OSMNBITEST_VIM_NAME")
302 if not vim_name:
303 raise TestException(
304 "Needed to define OSMNBITEST_VIM_XXX variables to create a real VIM for deployment")
305 else:
306 vim_name = "fakeVim"
307 # Get VIM
308 r = self.test("_VIMGET1", "Get VIM ID", "GET", "/admin/v1/vim_accounts?name={}".format(vim_name), headers_json,
309 None, 200, r_header_json, "json")
310 vims = r.json()
311 if vims:
312 return vims[0]["_id"]
313 # Add VIM
314 if test_osm:
315 # check needed environ parameters:
316 if not os.environ.get("OSMNBITEST_VIM_URL") or not os.environ.get("OSMNBITEST_VIM_TENANT"):
317 raise TestException("Env OSMNBITEST_VIM_URL and OSMNBITEST_VIM_TENANT are needed for create a real VIM"
318 " to deploy on whit the --test-osm option")
319 vim_data = "{{schema_version: '1.0', name: '{}', vim_type: {}, vim_url: '{}', vim_tenant_name: '{}', "\
320 "vim_user: {}, vim_password: {}".format(vim_name,
321 os.environ.get("OSMNBITEST_VIM_TYPE", "openstack"),
322 os.environ.get("OSMNBITEST_VIM_URL"),
323 os.environ.get("OSMNBITEST_VIM_TENANT"),
324 os.environ.get("OSMNBITEST_VIM_USER"),
325 os.environ.get("OSMNBITEST_VIM_PASSWORD"))
326 if os.environ.get("OSMNBITEST_VIM_CONFIG"):
327 vim_data += " ,config: {}".format(os.environ.get("OSMNBITEST_VIM_CONFIG"))
328 vim_data += "}"
329 else:
330 vim_data = "{schema_version: '1.0', name: fakeVim, vim_type: openstack, vim_url: 'http://10.11.12.13/fake'"\
331 ", vim_tenant_name: 'vimtenant', vim_user: vimuser, vim_password: vimpassword}"
332 r = self.test("_VIMGET2", "Create VIM", "POST", "/admin/v1/vim_accounts", headers_yaml, vim_data,
333 (201), {"Location": "/admin/v1/vim_accounts/", "Content-Type": "application/yaml"}, "yaml")
334 location = r.headers.get("Location")
335 return location[location.rfind("/") + 1:]
336
337
338 class TestNonAuthorized:
339 description = "test invalid URLs. methods and no authorization"
340
341 @staticmethod
342 def run(engine, test_osm, manual_check, test_params=None):
343 engine.remove_authorization()
344 test_not_authorized_list = (
345 ("NA1", "Invalid token", "GET", "/admin/v1/users", headers_json, None, 401, r_header_json, "json"),
346 ("NA2", "Invalid URL", "POST", "/admin/v1/nonexist", headers_yaml, None, 405, r_header_yaml, "yaml"),
347 ("NA3", "Invalid version", "DELETE", "/admin/v2/users", headers_yaml, None, 405, r_header_yaml, "yaml"),
348 )
349 for t in test_not_authorized_list:
350 engine.test(*t)
351
352
353 class TestUsersProjects:
354 description = "test project and user creation"
355
356 @staticmethod
357 def run(engine, test_osm, manual_check, test_params=None):
358 engine.get_autorization()
359 engine.test("PU1", "Create project non admin", "POST", "/admin/v1/projects", headers_json, {"name": "P1"},
360 (201, 204), {"Location": "/admin/v1/projects/", "Content-Type": "application/json"}, "json")
361 engine.test("PU2", "Create project admin", "POST", "/admin/v1/projects", headers_json,
362 {"name": "Padmin", "admin": True}, (201, 204),
363 {"Location": "/admin/v1/projects/", "Content-Type": "application/json"}, "json")
364 engine.test("PU3", "Create project bad format", "POST", "/admin/v1/projects", headers_json, {"name": 1}, 422,
365 r_header_json, "json")
366 engine.test("PU4", "Create user with bad project", "POST", "/admin/v1/users", headers_json,
367 {"username": "U1", "projects": ["P1", "P2", "Padmin"], "password": "pw1"}, 409,
368 r_header_json, "json")
369 engine.test("PU5", "Create user with bad project and force", "POST", "/admin/v1/users?FORCE=True", headers_json,
370 {"username": "U1", "projects": ["P1", "P2", "Padmin"], "password": "pw1"}, 201,
371 {"Location": "/admin/v1/users/", "Content-Type": "application/json"}, "json")
372 engine.test("PU6", "Create user 2", "POST", "/admin/v1/users", headers_json,
373 {"username": "U2", "projects": ["P1"], "password": "pw2"}, 201,
374 {"Location": "/admin/v1/users/", "Content-Type": "application/json"}, "json")
375
376 engine.test("PU7", "Edit user U1, delete P2 project", "PATCH", "/admin/v1/users/U1", headers_json,
377 {"projects": {"$'P2'": None}}, 204, None, None)
378 res = engine.test("PU1", "Check user U1, contains the right projects", "GET", "/admin/v1/users/U1",
379 headers_json, None, 200, None, json)
380 u1 = res.json()
381 # print(u1)
382 expected_projects = ["P1", "Padmin"]
383 if u1["projects"] != expected_projects:
384 raise TestException("User content projects '{}' different than expected '{}'. Edition has not done"
385 " properly".format(u1["projects"], expected_projects))
386
387 engine.test("PU8", "Edit user U1, set Padmin as default project", "PUT", "/admin/v1/users/U1", headers_json,
388 {"projects": {"$'Padmin'": None, "$+[0]": "Padmin"}}, 204, None, None)
389 res = engine.test("PU1", "Check user U1, contains the right projects", "GET", "/admin/v1/users/U1",
390 headers_json, None, 200, None, json)
391 u1 = res.json()
392 # print(u1)
393 expected_projects = ["Padmin", "P1"]
394 if u1["projects"] != expected_projects:
395 raise TestException("User content projects '{}' different than expected '{}'. Edition has not done"
396 " properly".format(u1["projects"], expected_projects))
397
398 engine.test("PU9", "Edit user U1, change password", "PATCH", "/admin/v1/users/U1", headers_json,
399 {"password": "pw1_new"}, 204, None, None)
400
401 engine.test("PU10", "Change to project P1 non existing", "POST", "/admin/v1/tokens/", headers_json,
402 {"project_id": "P1"}, 401, r_header_json, "json")
403
404 res = engine.test("PU1", "Change to user U1 project P1", "POST", "/admin/v1/tokens", headers_json,
405 {"username": "U1", "password": "pw1_new", "project_id": "P1"}, (200, 201),
406 r_header_json, "json")
407 response = res.json()
408 engine.set_header({"Authorization": "Bearer {}".format(response["id"])})
409
410 engine.test("PU11", "Edit user projects non admin", "PUT", "/admin/v1/users/U1", headers_json,
411 {"projects": {"$'P1'": None}}, 401, r_header_json, "json")
412 engine.test("PU12", "Add new project non admin", "POST", "/admin/v1/projects", headers_json,
413 {"name": "P2"}, 401, r_header_json, "json")
414 engine.test("PU13", "Add new user non admin", "POST", "/admin/v1/users", headers_json,
415 {"username": "U3", "projects": ["P1"], "password": "pw3"}, 401,
416 r_header_json, "json")
417
418 res = engine.test("PU14", "Change to user U1 project Padmin", "POST", "/admin/v1/tokens", headers_json,
419 {"project_id": "Padmin"}, (200, 201), r_header_json, "json")
420 response = res.json()
421 engine.set_header({"Authorization": "Bearer {}".format(response["id"])})
422
423 engine.test("PU15", "Add new project admin", "POST", "/admin/v1/projects", headers_json, {"name": "P2"},
424 (201, 204), {"Location": "/admin/v1/projects/", "Content-Type": "application/json"}, "json")
425 engine.test("PU16", "Add new user U3 admin", "POST", "/admin/v1/users",
426 headers_json, {"username": "U3", "projects": ["P2"], "password": "pw3"}, (201, 204),
427 {"Location": "/admin/v1/users/", "Content-Type": "application/json"}, "json")
428 engine.test("PU17", "Edit user projects admin", "PUT", "/admin/v1/users/U3", headers_json,
429 {"projects": ["P2"]}, 204, None, None)
430
431 engine.test("PU18", "Delete project P2 conflict", "DELETE", "/admin/v1/projects/P2", headers_json, None, 409,
432 r_header_json, "json")
433 engine.test("PU19", "Delete project P2 forcing", "DELETE", "/admin/v1/projects/P2?FORCE=True", headers_json,
434 None, 204, None, None)
435
436 engine.test("PU20", "Delete user U1. Conflict deleting own user", "DELETE", "/admin/v1/users/U1", headers_json,
437 None, 409, r_header_json, "json")
438 engine.test("PU21", "Delete user U2", "DELETE", "/admin/v1/users/U2", headers_json, None, 204, None, None)
439 engine.test("PU22", "Delete user U3", "DELETE", "/admin/v1/users/U3", headers_json, None, 204, None, None)
440 # change to admin
441 engine.remove_authorization() # To force get authorization
442 engine.get_autorization()
443 engine.test("PU23", "Delete user U1", "DELETE", "/admin/v1/users/U1", headers_json, None, 204, None, None)
444 engine.test("PU24", "Delete project P1", "DELETE", "/admin/v1/projects/P1", headers_json, None, 204, None, None)
445 engine.test("PU25", "Delete project Padmin", "DELETE", "/admin/v1/projects/Padmin", headers_json, None, 204,
446 None, None)
447
448
449 class TestFakeVim:
450 description = "Creates/edit/delete fake VIMs and SDN controllers"
451
452 def __init__(self):
453 self.vim = {
454 "schema_version": "1.0",
455 "schema_type": "No idea",
456 "name": "myVim",
457 "description": "Descriptor name",
458 "vim_type": "openstack",
459 "vim_url": "http://localhost:/vim",
460 "vim_tenant_name": "vimTenant",
461 "vim_user": "user",
462 "vim_password": "password",
463 "config": {"config_param": 1}
464 }
465 self.sdn = {
466 "name": "sdn-name",
467 "description": "sdn-description",
468 "dpid": "50:50:52:54:00:94:21:21",
469 "ip": "192.168.15.17",
470 "port": 8080,
471 "type": "opendaylight",
472 "version": "3.5.6",
473 "user": "user",
474 "password": "passwd"
475 }
476 self.port_mapping = [
477 {"compute_node": "compute node 1",
478 "ports": [{"pci": "0000:81:00.0", "switch_port": "port-2/1", "switch_mac": "52:54:00:94:21:21"},
479 {"pci": "0000:81:00.1", "switch_port": "port-2/2", "switch_mac": "52:54:00:94:21:22"}
480 ]},
481 {"compute_node": "compute node 2",
482 "ports": [{"pci": "0000:81:00.0", "switch_port": "port-2/3", "switch_mac": "52:54:00:94:21:23"},
483 {"pci": "0000:81:00.1", "switch_port": "port-2/4", "switch_mac": "52:54:00:94:21:24"}
484 ]}
485 ]
486
487 def run(self, engine, test_osm, manual_check, test_params=None):
488
489 vim_bad = self.vim.copy()
490 vim_bad.pop("name")
491
492 engine.get_autorization()
493 engine.test("FVIM1", "Create VIM", "POST", "/admin/v1/vim_accounts", headers_json, self.vim, (201, 204),
494 {"Location": "/admin/v1/vim_accounts/", "Content-Type": "application/json"}, "json")
495 engine.test("FVIM2", "Create VIM without name, bad schema", "POST", "/admin/v1/vim_accounts", headers_json,
496 vim_bad, 422, None, headers_json)
497 engine.test("FVIM3", "Create VIM name repeated", "POST", "/admin/v1/vim_accounts", headers_json, self.vim,
498 409, None, headers_json)
499 engine.test("FVIM4", "Show VIMs", "GET", "/admin/v1/vim_accounts", headers_yaml, None, 200, r_header_yaml,
500 "yaml")
501 engine.test("FVIM5", "Show VIM", "GET", "/admin/v1/vim_accounts/<FVIM1>", headers_yaml, None, 200,
502 r_header_yaml, "yaml")
503 if not test_osm:
504 # delete with FORCE
505 engine.test("FVIM6", "Delete VIM", "DELETE", "/admin/v1/vim_accounts/<FVIM1>?FORCE=True", headers_yaml,
506 None, 202, None, 0)
507 engine.test("FVIM7", "Check VIM is deleted", "GET", "/admin/v1/vim_accounts/<FVIM1>", headers_yaml, None,
508 404, r_header_yaml, "yaml")
509 else:
510 # delete and wait until is really deleted
511 engine.test("FVIM6", "Delete VIM", "DELETE", "/admin/v1/vim_accounts/<FVIM1>", headers_yaml, None, 202,
512 None, 0)
513 wait = timeout
514 while wait >= 0:
515 r = engine.test("FVIM7", "Check VIM is deleted", "GET", "/admin/v1/vim_accounts/<FVIM1>", headers_yaml,
516 None, None, r_header_yaml, "yaml")
517 if r.status_code == 404:
518 break
519 elif r.status_code == 200:
520 wait -= 5
521 sleep(5)
522 else:
523 raise TestException("Vim created at 'FVIM1' is not delete after {} seconds".format(timeout))
524
525
526 class TestVIMSDN(TestFakeVim):
527 description = "Creates VIM with SDN editing SDN controllers and port_mapping"
528
529 def __init__(self):
530 TestFakeVim.__init__(self)
531
532 def run(self, engine, test_osm, manual_check, test_params=None):
533 engine.get_autorization()
534 # Added SDN
535 engine.test("VIMSDN1", "Create SDN", "POST", "/admin/v1/sdns", headers_json, self.sdn, (201, 204),
536 {"Location": "/admin/v1/sdns/", "Content-Type": "application/json"}, "json")
537 # sleep(5)
538 # Edit SDN
539 engine.test("VIMSDN2", "Edit SDN", "PATCH", "/admin/v1/sdns/<VIMSDN1>", headers_json, {"name": "new_sdn_name"},
540 204, None, None)
541 # sleep(5)
542 # VIM with SDN
543 self.vim["config"]["sdn-controller"] = engine.test_ids["VIMSDN1"]
544 self.vim["config"]["sdn-port-mapping"] = self.port_mapping
545 engine.test("VIMSDN3", "Create VIM", "POST", "/admin/v1/vim_accounts", headers_json, self.vim, (200, 204, 201),
546 {"Location": "/admin/v1/vim_accounts/", "Content-Type": "application/json"}, "json"),
547
548 self.port_mapping[0]["compute_node"] = "compute node XX"
549 engine.test("VIMSDN4", "Edit VIM change port-mapping", "PUT", "/admin/v1/vim_accounts/<VIMSDN3>", headers_json,
550 {"config": {"sdn-port-mapping": self.port_mapping}}, 204, None, None)
551 engine.test("VIMSDN5", "Edit VIM remove port-mapping", "PUT", "/admin/v1/vim_accounts/<VIMSDN3>", headers_json,
552 {"config": {"sdn-port-mapping": None}}, 204, None, None)
553
554 if not test_osm:
555 # delete with FORCE
556 engine.test("VIMSDN6", "Delete VIM remove port-mapping", "DELETE",
557 "/admin/v1/vim_accounts/<VIMSDN3>?FORCE=True", headers_json, None, 202, None, 0)
558 engine.test("VIMSDN7", "Delete SDNC", "DELETE", "/admin/v1/sdns/<VIMSDN1>?FORCE=True", headers_json, None,
559 202, None, 0)
560
561 engine.test("VIMSDN8", "Check VIM is deleted", "GET", "/admin/v1/vim_accounts/<VIMSDN3>", headers_yaml,
562 None, 404, r_header_yaml, "yaml")
563 engine.test("VIMSDN9", "Check SDN is deleted", "GET", "/admin/v1/sdns/<VIMSDN1>", headers_yaml, None,
564 404, r_header_yaml, "yaml")
565 else:
566 # delete and wait until is really deleted
567 engine.test("VIMSDN6", "Delete VIM remove port-mapping", "DELETE", "/admin/v1/vim_accounts/<VIMSDN3>",
568 headers_json, None, (202, 201, 204), None, 0)
569 engine.test("VIMSDN7", "Delete SDN", "DELETE", "/admin/v1/sdns/<VIMSDN1>", headers_json, None,
570 (202, 201, 204), None, 0)
571 wait = timeout
572 while wait >= 0:
573 r = engine.test("VIMSDN8", "Check VIM is deleted", "GET", "/admin/v1/vim_accounts/<VIMSDN3>",
574 headers_yaml, None, None, r_header_yaml, "yaml")
575 if r.status_code == 404:
576 break
577 elif r.status_code == 200:
578 wait -= 5
579 sleep(5)
580 else:
581 raise TestException("Vim created at 'VIMSDN3' is not delete after {} seconds".format(timeout))
582 while wait >= 0:
583 r = engine.test("VIMSDN9", "Check SDNC is deleted", "GET", "/admin/v1/sdns/<VIMSDN1>",
584 headers_yaml, None, None, r_header_yaml, "yaml")
585 if r.status_code == 404:
586 break
587 elif r.status_code == 200:
588 wait -= 5
589 sleep(5)
590 else:
591 raise TestException("SDNC created at 'VIMSDN1' is not delete after {} seconds".format(timeout))
592
593
594 class TestDeploy:
595 description = "Base class for downloading descriptors from ETSI, onboard and deploy in real VIM"
596
597 def __init__(self):
598 self.step = 0
599 self.nsd_id = None
600 self.vim_id = None
601 self.nsd_test = None
602 self.ns_test = None
603 self.ns_id = None
604 self.vnfds_test = []
605 self.vnfds_id = []
606 self.descriptor_url = "https://osm-download.etsi.org/ftp/osm-3.0-three/2nd-hackfest/packages/"
607 self.vnfd_filenames = ("cirros_vnf.tar.gz",)
608 self.nsd_filename = "cirros_2vnf_ns.tar.gz"
609 self.descriptor_edit = None
610 self.uses_configuration = False
611 self.uss = {}
612 self.passwds = {}
613 self.cmds = {}
614 self.keys = {}
615 self.timeout = 120
616 self.passed_tests = 0
617 self.total_tests = 0
618 self.qforce = ""
619
620 def create_descriptors(self, engine):
621 temp_dir = os.path.dirname(os.path.abspath(__file__)) + "/temp/"
622 if not os.path.exists(temp_dir):
623 os.makedirs(temp_dir)
624 for vnfd_index, vnfd_filename in enumerate(self.vnfd_filenames):
625 if "/" in vnfd_filename:
626 vnfd_filename_path = vnfd_filename
627 if not os.path.exists(vnfd_filename_path):
628 raise TestException("File '{}' does not exist".format(vnfd_filename_path))
629 else:
630 vnfd_filename_path = temp_dir + vnfd_filename
631 if not os.path.exists(vnfd_filename_path):
632 with open(vnfd_filename_path, "wb") as file:
633 response = requests.get(self.descriptor_url + vnfd_filename)
634 if response.status_code >= 300:
635 raise TestException("Error downloading descriptor from '{}': {}".format(
636 self.descriptor_url + vnfd_filename, response.status_code))
637 file.write(response.content)
638 if vnfd_filename_path.endswith(".yaml"):
639 headers = headers_yaml
640 else:
641 headers = headers_zip_yaml
642 if self.step % 2 == 0:
643 # vnfd CREATE AND UPLOAD in one step:
644 test_name = "DEPLOY{}".format(self.step)
645 engine.test(test_name, "Onboard VNFD in one step", "POST",
646 "/vnfpkgm/v1/vnf_packages_content" + self.qforce, headers, "@b" + vnfd_filename_path, 201,
647 {"Location": "/vnfpkgm/v1/vnf_packages_content/", "Content-Type": "application/yaml"},
648 "yaml")
649 self.vnfds_test.append(test_name)
650 self.vnfds_id.append(engine.test_ids["last_id"])
651 self.step += 1
652 else:
653 # vnfd CREATE AND UPLOAD ZIP
654 test_name = "DEPLOY{}".format(self.step)
655 engine.test(test_name, "Onboard VNFD step 1", "POST", "/vnfpkgm/v1/vnf_packages",
656 headers_json, None, 201,
657 {"Location": "/vnfpkgm/v1/vnf_packages/", "Content-Type": "application/json"}, "json")
658 self.vnfds_test.append(test_name)
659 self.vnfds_id.append(engine.test_ids["last_id"])
660 self.step += 1
661 # location = r.headers["Location"]
662 # vnfd_id = location[location.rfind("/")+1:]
663 engine.test("DEPLOY{}".format(self.step), "Onboard VNFD step 2 as ZIP", "PUT",
664 "/vnfpkgm/v1/vnf_packages/<>/package_content" + self.qforce,
665 headers, "@b" + vnfd_filename_path, 204, None, 0)
666 self.step += 2
667
668 if self.descriptor_edit:
669 if "vnfd{}".format(vnfd_index) in self.descriptor_edit:
670 # Modify VNFD
671 engine.test("DEPLOY{}".format(self.step), "Edit VNFD ", "PATCH",
672 "/vnfpkgm/v1/vnf_packages/{}".format(self.vnfds_id[-1]),
673 headers_yaml, self.descriptor_edit["vnfd{}".format(vnfd_index)], 204, None, None)
674 self.step += 1
675
676 if "/" in self.nsd_filename:
677 nsd_filename_path = self.nsd_filename
678 if not os.path.exists(nsd_filename_path):
679 raise TestException("File '{}' does not exist".format(nsd_filename_path))
680 else:
681 nsd_filename_path = temp_dir + self.nsd_filename
682 if not os.path.exists(nsd_filename_path):
683 with open(nsd_filename_path, "wb") as file:
684 response = requests.get(self.descriptor_url + self.nsd_filename)
685 if response.status_code >= 300:
686 raise TestException("Error downloading descriptor from '{}': {}".format(
687 self.descriptor_url + self.nsd_filename, response.status_code))
688 file.write(response.content)
689 if nsd_filename_path.endswith(".yaml"):
690 headers = headers_yaml
691 else:
692 headers = headers_zip_yaml
693
694 self.nsd_test = "DEPLOY" + str(self.step)
695 if self.step % 2 == 0:
696 # nsd CREATE AND UPLOAD in one step:
697 engine.test("DEPLOY{}".format(self.step), "Onboard NSD in one step", "POST",
698 "/nsd/v1/ns_descriptors_content" + self.qforce, headers, "@b" + nsd_filename_path, 201,
699 {"Location": "/nsd/v1/ns_descriptors_content/", "Content-Type": "application/yaml"}, yaml)
700 self.step += 1
701 self.nsd_id = engine.test_ids["last_id"]
702 else:
703 # nsd CREATE AND UPLOAD ZIP
704 engine.test("DEPLOY{}".format(self.step), "Onboard NSD step 1", "POST", "/nsd/v1/ns_descriptors",
705 headers_json, None, 201,
706 {"Location": "/nsd/v1/ns_descriptors/", "Content-Type": "application/json"}, "json")
707 self.step += 1
708 self.nsd_id = engine.test_ids["last_id"]
709 # location = r.headers["Location"]
710 # vnfd_id = location[location.rfind("/")+1:]
711 engine.test("DEPLOY{}".format(self.step), "Onboard NSD step 2 as ZIP", "PUT",
712 "/nsd/v1/ns_descriptors/<>/nsd_content" + self.qforce,
713 headers, "@b" + nsd_filename_path, 204, None, 0)
714 self.step += 2
715
716 if self.descriptor_edit and "nsd" in self.descriptor_edit:
717 # Modify NSD
718 engine.test("DEPLOY{}".format(self.step), "Edit NSD ", "PATCH",
719 "/nsd/v1/ns_descriptors/{}".format(self.nsd_id),
720 headers_yaml, self.descriptor_edit["nsd"], 204, None, None)
721 self.step += 1
722
723 def delete_descriptors(self, engine):
724 # delete descriptors
725 engine.test("DEPLOY{}".format(self.step), "Delete NSSD SOL005", "DELETE",
726 "/nsd/v1/ns_descriptors/{}".format(self.nsd_id),
727 headers_yaml, None, 204, None, 0)
728 self.step += 1
729 for vnfd_id in self.vnfds_id:
730 engine.test("DEPLOY{}".format(self.step), "Delete VNFD SOL005", "DELETE",
731 "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_id), headers_yaml, None, 204, None, 0)
732 self.step += 1
733
734 def instantiate(self, engine, ns_data):
735 ns_data_text = yaml.safe_dump(ns_data, default_flow_style=True, width=256)
736 # create NS Two steps
737 r = engine.test("DEPLOY{}".format(self.step), "Create NS step 1", "POST", "/nslcm/v1/ns_instances",
738 headers_yaml, ns_data_text, 201,
739 {"Location": "nslcm/v1/ns_instances/", "Content-Type": "application/yaml"}, "yaml")
740 self.ns_test = "DEPLOY{}".format(self.step)
741 self.ns_id = engine.test_ids["last_id"]
742 self.step += 1
743 r = engine.test("DEPLOY{}".format(self.step), "Instantiate NS step 2", "POST",
744 "/nslcm/v1/ns_instances/<{}>/instantiate".format(self.ns_test), headers_yaml, ns_data_text,
745 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
746 nslcmop_test = "DEPLOY{}".format(self.step)
747 self.step += 1
748
749 if test_osm:
750 # Wait until status is Ok
751 wait = timeout_configure if self.uses_configuration else timeout_deploy
752 while wait >= 0:
753 r = engine.test("DEPLOY{}".format(self.step), "Wait until NS is deployed and configured", "GET",
754 "/nslcm/v1/ns_lcm_op_occs/<{}>".format(nslcmop_test), headers_json, None,
755 200, r_header_json, "json")
756 nslcmop = r.json()
757 if "COMPLETED" in nslcmop["operationState"]:
758 break
759 elif "FAILED" in nslcmop["operationState"]:
760 raise TestException("NS instantiate has failed: {}".format(nslcmop["detailed-status"]))
761 wait -= 5
762 sleep(5)
763 else:
764 raise TestException("NS instantiate is not done after {} seconds".format(timeout_deploy))
765 self.step += 1
766
767 def _wait_nslcmop_ready(self, engine, nslcmop_test, timeout_deploy, expected_fail=False):
768 wait = timeout
769 while wait >= 0:
770 r = engine.test("DEPLOY{}".format(self.step), "Wait to ns lcm operation complete", "GET",
771 "/nslcm/v1/ns_lcm_op_occs/<{}>".format(nslcmop_test), headers_json, None,
772 200, r_header_json, "json")
773 nslcmop = r.json()
774 if "COMPLETED" in nslcmop["operationState"]:
775 if expected_fail:
776 raise TestException("NS terminate has success, expecting failing: {}".format(
777 nslcmop["detailed-status"]))
778 break
779 elif "FAILED" in nslcmop["operationState"]:
780 if not expected_fail:
781 raise TestException("NS terminate has failed: {}".format(nslcmop["detailed-status"]))
782 break
783 wait -= 5
784 sleep(5)
785 else:
786 raise TestException("NS instantiate is not terminate after {} seconds".format(timeout))
787
788 def terminate(self, engine):
789 # remove deployment
790 if test_osm:
791 r = engine.test("DEPLOY{}".format(self.step), "Terminate NS", "POST",
792 "/nslcm/v1/ns_instances/<{}>/terminate".format(self.ns_test), headers_yaml, None,
793 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
794 nslcmop2_test = "DEPLOY{}".format(self.step)
795 self.step += 1
796 # Wait until status is Ok
797 self._wait_nslcmop_ready(engine, nslcmop2_test, timeout_deploy)
798
799 r = engine.test("DEPLOY{}".format(self.step), "Delete NS", "DELETE",
800 "/nslcm/v1/ns_instances/<{}>".format(self.ns_test), headers_yaml, None,
801 204, None, 0)
802 self.step += 1
803 else:
804 r = engine.test("DEPLOY{}".format(self.step), "Delete NS with FORCE", "DELETE",
805 "/nslcm/v1/ns_instances/<{}>?FORCE=True".format(self.ns_test), headers_yaml, None,
806 204, None, 0)
807 self.step += 1
808
809 # check all it is deleted
810 r = engine.test("DEPLOY{}".format(self.step), "Check NS is deleted", "GET",
811 "/nslcm/v1/ns_instances/<{}>".format(self.ns_test), headers_yaml, None,
812 404, None, "yaml")
813 self.step += 1
814 r = engine.test("DEPLOY{}".format(self.step), "Check NSLCMOPs are deleted", "GET",
815 "/nslcm/v1/ns_lcm_op_occs?nsInstanceId=<{}>".format(self.ns_test), headers_json, None,
816 200, None, "json")
817 nslcmops = r.json()
818 if not isinstance(nslcmops, list) or nslcmops:
819 raise TestException("NS {} deleted but with ns_lcm_op_occ active: {}".format(self.ns_test, nslcmops))
820
821 def test_ns(self, engine, test_osm, commands=None, users=None, passwds=None, keys=None, timeout=0):
822
823 n = 0
824 r = engine.test("TEST_NS{}".format(n), "GET VNFR_IDs", "GET",
825 "/nslcm/v1/ns_instances/{}".format(self.ns_id), headers_json, None,
826 200, r_header_json, "json")
827 n += 1
828 ns_data = r.json()
829
830 vnfr_list = ns_data['constituent-vnfr-ref']
831 time = 0
832
833 for vnfr_id in vnfr_list:
834 self.total_tests += 1
835 r = engine.test("TEST_NS{}".format(n), "GET IP_ADDRESS OF VNFR", "GET",
836 "/nslcm/v1/vnfrs/{}".format(vnfr_id), headers_json, None,
837 200, r_header_json, "json")
838 n += 1
839 vnfr_data = r.json()
840
841 if vnfr_data.get("ip-address"):
842 name = "TEST_NS{}".format(n)
843 description = "Run tests in VNFR with IP {}".format(vnfr_data['ip-address'])
844 n += 1
845 test_description = "Test {} {}".format(name, description)
846 logger.warning(test_description)
847 vnf_index = str(vnfr_data["member-vnf-index-ref"])
848 while timeout >= time:
849 result, message = self.do_checks([vnfr_data["ip-address"]],
850 vnf_index=vnfr_data["member-vnf-index-ref"],
851 commands=commands.get(vnf_index), user=users.get(vnf_index),
852 passwd=passwds.get(vnf_index), key=keys.get(vnf_index))
853 if result == 1:
854 logger.warning(message)
855 break
856 elif result == 0:
857 time += 20
858 sleep(20)
859 elif result == -1:
860 logger.critical(message)
861 break
862 else:
863 time -= 20
864 logger.critical(message)
865 else:
866 logger.critical("VNFR {} has not mgmt address. Check failed".format(vnfr_id))
867
868 def do_checks(self, ip, vnf_index, commands=[], user=None, passwd=None, key=None):
869 try:
870 import urllib3
871 from pssh.clients import ParallelSSHClient
872 from pssh.utils import load_private_key
873 from ssh2 import exceptions as ssh2Exception
874 except ImportError as e:
875 logger.critical("Package <pssh> or/and <urllib3> is not installed. Please add them with 'pip3 install "
876 "parallel-ssh urllib3': {}".format(e))
877 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
878 try:
879 p_host = os.environ.get("PROXY_HOST")
880 p_user = os.environ.get("PROXY_USER")
881 p_password = os.environ.get("PROXY_PASSWD")
882
883 if key:
884 pkey = load_private_key(key)
885 else:
886 pkey = None
887
888 client = ParallelSSHClient(ip, user=user, password=passwd, pkey=pkey, proxy_host=p_host,
889 proxy_user=p_user, proxy_password=p_password, timeout=10, num_retries=0)
890 for cmd in commands:
891 output = client.run_command(cmd)
892 client.join(output)
893 if output[ip[0]].exit_code:
894 return -1, " VNFR {} could not be checked: {}".format(ip[0], output[ip[0]].stderr)
895 else:
896 self.passed_tests += 1
897 return 1, " Test successful"
898 except (ssh2Exception.ChannelFailure, ssh2Exception.SocketDisconnectError, ssh2Exception.SocketTimeout,
899 ssh2Exception.SocketRecvError) as e:
900 return 0, "Timeout accessing the VNFR {}: {}".format(ip[0], str(e))
901 except Exception as e:
902 return -1, "ERROR checking the VNFR {}: {}".format(ip[0], str(e))
903
904 def aditional_operations(self, engine, test_osm, manual_check):
905 pass
906
907 def run(self, engine, test_osm, manual_check, test_params=None):
908 engine.get_autorization()
909 nsname = os.environ.get("OSMNBITEST_NS_NAME", "OSMNBITEST")
910 if test_params:
911 if "vnfd-files" in test_params:
912 self.vnfd_filenames = test_params["vnfd-files"].split(",")
913 if "nsd-file" in test_params:
914 self.nsd_filename = test_params["nsd-file"]
915 if test_params.get("ns-name"):
916 nsname = test_params["ns-name"]
917 self.create_descriptors(engine)
918
919 # create real VIM if not exist
920 self.vim_id = engine.get_create_vim(test_osm)
921 ns_data = {"nsDescription": "default description", "nsName": nsname, "nsdId": self.nsd_id,
922 "vimAccountId": self.vim_id}
923 if test_params and test_params.get("ns-config"):
924 if isinstance(test_params["ns-config"], str):
925 ns_data.update(yaml.load(test_params["ns-config"]))
926 else:
927 ns_data.update(test_params["ns-config"])
928 self.instantiate(engine, ns_data)
929
930 if manual_check:
931 input('NS has been deployed. Perform manual check and press enter to resume')
932 elif test_osm:
933 self.test_ns(engine, test_osm, self.cmds, self.uss, self.pss, self.keys, self.timeout)
934 self.aditional_operations(engine, test_osm, manual_check)
935 self.terminate(engine)
936 self.delete_descriptors(engine)
937 self.print_results()
938
939 def print_results(self):
940 print("\n\n\n--------------------------------------------")
941 print("TEST RESULTS:\n PASSED TESTS: {} - TOTAL TESTS: {}".format(self.total_tests, self.passed_tests))
942 print("--------------------------------------------")
943
944
945 class TestDeployHackfestCirros(TestDeploy):
946 description = "Load and deploy Hackfest cirros_2vnf_ns example"
947
948 def __init__(self):
949 super().__init__()
950 self.vnfd_filenames = ("cirros_vnf.tar.gz",)
951 self.nsd_filename = "cirros_2vnf_ns.tar.gz"
952 self.cmds = {'1': ['ls -lrt', ], '2': ['ls -lrt', ]}
953 self.uss = {'1': "cirros", '2': "cirros"}
954 self.pss = {'1': "cubswin:)", '2': "cubswin:)"}
955
956
957 class TestDeployHackfest1(TestDeploy):
958 description = "Load and deploy Hackfest_1_vnfd example"
959
960 def __init__(self):
961 super().__init__()
962 self.vnfd_filenames = ("hackfest_1_vnfd.tar.gz",)
963 self.nsd_filename = "hackfest_1_nsd.tar.gz"
964 # self.cmds = {'1': ['ls -lrt', ], '2': ['ls -lrt', ]}
965 # self.uss = {'1': "cirros", '2': "cirros"}
966 # self.pss = {'1': "cubswin:)", '2': "cubswin:)"}
967
968
969 class TestDeployHackfestCirrosScaling(TestDeploy):
970 description = "Load and deploy Hackfest cirros_2vnf_ns example with scaling modifications"
971
972 def __init__(self):
973 super().__init__()
974 self.vnfd_filenames = ("cirros_vnf.tar.gz",)
975 self.nsd_filename = "cirros_2vnf_ns.tar.gz"
976
977 def create_descriptors(self, engine):
978 super().create_descriptors(engine)
979 # Modify VNFD to add scaling and count=2
980 payload = """
981 vdu:
982 "$id: 'cirros_vnfd-VM'":
983 count: 2
984 scaling-group-descriptor:
985 - name: "scale_cirros"
986 max-instance-count: 2
987 vdu:
988 - vdu-id-ref: cirros_vnfd-VM
989 count: 2
990 """
991 engine.test("DEPLOY{}".format(self.step), "Edit VNFD ", "PATCH",
992 "/vnfpkgm/v1/vnf_packages/{}".format(self.vnfds_id[0]),
993 headers_yaml, payload, 204, None, None)
994 self.step += 1
995
996 def aditional_operations(self, engine, test_osm, manual_check):
997 if not test_osm:
998 return
999 # 2 perform scale out twice
1000 payload = '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_OUT, scaleByStepData: ' \
1001 '{scaling-group-descriptor: scale_cirros, member-vnf-index: "1"}}}'
1002 for i in range(0, 2):
1003 engine.test("DEPLOY{}".format(self.step), "Execute scale action over NS", "POST",
1004 "/nslcm/v1/ns_instances/<{}>/scale".format(self.ns_test), headers_yaml, payload,
1005 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
1006 nslcmop2_scale_out = "DEPLOY{}".format(self.step)
1007 self._wait_nslcmop_ready(engine, nslcmop2_scale_out, timeout_deploy)
1008 if manual_check:
1009 input('NS scale out done. Check that two more vdus are there')
1010 # TODO check automatic
1011
1012 # 2 perform scale in
1013 payload = '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_IN, scaleByStepData: ' \
1014 '{scaling-group-descriptor: scale_cirros, member-vnf-index: "1"}}}'
1015 for i in range(0, 2):
1016 engine.test("DEPLOY{}".format(self.step), "Execute scale IN action over NS", "POST",
1017 "/nslcm/v1/ns_instances/<{}>/scale".format(self.ns_test), headers_yaml, payload,
1018 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
1019 nslcmop2_scale_in = "DEPLOY{}".format(self.step)
1020 self._wait_nslcmop_ready(engine, nslcmop2_scale_in, timeout_deploy)
1021 if manual_check:
1022 input('NS scale in done. Check that two less vdus are there')
1023 # TODO check automatic
1024
1025 # perform scale in that must fail as reached limit
1026 engine.test("DEPLOY{}".format(self.step), "Execute scale IN out of limit action over NS", "POST",
1027 "/nslcm/v1/ns_instances/<{}>/scale".format(self.ns_test), headers_yaml, payload,
1028 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
1029 nslcmop2_scale_in = "DEPLOY{}".format(self.step)
1030 self._wait_nslcmop_ready(engine, nslcmop2_scale_in, timeout_deploy, expected_fail=True)
1031
1032
1033 class TestDeployIpMac(TestDeploy):
1034 description = "Load and deploy descriptor examples setting mac, ip address at descriptor and instantiate params"
1035
1036 def __init__(self):
1037 super().__init__()
1038 self.vnfd_filenames = ("vnfd_2vdu_set_ip_mac2.yaml", "vnfd_2vdu_set_ip_mac.yaml")
1039 self.nsd_filename = "scenario_2vdu_set_ip_mac.yaml"
1040 self.descriptor_url = \
1041 "https://osm.etsi.org/gitweb/?p=osm/RO.git;a=blob_plain;f=test/RO_tests/v3_2vdu_set_ip_mac/"
1042 self.cmds = {'1': ['ls -lrt', ], '2': ['ls -lrt', ]}
1043 self.uss = {'1': "osm", '2': "osm"}
1044 self.pss = {'1': "osm4u", '2': "osm4u"}
1045 self.timeout = 360
1046
1047 def run(self, engine, test_osm, manual_check, test_params=None):
1048 # super().run(engine, test_osm, manual_check, test_params)
1049 # run again setting IPs with instantiate parameters
1050 instantiation_params = {
1051 "vnf": [
1052 {
1053 "member-vnf-index": "1",
1054 "internal-vld": [
1055 {
1056 "name": "internal_vld1", # net_internal
1057 "ip-profile": {
1058 "ip-version": "ipv4",
1059 "subnet-address": "10.9.8.0/24",
1060 "dhcp-params": {"count": 100, "start-address": "10.9.8.100"}
1061 },
1062 "internal-connection-point": [
1063 {
1064 "id-ref": "eth2",
1065 "ip-address": "10.9.8.2",
1066 },
1067 {
1068 "id-ref": "eth3",
1069 "ip-address": "10.9.8.3",
1070 }
1071 ]
1072 },
1073 ],
1074
1075 "vdu": [
1076 {
1077 "id": "VM1",
1078 "interface": [
1079 # {
1080 # "name": "iface11",
1081 # "floating-ip-required": True,
1082 # },
1083 {
1084 "name": "iface13",
1085 "mac-address": "52:33:44:55:66:13"
1086 },
1087 ],
1088 },
1089 {
1090 "id": "VM2",
1091 "interface": [
1092 {
1093 "name": "iface21",
1094 "ip-address": "10.31.31.22",
1095 "mac-address": "52:33:44:55:66:21"
1096 },
1097 ],
1098 },
1099 ]
1100 },
1101 ]
1102 }
1103
1104 super().run(engine, test_osm, manual_check, test_params={"ns-config": instantiation_params})
1105
1106
1107 class TestDeployHackfest4(TestDeploy):
1108 description = "Load and deploy Hackfest 4 example."
1109
1110 def __init__(self):
1111 super().__init__()
1112 self.vnfd_filenames = ("hackfest_4_vnfd.tar.gz",)
1113 self.nsd_filename = "hackfest_4_nsd.tar.gz"
1114 self.uses_configuration = True
1115 self.cmds = {'1': ['ls -lrt', ], '2': ['ls -lrt', ]}
1116 self.uss = {'1': "ubuntu", '2': "ubuntu"}
1117 self.pss = {'1': "osm4u", '2': "osm4u"}
1118
1119 def create_descriptors(self, engine):
1120 super().create_descriptors(engine)
1121 # Modify VNFD to add scaling
1122 payload = """
1123 scaling-group-descriptor:
1124 - name: "scale_dataVM"
1125 max-instance-count: 10
1126 scaling-policy:
1127 - name: "auto_cpu_util_above_threshold"
1128 scaling-type: "automatic"
1129 threshold-time: 0
1130 cooldown-time: 60
1131 scaling-criteria:
1132 - name: "cpu_util_above_threshold"
1133 scale-in-threshold: 15
1134 scale-in-relational-operation: "LE"
1135 scale-out-threshold: 60
1136 scale-out-relational-operation: "GE"
1137 vnf-monitoring-param-ref: "all_aaa_cpu_util"
1138 vdu:
1139 - vdu-id-ref: dataVM
1140 count: 1
1141 scaling-config-action:
1142 - trigger: post-scale-out
1143 vnf-config-primitive-name-ref: touch
1144 - trigger: pre-scale-in
1145 vnf-config-primitive-name-ref: touch
1146 vnf-configuration:
1147 config-primitive:
1148 - name: touch
1149 parameter:
1150 - name: filename
1151 data-type: STRING
1152 default-value: '/home/ubuntu/touched'
1153 """
1154 engine.test("DEPLOY{}".format(self.step), "Edit VNFD ", "PATCH",
1155 "/vnfpkgm/v1/vnf_packages/<{}>".format(self.vnfds_test[0]), headers_yaml, payload, 204, None, None)
1156 self.step += 1
1157
1158
1159 class TestDeployHackfest3Charmed(TestDeploy):
1160 description = "Load and deploy Hackfest 3charmed_ns example. Modifies it for adding scaling and performs " \
1161 "primitive actions and scaling"
1162
1163 def __init__(self):
1164 super().__init__()
1165 self.vnfd_filenames = ("hackfest_3charmed_vnfd.tar.gz",)
1166 self.nsd_filename = "hackfest_3charmed_nsd.tar.gz"
1167 self.uses_configuration = True
1168 self.cmds = {'1': [''], '2': ['ls -lrt /home/ubuntu/first-touch', ]}
1169 self.uss = {'1': "ubuntu", '2': "ubuntu"}
1170 self.pss = {'1': "osm4u", '2': "osm4u"}
1171
1172 # def create_descriptors(self, engine):
1173 # super().create_descriptors(engine)
1174 # # Modify VNFD to add scaling
1175 # payload = """
1176 # scaling-group-descriptor:
1177 # - name: "scale_dataVM"
1178 # max-instance-count: 10
1179 # scaling-policy:
1180 # - name: "auto_cpu_util_above_threshold"
1181 # scaling-type: "automatic"
1182 # threshold-time: 0
1183 # cooldown-time: 60
1184 # scaling-criteria:
1185 # - name: "cpu_util_above_threshold"
1186 # scale-in-threshold: 15
1187 # scale-in-relational-operation: "LE"
1188 # scale-out-threshold: 60
1189 # scale-out-relational-operation: "GE"
1190 # vnf-monitoring-param-ref: "all_aaa_cpu_util"
1191 # vdu:
1192 # - vdu-id-ref: dataVM
1193 # count: 1
1194 # scaling-config-action:
1195 # - trigger: post-scale-out
1196 # vnf-config-primitive-name-ref: touch
1197 # - trigger: pre-scale-in
1198 # vnf-config-primitive-name-ref: touch
1199 # vnf-configuration:
1200 # config-primitive:
1201 # - name: touch
1202 # parameter:
1203 # - name: filename
1204 # data-type: STRING
1205 # default-value: '/home/ubuntu/touched'
1206 # """
1207 # engine.test("DEPLOY{}".format(self.step), "Edit VNFD ", "PATCH",
1208 # "/vnfpkgm/v1/vnf_packages/<{}>".format(self.vnfds_test[0]),
1209 # headers_yaml, payload, 200,
1210 # r_header_yaml, yaml)
1211 # self.vnfds_test.append("DEPLOY" + str(self.step))
1212 # self.step += 1
1213
1214 def aditional_operations(self, engine, test_osm, manual_check):
1215 if not test_osm:
1216 return
1217 # 1 perform action
1218 payload = '{member_vnf_index: "2", primitive: touch, primitive_params: { filename: /home/ubuntu/OSMTESTNBI }}'
1219 engine.test("DEPLOY{}".format(self.step), "Executer service primitive over NS", "POST",
1220 "/nslcm/v1/ns_instances/<{}>/action".format(self.ns_test), headers_yaml, payload,
1221 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
1222 nslcmop2_action = "DEPLOY{}".format(self.step)
1223 self.step += 1
1224 # Wait until status is Ok
1225 self._wait_nslcmop_ready(engine, nslcmop2_action, timeout_deploy)
1226 if manual_check:
1227 input('NS service primitive has been executed. Check that file /home/ubuntu/OSMTESTNBI is present at '
1228 'TODO_PUT_IP')
1229 elif test_osm:
1230 cmds = {'1': [''], '2': ['ls -lrt /home/ubuntu/OSMTESTNBI', ]}
1231 uss = {'1': "ubuntu", '2': "ubuntu"}
1232 pss = {'1': "osm4u", '2': "osm4u"}
1233 self.test_ns(engine, test_osm, cmds, uss, pss, self.keys, self.timeout)
1234
1235 # # 2 perform scale out
1236 # payload = '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_OUT, scaleByStepData: ' \
1237 # '{scaling-group-descriptor: scale_dataVM, member-vnf-index: "1"}}}'
1238 # engine.test("DEPLOY{}".format(self.step), "Execute scale action over NS", "POST",
1239 # "/nslcm/v1/ns_instances/<{}>/scale".format(self.ns_test), headers_yaml, payload,
1240 # 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
1241 # nslcmop2_scale_out = "DEPLOY{}".format(self.step)
1242 # self._wait_nslcmop_ready(engine, nslcmop2_scale_out, timeout_deploy)
1243 # if manual_check:
1244 # input('NS scale out done. Check that file /home/ubuntu/touched is present and new VM is created')
1245 # # TODO check automatic
1246 #
1247 # # 2 perform scale in
1248 # payload = '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_IN, scaleByStepData: ' \
1249 # '{scaling-group-descriptor: scale_dataVM, member-vnf-index: "1"}}}'
1250 # engine.test("DEPLOY{}".format(self.step), "Execute scale action over NS", "POST",
1251 # "/nslcm/v1/ns_instances/<{}>/scale".format(self.ns_test), headers_yaml, payload,
1252 # 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
1253 # nslcmop2_scale_in = "DEPLOY{}".format(self.step)
1254 # self._wait_nslcmop_ready(engine, nslcmop2_scale_in, timeout_deploy)
1255 # if manual_check:
1256 # input('NS scale in done. Check that file /home/ubuntu/touched is updated and new VM is deleted')
1257 # # TODO check automatic
1258
1259
1260 class TestDeploySingleVdu(TestDeployHackfest3Charmed):
1261 description = "Generate a single VDU base on editing Hackfest3Charmed descriptors and deploy"
1262
1263 def __init__(self):
1264 super().__init__()
1265 self.qforce = "?FORCE=True"
1266 self.descriptor_edit = {
1267 # Modify VNFD to remove one VDU
1268 "vnfd0": {
1269 "vdu": {
1270 "$[0]": {
1271 "interface": {"$[0]": {"external-connection-point-ref": "pdu-mgmt"}}
1272 },
1273 "$[1]": None
1274 },
1275 "vnf-configuration": None,
1276 "connection-point": {
1277 "$[0]": {
1278 "id": "pdu-mgmt",
1279 "name": "pdu-mgmt",
1280 "short-name": "pdu-mgmt"
1281 },
1282 "$[1]": None
1283 },
1284 "mgmt-interface": {"cp": "pdu-mgmt"},
1285 "description": "A vnf single vdu to be used as PDU",
1286 "id": "vdu-as-pdu",
1287 "internal-vld": {
1288 "$[0]": {
1289 "id": "pdu_internal",
1290 "name": "pdu_internal",
1291 "internal-connection-point": {"$[1]": None},
1292 "short-name": "pdu_internal",
1293 "type": "ELAN"
1294 }
1295 }
1296 },
1297
1298 # Modify NSD accordingly
1299 "nsd": {
1300 "constituent-vnfd": {
1301 "$[0]": {"vnfd-id-ref": "vdu-as-pdu"},
1302 "$[1]": None,
1303 },
1304 "description": "A nsd to deploy the vnf to act as as PDU",
1305 "id": "nsd-as-pdu",
1306 "name": "nsd-as-pdu",
1307 "short-name": "nsd-as-pdu",
1308 "vld": {
1309 "$[0]": {
1310 "id": "mgmt_pdu",
1311 "name": "mgmt_pdu",
1312 "short-name": "mgmt_pdu",
1313 "vnfd-connection-point-ref": {
1314 "$[0]": {
1315 "vnfd-connection-point-ref": "pdu-mgmt",
1316 "vnfd-id-ref": "vdu-as-pdu",
1317 },
1318 "$[1]": None
1319 },
1320 "type": "ELAN"
1321 },
1322 "$[1]": None,
1323 }
1324 }
1325 }
1326
1327
1328 class TestDeployHnfd(TestDeployHackfest3Charmed):
1329 description = "Generate a HNFD base on editing Hackfest3Charmed descriptors and deploy"
1330
1331 def __init__(self):
1332 super().__init__()
1333 self.pduDeploy = TestDeploySingleVdu()
1334 self.pdu_interface_0 = {}
1335 self.pdu_interface_1 = {}
1336
1337 self.pdu_id = None
1338 # self.vnf_to_pdu = """
1339 # vdu:
1340 # "$[0]":
1341 # pdu-type: PDU-TYPE-1
1342 # interface:
1343 # "$[0]":
1344 # name: mgmt-iface
1345 # "$[1]":
1346 # name: pdu-iface-internal
1347 # id: hfn1
1348 # description: HFND, one PDU + One VDU
1349 # name: hfn1
1350 # short-name: hfn1
1351 #
1352 # """
1353
1354 self.pdu_descriptor = {
1355 "name": "my-PDU",
1356 "type": "PDU-TYPE-1",
1357 "vim_accounts": "to-override",
1358 "interfaces": [
1359 {
1360 "name": "mgmt-iface",
1361 "mgmt": True,
1362 "type": "overlay",
1363 "ip-address": "to override",
1364 "mac-address": "mac_address",
1365 "vim-network-name": "mgmt",
1366 },
1367 {
1368 "name": "pdu-iface-internal",
1369 "mgmt": False,
1370 "type": "overlay",
1371 "ip-address": "to override",
1372 "mac-address": "mac_address",
1373 "vim-network-name": "pdu_internal", # OSMNBITEST-PDU-pdu_internal
1374 },
1375 ]
1376 }
1377 self.vnfd_filenames = ("hackfest_3charmed_vnfd.tar.gz", "hackfest_3charmed_vnfd.tar.gz")
1378
1379 self.descriptor_edit = {
1380 "vnfd0": {
1381 "id": "hfnd1",
1382 "name": "hfn1",
1383 "short-name": "hfn1",
1384 "vdu": {
1385 "$[0]": {
1386 "pdu-type": "PDU-TYPE-1",
1387 "interface": {
1388 "$[0]": {"name": "mgmt-iface"},
1389 "$[1]": {"name": "pdu-iface-internal"},
1390 }
1391 }
1392 }
1393 },
1394 "nsd": {
1395 "constituent-vnfd": {
1396 "$[1]": {"vnfd-id-ref": "hfnd1"}
1397 }
1398 }
1399 }
1400
1401 def create_descriptors(self, engine):
1402 super().create_descriptors(engine)
1403
1404 # Create PDU
1405 self.pdu_descriptor["interfaces"][0].update(self.pdu_interface_0)
1406 self.pdu_descriptor["interfaces"][1].update(self.pdu_interface_1)
1407 self.pdu_descriptor["vim_accounts"] = [self.vim_id]
1408 # TODO get vim-network-name from vnfr.vld.name
1409 self.pdu_descriptor["interfaces"][1]["vim-network-name"] = "{}-{}-{}".format(
1410 os.environ.get("OSMNBITEST_NS_NAME", "OSMNBITEST"),
1411 "PDU", self.pdu_descriptor["interfaces"][1]["vim-network-name"])
1412 test_name = "DEPLOY{}".format(self.step)
1413 engine.test(test_name, "Onboard PDU descriptor", "POST", "/pdu/v1/pdu_descriptors",
1414 {"Location": "/pdu/v1/pdu_descriptors/", "Content-Type": "application/yaml"}, self.pdu_descriptor,
1415 201, r_header_yaml, "yaml")
1416 self.pdu_id = engine.test_ids["last_id"]
1417 self.step += 1
1418
1419 def run(self, engine, test_osm, manual_check, test_params=None):
1420 engine.get_autorization()
1421 nsname = os.environ.get("OSMNBITEST_NS_NAME", "OSMNBITEST")
1422
1423 # create real VIM if not exist
1424 self.vim_id = engine.get_create_vim(test_osm)
1425 # instanciate PDU
1426 self.pduDeploy.create_descriptors(engine)
1427 self.pduDeploy.instantiate(engine, {"nsDescription": "to be used as PDU", "nsName": nsname + "-PDU",
1428 "nsdId": self.pduDeploy.nsd_id, "vimAccountId": self.vim_id})
1429 if manual_check:
1430 input('VNF to be used as PDU has been deployed. Perform manual check and press enter to resume')
1431 elif test_osm:
1432 self.pduDeploy.test_ns(engine, test_osm, self.pduDeploy.cmds, self.pduDeploy.uss, self.pduDeploy.pss,
1433 self.pduDeploy.keys, self.pduDeploy.timeout)
1434
1435 if test_osm:
1436 r = engine.test("DEPLOY{}".format(self.step), "GET IP_ADDRESS OF VNFR", "GET",
1437 "/nslcm/v1/vnfrs?nsr-id-ref={}".format(self.pduDeploy.ns_id), headers_json, None,
1438 200, r_header_json, "json")
1439 self.step += 1
1440 vnfr_data = r.json()
1441 # print(vnfr_data)
1442
1443 self.pdu_interface_0["ip-address"] = vnfr_data[0]["vdur"][0]["interfaces"][0].get("ip-address")
1444 self.pdu_interface_1["ip-address"] = vnfr_data[0]["vdur"][0]["interfaces"][1].get("ip-address")
1445 self.pdu_interface_0["mac-address"] = vnfr_data[0]["vdur"][0]["interfaces"][0].get("mac-address")
1446 self.pdu_interface_1["mac-address"] = vnfr_data[0]["vdur"][0]["interfaces"][1].get("mac-address")
1447 if not self.pdu_interface_0["ip-address"]:
1448 raise TestException("Vnfr has not managment ip address")
1449 else:
1450 self.pdu_interface_0["ip-address"] = "192.168.10.10"
1451 self.pdu_interface_1["ip-address"] = "192.168.11.10"
1452 self.pdu_interface_0["mac-address"] = "52:33:44:55:66:13"
1453 self.pdu_interface_1["mac-address"] = "52:33:44:55:66:14"
1454
1455 self.create_descriptors(engine)
1456
1457 ns_data = {"nsDescription": "default description", "nsName": nsname, "nsdId": self.nsd_id,
1458 "vimAccountId": self.vim_id}
1459 if test_params and test_params.get("ns-config"):
1460 if isinstance(test_params["ns-config"], str):
1461 ns_data.update(yaml.load(test_params["ns-config"]))
1462 else:
1463 ns_data.update(test_params["ns-config"])
1464
1465 self.instantiate(engine, ns_data)
1466 if manual_check:
1467 input('NS has been deployed. Perform manual check and press enter to resume')
1468 elif test_osm:
1469 self.test_ns(engine, test_osm, self.cmds, self.uss, self.pss, self.keys, self.timeout)
1470 self.aditional_operations(engine, test_osm, manual_check)
1471 self.terminate(engine)
1472 self.pduDeploy.terminate(engine)
1473 self.delete_descriptors(engine)
1474 self.pduDeploy.delete_descriptors(engine)
1475
1476 self.step += 1
1477
1478 self.print_results()
1479
1480 def delete_descriptors(self, engine):
1481 super().delete_descriptors(engine)
1482 # delete pdu
1483 engine.test("DEPLOY{}".format(self.step), "Delete PDU SOL005", "DELETE",
1484 "/pdu/v1/pdu_descriptors/{}".format(self.pdu_id),
1485 headers_yaml, None, 204, None, 0)
1486
1487
1488 class TestDescriptors:
1489 description = "Test VNFD, NSD, PDU descriptors CRUD and dependencies"
1490
1491 def __init__(self):
1492 self.step = 0
1493 self.vnfd_filename = "hackfest_3charmed_vnfd.tar.gz"
1494 self.nsd_filename = "hackfest_3charmed_nsd.tar.gz"
1495 self.descriptor_url = "https://osm-download.etsi.org/ftp/osm-3.0-three/2nd-hackfest/packages/"
1496 self.vnfd_id = None
1497 self.nsd_id = None
1498
1499 def run(self, engine, test_osm, manual_check, test_params=None):
1500 engine.get_autorization()
1501 temp_dir = os.path.dirname(os.path.abspath(__file__)) + "/temp/"
1502 if not os.path.exists(temp_dir):
1503 os.makedirs(temp_dir)
1504
1505 # download files
1506 for filename in (self.vnfd_filename, self.nsd_filename):
1507 filename_path = temp_dir + filename
1508 if not os.path.exists(filename_path):
1509 with open(filename_path, "wb") as file:
1510 response = requests.get(self.descriptor_url + filename)
1511 if response.status_code >= 300:
1512 raise TestException("Error downloading descriptor from '{}': {}".format(
1513 self.descriptor_url + filename, response.status_code))
1514 file.write(response.content)
1515
1516 vnfd_filename_path = temp_dir + self.vnfd_filename
1517 nsd_filename_path = temp_dir + self.nsd_filename
1518
1519 # vnfd CREATE AND UPLOAD in one step:
1520 test_name = "DESCRIPTOR{}".format(self.step)
1521 engine.test(test_name, "Onboard VNFD in one step", "POST",
1522 "/vnfpkgm/v1/vnf_packages_content", headers_zip_yaml, "@b" + vnfd_filename_path, 201,
1523 {"Location": "/vnfpkgm/v1/vnf_packages_content/", "Content-Type": "application/yaml"}, "yaml")
1524 self.vnfd_id = engine.test_ids["last_id"]
1525 self.step += 1
1526
1527 # get vnfd descriptor
1528 engine.test("DESCRIPTOR" + str(self.step), "Get VNFD descriptor", "GET",
1529 "/vnfpkgm/v1/vnf_packages/{}".format(self.vnfd_id), headers_yaml, None, 200, r_header_yaml, "yaml")
1530 self.step += 1
1531
1532 # get vnfd file descriptor
1533 engine.test("DESCRIPTOR" + str(self.step), "Get VNFD file descriptor", "GET",
1534 "/vnfpkgm/v1/vnf_packages/{}/vnfd".format(self.vnfd_id), headers_text, None, 200,
1535 r_header_text, "text", temp_dir+"vnfd-yaml")
1536 self.step += 1
1537 # TODO compare files: diff vnfd-yaml hackfest_3charmed_vnfd/hackfest_3charmed_vnfd.yaml
1538
1539 # get vnfd zip file package
1540 engine.test("DESCRIPTOR" + str(self.step), "Get VNFD zip package", "GET",
1541 "/vnfpkgm/v1/vnf_packages/{}/package_content".format(self.vnfd_id), headers_zip, None, 200,
1542 r_header_zip, "zip", temp_dir+"vnfd-zip")
1543 self.step += 1
1544 # TODO compare files: diff vnfd-zip hackfest_3charmed_vnfd.tar.gz
1545
1546 # get vnfd artifact
1547 engine.test("DESCRIPTOR" + str(self.step), "Get VNFD artifact package", "GET",
1548 "/vnfpkgm/v1/vnf_packages/{}/artifacts/icons/osm.png".format(self.vnfd_id), headers_zip, None, 200,
1549 r_header_octect, "octet-string", temp_dir+"vnfd-icon")
1550 self.step += 1
1551 # TODO compare files: diff vnfd-icon hackfest_3charmed_vnfd/icons/osm.png
1552
1553 # nsd CREATE AND UPLOAD in one step:
1554 test_name = "DESCRIPTOR{}".format(self.step)
1555 engine.test(test_name, "Onboard NSD in one step", "POST",
1556 "/nsd/v1/ns_descriptors_content", headers_zip_yaml, "@b" + nsd_filename_path, 201,
1557 {"Location": "/nsd/v1/ns_descriptors_content/", "Content-Type": "application/yaml"}, "yaml")
1558 self.nsd_id = engine.test_ids["last_id"]
1559 self.step += 1
1560
1561 # get nsd descriptor
1562 engine.test("DESCRIPTOR" + str(self.step), "Get NSD descriptor", "GET",
1563 "/nsd/v1/ns_descriptors/{}".format(self.nsd_id), headers_yaml, None, 200, r_header_yaml, "yaml")
1564 self.step += 1
1565
1566 # get nsd file descriptor
1567 engine.test("DESCRIPTOR" + str(self.step), "Get NSD file descriptor", "GET",
1568 "/nsd/v1/ns_descriptors/{}/nsd".format(self.nsd_id), headers_text, None, 200,
1569 r_header_text, "text", temp_dir+"nsd-yaml")
1570 self.step += 1
1571 # TODO compare files: diff nsd-yaml hackfest_3charmed_nsd/hackfest_3charmed_nsd.yaml
1572
1573 # get nsd zip file package
1574 engine.test("DESCRIPTOR" + str(self.step), "Get NSD zip package", "GET",
1575 "/nsd/v1/ns_descriptors/{}/nsd_content".format(self.nsd_id), headers_zip, None, 200,
1576 r_header_zip, "zip", temp_dir+"nsd-zip")
1577 self.step += 1
1578 # TODO compare files: diff nsd-zip hackfest_3charmed_nsd.tar.gz
1579
1580 # get nsd artifact
1581 engine.test("DESCRIPTOR" + str(self.step), "Get NSD artifact package", "GET",
1582 "/nsd/v1/ns_descriptors/{}/artifacts/icons/osm.png".format(self.nsd_id), headers_zip, None, 200,
1583 r_header_octect, "octet-string", temp_dir+"nsd-icon")
1584 self.step += 1
1585 # TODO compare files: diff nsd-icon hackfest_3charmed_nsd/icons/osm.png
1586
1587 # vnfd DELETE
1588 test_rest.test("DESCRIPTOR" + str(self.step), "Delete VNFD conflict", "DELETE",
1589 "/vnfpkgm/v1/vnf_packages/{}".format(self.vnfd_id), headers_yaml, None, 409, None, None)
1590 self.step += 1
1591
1592 test_rest.test("DESCRIPTOR" + str(self.step), "Delete VNFD force", "DELETE",
1593 "/vnfpkgm/v1/vnf_packages/{}?FORCE=TRUE".format(self.vnfd_id), headers_yaml, None, 204, None, 0)
1594 self.step += 1
1595
1596 # nsd DELETE
1597 test_rest.test("DESCRIPTOR" + str(self.step), "Delete NSD", "DELETE",
1598 "/nsd/v1/ns_descriptors/{}".format(self.nsd_id), headers_yaml, None, 204, None, 0)
1599 self.step += 1
1600
1601
1602 class TestNetSliceTemplates:
1603 description = "Upload a NST to OSM"
1604
1605 def __init__(self):
1606 self.nst_filenames = ("@./cirros_slice/cirros_slice.yaml")
1607
1608 def run(self, engine, test_osm, manual_check, test_params=None):
1609 # nst CREATE
1610 engine.get_autorization()
1611 r = engine.test("NST1", "Onboard NST", "POST", "/nst/v1/netslice_templates_content", headers_yaml,
1612 self.nst_filenames,
1613 201, {"Location": "/nst/v1/netslice_templates_content", "Content-Type": "application/yaml"},
1614 "yaml")
1615 location = r.headers["Location"]
1616 nst_id = location[location.rfind("/")+1:]
1617
1618 # nstd SHOW OSM format
1619 r = engine.test("NST2", "Show NSTD OSM format", "GET",
1620 "/nst/v1/netslice_templates/{}".format(nst_id), headers_json, None,
1621 200, r_header_json, "json")
1622
1623 # nstd DELETE
1624 r = engine.test("NST3", "Delete NSTD", "DELETE",
1625 "/nst/v1/netslice_templates/{}".format(nst_id), headers_json, None,
1626 204, None, 0)
1627
1628
1629 class TestNetSliceInstances:
1630 description = "Upload a NST to OSM"
1631
1632 def __init__(self):
1633 self.nst_filenames = ("@./cirros_slice/cirros_slice.yaml")
1634
1635 def run(self, engine, test_osm, manual_check, test_params=None):
1636 # nst CREATE
1637 engine.get_autorization()
1638 r = engine.test("NST1", "Onboard NST", "POST", "/nst/v1/netslice_templates_content", headers_yaml,
1639 self.nst_filenames, 201,
1640 {"Location": "/nst/v1/netslice_templates_content", "Content-Type": "application/yaml"}, "yaml")
1641 location = r.headers["Location"]
1642 nst_id = location[location.rfind("/")+1:]
1643
1644 # nsi CREATE
1645 if test_osm:
1646 self.vim_id = engine.get_create_vim(test_osm)
1647 else:
1648 r = engine.test("VIM1", "Get available VIM", "GET", "/admin/v1/vim_accounts",
1649 headers_json, None, None, r_header_json, "json")
1650 r_json = json.loads(r.text)
1651 vim = r_json[0]
1652 self.vim_id = vim["_id"]
1653
1654 ns_data = {"nsiDescription": "default description", "nsiName": "my_slice", "nstId": nst_id,
1655 "vimAccountId": self.vim_id}
1656 ns_data_text = yaml.safe_dump(ns_data, default_flow_style=True, width=256)
1657
1658 r = engine.test("NSI1", "Onboard NSI", "POST", "/nsilcm/v1/netslice_instances_content", headers_yaml,
1659 ns_data_text, 201,
1660 {"Location": "/nsilcm/v1/netslice_instances_content", "Content-Type":
1661 "application/yaml"}, "yaml")
1662 location = r.headers["Location"]
1663 nsi_id = location[location.rfind("/")+1:]
1664
1665 # TODO: Improve the wait with a polling if NSI was deployed
1666 wait = 120
1667 sleep(wait)
1668
1669 # Check deployment
1670 r = engine.test("NSI2", "Wait until NSI is deployed", "GET",
1671 "/nsilcm/v1/netslice_instances_content/{}".format(nsi_id), headers_json, None,
1672 200, r_header_json, "json")
1673
1674 # nsi DELETE
1675 r = engine.test("NSI3", "Delete NSI", "DELETE",
1676 "/nsilcm/v1/netslice_instances_content/{}".format(nsi_id), headers_json, None,
1677 202, r_header_json, "json")
1678
1679 sleep(60)
1680
1681 # nstd DELETE
1682 r = engine.test("NST2", "Delete NSTD", "DELETE",
1683 "/nst/v1/netslice_templates/{}".format(nst_id), headers_json, None,
1684 204, None, 0)
1685
1686
1687 if __name__ == "__main__":
1688 global logger
1689 test = ""
1690
1691 # Disable warnings from self-signed certificates.
1692 requests.packages.urllib3.disable_warnings()
1693 try:
1694 logging.basicConfig(format="%(levelname)s %(message)s", level=logging.ERROR)
1695 logger = logging.getLogger('NBI')
1696 # load parameters and configuration
1697 opts, args = getopt.getopt(sys.argv[1:], "hvu:p:",
1698 ["url=", "user=", "password=", "help", "version", "verbose", "no-verbose",
1699 "project=", "insecure", "timeout", "timeout-deploy", "timeout-configure",
1700 "test=", "list", "test-osm", "manual-check", "params="])
1701 url = "https://localhost:9999/osm"
1702 user = password = project = "admin"
1703 test_osm = False
1704 manual_check = False
1705 verbose = 0
1706 verify = True
1707 test_classes = {
1708 "NonAuthorized": TestNonAuthorized,
1709 "FakeVIM": TestFakeVim,
1710 "TestUsersProjects": TestUsersProjects,
1711 "VIM-SDN": TestVIMSDN,
1712 "Deploy-Custom": TestDeploy,
1713 "Deploy-Hackfest-Cirros": TestDeployHackfestCirros,
1714 "Deploy-Hackfest-Cirros-Scaling": TestDeployHackfestCirrosScaling,
1715 "Deploy-Hackfest-3Charmed": TestDeployHackfest3Charmed,
1716 "Deploy-Hackfest-4": TestDeployHackfest4,
1717 "Deploy-CirrosMacIp": TestDeployIpMac,
1718 "TestDescriptors": TestDescriptors,
1719 "TestDeployHackfest1": TestDeployHackfest1,
1720 # "Deploy-MultiVIM": TestDeployMultiVIM,
1721 "DeploySingleVdu": TestDeploySingleVdu,
1722 "DeployHnfd": TestDeployHnfd,
1723 "Upload-Slice-Template": TestNetSliceTemplates,
1724 "Deploy-Slice-Instance": TestNetSliceInstances,
1725 }
1726 test_to_do = []
1727 test_params = {}
1728
1729 for o, a in opts:
1730 # print("parameter:", o, a)
1731 if o == "--version":
1732 print("test version " + __version__ + ' ' + version_date)
1733 exit()
1734 elif o == "--list":
1735 for test, test_class in test_classes.items():
1736 print("{:20} {}".format(test + ":", test_class.description))
1737 exit()
1738 elif o in ("-v", "--verbose"):
1739 verbose += 1
1740 elif o == "no-verbose":
1741 verbose = -1
1742 elif o in ("-h", "--help"):
1743 usage()
1744 sys.exit()
1745 elif o == "--test-osm":
1746 test_osm = True
1747 elif o == "--manual-check":
1748 manual_check = True
1749 elif o == "--url":
1750 url = a
1751 elif o in ("-u", "--user"):
1752 user = a
1753 elif o in ("-p", "--password"):
1754 password = a
1755 elif o == "--project":
1756 project = a
1757 elif o == "--test":
1758 # print("asdfadf", o, a, a.split(","))
1759 for _test in a.split(","):
1760 if _test not in test_classes:
1761 print("Invalid test name '{}'. Use option '--list' to show available tests".format(_test),
1762 file=sys.stderr)
1763 exit(1)
1764 test_to_do.append(_test)
1765 elif o == "--params":
1766 param_key, _, param_value = a.partition("=")
1767 text_index = len(test_to_do)
1768 if text_index not in test_params:
1769 test_params[text_index] = {}
1770 test_params[text_index][param_key] = param_value
1771 elif o == "--insecure":
1772 verify = False
1773 elif o == "--timeout":
1774 timeout = int(a)
1775 elif o == "--timeout-deploy":
1776 timeout_deploy = int(a)
1777 elif o == "--timeout-configure":
1778 timeout_configure = int(a)
1779 else:
1780 assert False, "Unhandled option"
1781 if verbose == 0:
1782 logger.setLevel(logging.WARNING)
1783 elif verbose > 1:
1784 logger.setLevel(logging.DEBUG)
1785 else:
1786 logger.setLevel(logging.ERROR)
1787
1788 test_rest = TestRest(url, user=user, password=password, project=project)
1789 # print("tests to do:", test_to_do)
1790 if test_to_do:
1791 text_index = 0
1792 for test in test_to_do:
1793 text_index += 1
1794 test_class = test_classes[test]
1795 test_class().run(test_rest, test_osm, manual_check, test_params.get(text_index))
1796 else:
1797 for test, test_class in test_classes.items():
1798 test_class().run(test_rest, test_osm, manual_check, test_params.get(0))
1799 exit(0)
1800
1801 # get token
1802 print("PASS")
1803
1804 except TestException as e:
1805 logger.error(test + "Test {} Exception: {}".format(test, str(e)))
1806 exit(1)
1807 except getopt.GetoptError as e:
1808 logger.error(e)
1809 print(e, file=sys.stderr)
1810 exit(1)
1811 except Exception as e:
1812 logger.critical(test + " Exception: " + str(e), exc_info=True)