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