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


8adf20e5545408014bcfb6a8b4d9aac223b0c0f6
[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
120 def set_header(self, header):
121 self.s.headers.update(header)
122
123 def unset_header(self, key):
124 if key in self.s.headers:
125 del self.s.headers[key]
126
127 def test(self, name, description, method, url, headers, payload, expected_codes, expected_headers,
128 expected_payload):
129 """
130 Performs an http request and check http code response. Exit if different than allowed. It get the returned id
131 that can be used by following test in the URL with {name} where name is the name of the test
132 :param name: short name of the test
133 :param description: description of the test
134 :param method: HTTP method: GET,PUT,POST,DELETE,...
135 :param url: complete URL or relative URL
136 :param headers: request headers to add to the base headers
137 :param payload: Can be a dict, transformed to json, a text or a file if starts with '@'
138 :param expected_codes: expected response codes, can be int, int tuple or int range
139 :param expected_headers: expected response headers, dict with key values
140 :param expected_payload: expected payload, 0 if empty, 'yaml', 'json', 'text', 'zip'
141 :return: requests response
142 """
143 r = None
144 try:
145 if not self.s:
146 self.s = requests.session()
147 # URL
148 if not url:
149 url = self.url_base
150 elif not url.startswith("http"):
151 url = self.url_base + url
152
153 var_start = url.find("<") + 1
154 while var_start:
155 var_end = url.find(">", var_start)
156 if var_end == -1:
157 break
158 var_name = url[var_start:var_end]
159 if var_name in self.test_ids:
160 url = url[:var_start-1] + self.test_ids[var_name] + url[var_end+1:]
161 var_start += len(self.test_ids[var_name])
162 var_start = url.find("<", var_start) + 1
163 if payload:
164 if isinstance(payload, str):
165 if payload.startswith("@"):
166 mode = "r"
167 file_name = payload[1:]
168 if payload.startswith("@b"):
169 mode = "rb"
170 file_name = payload[2:]
171 with open(file_name, mode) as f:
172 payload = f.read()
173 elif isinstance(payload, dict):
174 payload = json.dumps(payload)
175
176 test_description = "Test {} {} {} {}".format(name, description, method, url)
177 if self.old_test_description != test_description:
178 self.old_test_description = test_description
179 logger.warning(test_description)
180 stream = False
181 # if expected_payload == "zip":
182 # stream = True
183 r = getattr(self.s, method.lower())(url, data=payload, headers=headers, verify=self.verify, stream=stream)
184 logger.debug("RX {}: {}".format(r.status_code, r.text))
185
186 # check response
187 if expected_codes:
188 if isinstance(expected_codes, int):
189 expected_codes = (expected_codes,)
190 if r.status_code not in expected_codes:
191 raise TestException(
192 "Got status {}. Expected {}. {}".format(r.status_code, expected_codes, r.text))
193
194 if expected_headers:
195 for header_key, header_val in expected_headers.items():
196 if header_key.lower() not in r.headers:
197 raise TestException("Header {} not present".format(header_key))
198 if header_val and header_val.lower() not in r.headers[header_key]:
199 raise TestException("Header {} does not contain {} but {}".format(header_key, header_val,
200 r.headers[header_key]))
201
202 if expected_payload is not None:
203 if expected_payload == 0 and len(r.content) > 0:
204 raise TestException("Expected empty payload")
205 elif expected_payload == "json":
206 try:
207 r.json()
208 except Exception as e:
209 raise TestException("Expected json response payload, but got Exception {}".format(e))
210 elif expected_payload == "yaml":
211 try:
212 yaml.safe_load(r.text)
213 except Exception as e:
214 raise TestException("Expected yaml response payload, but got Exception {}".format(e))
215 elif expected_payload == "zip":
216 if len(r.content) == 0:
217 raise TestException("Expected some response payload, but got empty")
218 # try:
219 # tar = tarfile.open(None, 'r:gz', fileobj=r.raw)
220 # for tarinfo in tar:
221 # tarname = tarinfo.name
222 # print(tarname)
223 # except Exception as e:
224 # raise TestException("Expected zip response payload, but got Exception {}".format(e))
225 elif expected_payload == "text":
226 if len(r.content) == 0:
227 raise TestException("Expected some response payload, but got empty")
228 # r.text
229 location = r.headers.get("Location")
230 if location:
231 _id = location[location.rfind("/") + 1:]
232 if _id:
233 self.test_ids[name] = str(_id)
234 self.test_ids[""] = str(_id) # last id
235 return r
236 except TestException as e:
237 r_status_code = None
238 r_text = None
239 if r:
240 r_status_code = r.status_code
241 r_text = r.text
242 logger.error("{} \nRX code{}: {}".format(e, r_status_code, r_text))
243 exit(1)
244 except IOError as e:
245 logger.error("Cannot open file {}".format(e))
246 exit(1)
247
248 def get_autorization(self): # user=None, password=None, project=None):
249 if self.token: # and self.user == user and self.password == password and self.project == project:
250 return
251 # self.user = user
252 # self.password = password
253 # self.project = project
254 r = self.test("TOKEN", "Obtain token", "POST", "/admin/v1/tokens", headers_json,
255 {"username": self.user, "password": self.password, "project_id": self.project},
256 (200, 201), r_header_json, "json")
257 response = r.json()
258 self.token = response["id"]
259 self.set_header({"Authorization": "Bearer {}".format(self.token)})
260
261 def remove_authorization(self):
262 if self.token:
263 self.test("TOKEN_DEL", "Delete token", "DELETE", "/admin/v1/tokens/{}".format(self.token), headers_json,
264 None, (200, 201, 204), None, None)
265 self.token = None
266 self.unset_header("Authorization")
267
268 def get_create_vim(self, test_osm):
269 if self.vim_id:
270 return self.vim_id
271 self.get_autorization()
272 if test_osm:
273 vim_name = os.environ.get("OSMNBITEST_VIM_NAME")
274 if not vim_name:
275 raise TestException(
276 "Needed to define OSMNBITEST_VIM_XXX variables to create a real VIM for deployment")
277 else:
278 vim_name = "fakeVim"
279 # Get VIM
280 r = self.test("_VIMGET1", "Get VIM ID", "GET", "/admin/v1/vim_accounts?name={}".format(vim_name), headers_json,
281 None, 200, r_header_json, "json")
282 vims = r.json()
283 if vims:
284 return vims[0]["_id"]
285 # Add VIM
286 if test_osm:
287 # check needed environ parameters:
288 if not os.environ.get("OSMNBITEST_VIM_URL") or not os.environ.get("OSMNBITEST_VIM_TENANT"):
289 raise TestException("Env OSMNBITEST_VIM_URL and OSMNBITEST_VIM_TENANT are needed for create a real VIM"
290 " to deploy on whit the --test-osm option")
291 vim_data = "{{schema_version: '1.0', name: '{}', vim_type: {}, vim_url: '{}', vim_tenant_name: '{}', "\
292 "vim_user: {}, vim_password: {}".format(vim_name,
293 os.environ.get("OSMNBITEST_VIM_TYPE", "openstack"),
294 os.environ.get("OSMNBITEST_VIM_URL"),
295 os.environ.get("OSMNBITEST_VIM_TENANT"),
296 os.environ.get("OSMNBITEST_VIM_USER"),
297 os.environ.get("OSMNBITEST_VIM_PASSWORD"))
298 if os.environ.get("OSMNBITEST_VIM_CONFIG"):
299 vim_data += " ,config: {}".format(os.environ.get("OSMNBITEST_VIM_CONFIG"))
300 vim_data += "}"
301 else:
302 vim_data = "{schema_version: '1.0', name: fakeVim, vim_type: openstack, vim_url: 'http://10.11.12.13/fake'"\
303 ", vim_tenant_name: 'vimtenant', vim_user: vimuser, vim_password: vimpassword}"
304 r = self.test("_VIMGET2", "Create VIM", "POST", "/admin/v1/vim_accounts", headers_yaml, vim_data,
305 (201), {"Location": "/admin/v1/vim_accounts/", "Content-Type": "application/yaml"}, "yaml")
306 location = r.headers.get("Location")
307 return location[location.rfind("/") + 1:]
308
309
310 class TestNonAuthorized:
311 description = "test invalid URLs. methods and no authorization"
312
313 @staticmethod
314 def run(engine, test_osm, manual_check, test_params=None):
315 engine.remove_authorization()
316 test_not_authorized_list = (
317 ("NA1", "Invalid token", "GET", "/admin/v1/users", headers_json, None, 401, r_header_json, "json"),
318 ("NA2", "Invalid URL", "POST", "/admin/v1/nonexist", headers_yaml, None, 405, r_header_yaml, "yaml"),
319 ("NA3", "Invalid version", "DELETE", "/admin/v2/users", headers_yaml, None, 405, r_header_yaml, "yaml"),
320 )
321 for t in test_not_authorized_list:
322 engine.test(*t)
323
324
325 class TestUsersProjects:
326 description = "test project and user creation"
327
328 @staticmethod
329 def run(engine, test_osm, manual_check, test_params=None):
330 engine.get_autorization()
331 engine.test("PU1", "Create project non admin", "POST", "/admin/v1/projects", headers_json, {"name": "P1"},
332 (201, 204), {"Location": "/admin/v1/projects/", "Content-Type": "application/json"}, "json")
333 engine.test("PU2", "Create project admin", "POST", "/admin/v1/projects", headers_json,
334 {"name": "Padmin", "admin": True}, (201, 204),
335 {"Location": "/admin/v1/projects/", "Content-Type": "application/json"}, "json")
336 engine.test("PU3", "Create project bad format", "POST", "/admin/v1/projects", headers_json, {"name": 1}, 422,
337 r_header_json, "json")
338 engine.test("PU4", "Create user with bad project", "POST", "/admin/v1/users", headers_json,
339 {"username": "U1", "projects": ["P1", "P2", "Padmin"], "password": "pw1"}, 409,
340 r_header_json, "json")
341 engine.test("PU5", "Create user with bad project and force", "POST", "/admin/v1/users?FORCE=True", headers_json,
342 {"username": "U1", "projects": ["P1", "P2", "Padmin"], "password": "pw1"}, 201,
343 {"Location": "/admin/v1/users/", "Content-Type": "application/json"}, "json")
344 engine.test("PU6", "Create user 2", "POST", "/admin/v1/users", headers_json,
345 {"username": "U2", "projects": ["P1"], "password": "pw2"}, 201,
346 {"Location": "/admin/v1/users/", "Content-Type": "application/json"}, "json")
347
348 engine.test("PU7", "Edit user U1, delete P2 project", "PATCH", "/admin/v1/users/U1", headers_json,
349 {"projects": {"$'P2'": None}}, 204, None, None)
350 res = engine.test("PU1", "Check user U1, contains the right projects", "GET", "/admin/v1/users/U1",
351 headers_json, None, 200, None, json)
352 u1 = res.json()
353 # print(u1)
354 expected_projects = ["P1", "Padmin"]
355 if u1["projects"] != expected_projects:
356 raise TestException("User content projects '{}' different than expected '{}'. Edition has not done"
357 " properly".format(u1["projects"], expected_projects))
358
359 engine.test("PU8", "Edit user U1, set Padmin as default project", "PUT", "/admin/v1/users/U1", headers_json,
360 {"projects": {"$'Padmin'": None, "$+[0]": "Padmin"}}, 204, None, None)
361 res = engine.test("PU1", "Check user U1, contains the right projects", "GET", "/admin/v1/users/U1",
362 headers_json, None, 200, None, json)
363 u1 = res.json()
364 # print(u1)
365 expected_projects = ["Padmin", "P1"]
366 if u1["projects"] != expected_projects:
367 raise TestException("User content projects '{}' different than expected '{}'. Edition has not done"
368 " properly".format(u1["projects"], expected_projects))
369
370 engine.test("PU9", "Edit user U1, change password", "PATCH", "/admin/v1/users/U1", headers_json,
371 {"password": "pw1_new"}, 204, None, None)
372
373 engine.test("PU10", "Change to project P1 non existing", "POST", "/admin/v1/tokens/", headers_json,
374 {"project_id": "P1"}, 401, r_header_json, "json")
375
376 res = engine.test("PU1", "Change to user U1 project P1", "POST", "/admin/v1/tokens", headers_json,
377 {"username": "U1", "password": "pw1_new", "project_id": "P1"}, (200, 201),
378 r_header_json, "json")
379 response = res.json()
380 engine.set_header({"Authorization": "Bearer {}".format(response["id"])})
381
382 engine.test("PU11", "Edit user projects non admin", "PUT", "/admin/v1/users/U1", headers_json,
383 {"projects": {"$'P1'": None}}, 401, r_header_json, "json")
384 engine.test("PU12", "Add new project non admin", "POST", "/admin/v1/projects", headers_json,
385 {"name": "P2"}, 401, r_header_json, "json")
386 engine.test("PU13", "Add new user non admin", "POST", "/admin/v1/users", headers_json,
387 {"username": "U3", "projects": ["P1"], "password": "pw3"}, 401,
388 r_header_json, "json")
389
390 res = engine.test("PU14", "Change to user U1 project Padmin", "POST", "/admin/v1/tokens", headers_json,
391 {"project_id": "Padmin"}, (200, 201), r_header_json, "json")
392 response = res.json()
393 engine.set_header({"Authorization": "Bearer {}".format(response["id"])})
394
395 engine.test("PU15", "Add new project admin", "POST", "/admin/v1/projects", headers_json, {"name": "P2"},
396 (201, 204), {"Location": "/admin/v1/projects/", "Content-Type": "application/json"}, "json")
397 engine.test("PU16", "Add new user U3 admin", "POST", "/admin/v1/users",
398 headers_json, {"username": "U3", "projects": ["P2"], "password": "pw3"}, (201, 204),
399 {"Location": "/admin/v1/users/", "Content-Type": "application/json"}, "json")
400 engine.test("PU17", "Edit user projects admin", "PUT", "/admin/v1/users/U3", headers_json,
401 {"projects": ["P2"]}, 204, None, None)
402
403 engine.test("PU18", "Delete project P2 conflict", "DELETE", "/admin/v1/projects/P2", headers_json, None, 409,
404 r_header_json, "json")
405 engine.test("PU19", "Delete project P2 forcing", "DELETE", "/admin/v1/projects/P2?FORCE=True", headers_json,
406 None, 204, None, None)
407
408 engine.test("PU20", "Delete user U1. Conflict deleting own user", "DELETE", "/admin/v1/users/U1", headers_json,
409 None, 409, r_header_json, "json")
410 engine.test("PU21", "Delete user U2", "DELETE", "/admin/v1/users/U2", headers_json, None, 204, None, None)
411 engine.test("PU22", "Delete user U3", "DELETE", "/admin/v1/users/U3", headers_json, None, 204, None, None)
412 # change to admin
413 engine.remove_authorization() # To force get authorization
414 engine.get_autorization()
415 engine.test("PU23", "Delete user U1", "DELETE", "/admin/v1/users/U1", headers_json, None, 204, None, None)
416 engine.test("PU24", "Delete project P1", "DELETE", "/admin/v1/projects/P1", headers_json, None, 204, None, None)
417 engine.test("PU25", "Delete project Padmin", "DELETE", "/admin/v1/projects/Padmin", headers_json, None, 204,
418 None, None)
419
420
421 class TestFakeVim:
422 description = "Creates/edit/delete fake VIMs and SDN controllers"
423
424 def __init__(self):
425 self.vim = {
426 "schema_version": "1.0",
427 "schema_type": "No idea",
428 "name": "myVim",
429 "description": "Descriptor name",
430 "vim_type": "openstack",
431 "vim_url": "http://localhost:/vim",
432 "vim_tenant_name": "vimTenant",
433 "vim_user": "user",
434 "vim_password": "password",
435 "config": {"config_param": 1}
436 }
437 self.sdn = {
438 "name": "sdn-name",
439 "description": "sdn-description",
440 "dpid": "50:50:52:54:00:94:21:21",
441 "ip": "192.168.15.17",
442 "port": 8080,
443 "type": "opendaylight",
444 "version": "3.5.6",
445 "user": "user",
446 "password": "passwd"
447 }
448 self.port_mapping = [
449 {"compute_node": "compute node 1",
450 "ports": [{"pci": "0000:81:00.0", "switch_port": "port-2/1", "switch_mac": "52:54:00:94:21:21"},
451 {"pci": "0000:81:00.1", "switch_port": "port-2/2", "switch_mac": "52:54:00:94:21:22"}
452 ]},
453 {"compute_node": "compute node 2",
454 "ports": [{"pci": "0000:81:00.0", "switch_port": "port-2/3", "switch_mac": "52:54:00:94:21:23"},
455 {"pci": "0000:81:00.1", "switch_port": "port-2/4", "switch_mac": "52:54:00:94:21:24"}
456 ]}
457 ]
458
459 def run(self, engine, test_osm, manual_check, test_params=None):
460
461 vim_bad = self.vim.copy()
462 vim_bad.pop("name")
463
464 engine.get_autorization()
465 engine.test("FVIM1", "Create VIM", "POST", "/admin/v1/vim_accounts", headers_json, self.vim, (201, 204),
466 {"Location": "/admin/v1/vim_accounts/", "Content-Type": "application/json"}, "json")
467 engine.test("FVIM2", "Create VIM without name, bad schema", "POST", "/admin/v1/vim_accounts", headers_json,
468 vim_bad, 422, None, headers_json)
469 engine.test("FVIM3", "Create VIM name repeated", "POST", "/admin/v1/vim_accounts", headers_json, self.vim,
470 409, None, headers_json)
471 engine.test("FVIM4", "Show VIMs", "GET", "/admin/v1/vim_accounts", headers_yaml, None, 200, r_header_yaml,
472 "yaml")
473 engine.test("FVIM5", "Show VIM", "GET", "/admin/v1/vim_accounts/<FVIM1>", headers_yaml, None, 200,
474 r_header_yaml, "yaml")
475 if not test_osm:
476 # delete with FORCE
477 engine.test("FVIM6", "Delete VIM", "DELETE", "/admin/v1/vim_accounts/<FVIM1>?FORCE=True", headers_yaml,
478 None, 202, None, 0)
479 engine.test("FVIM7", "Check VIM is deleted", "GET", "/admin/v1/vim_accounts/<FVIM1>", headers_yaml, None,
480 404, r_header_yaml, "yaml")
481 else:
482 # delete and wait until is really deleted
483 engine.test("FVIM6", "Delete VIM", "DELETE", "/admin/v1/vim_accounts/<FVIM1>", headers_yaml, None, 202,
484 None, 0)
485 wait = timeout
486 while wait >= 0:
487 r = engine.test("FVIM7", "Check VIM is deleted", "GET", "/admin/v1/vim_accounts/<FVIM1>", headers_yaml,
488 None, None, r_header_yaml, "yaml")
489 if r.status_code == 404:
490 break
491 elif r.status_code == 200:
492 wait -= 5
493 sleep(5)
494 else:
495 raise TestException("Vim created at 'FVIM1' is not delete after {} seconds".format(timeout))
496
497
498 class TestVIMSDN(TestFakeVim):
499 description = "Creates VIM with SDN editing SDN controllers and port_mapping"
500
501 def __init__(self):
502 TestFakeVim.__init__(self)
503
504 def run(self, engine, test_osm, manual_check, test_params=None):
505 engine.get_autorization()
506 # Added SDN
507 engine.test("VIMSDN1", "Create SDN", "POST", "/admin/v1/sdns", headers_json, self.sdn, (201, 204),
508 {"Location": "/admin/v1/sdns/", "Content-Type": "application/json"}, "json")
509 # sleep(5)
510 # Edit SDN
511 engine.test("VIMSDN2", "Edit SDN", "PATCH", "/admin/v1/sdns/<VIMSDN1>", headers_json, {"name": "new_sdn_name"},
512 204, None, None)
513 # sleep(5)
514 # VIM with SDN
515 self.vim["config"]["sdn-controller"] = engine.test_ids["VIMSDN1"]
516 self.vim["config"]["sdn-port-mapping"] = self.port_mapping
517 engine.test("VIMSDN3", "Create VIM", "POST", "/admin/v1/vim_accounts", headers_json, self.vim, (200, 204, 201),
518 {"Location": "/admin/v1/vim_accounts/", "Content-Type": "application/json"}, "json"),
519
520 self.port_mapping[0]["compute_node"] = "compute node XX"
521 engine.test("VIMSDN4", "Edit VIM change port-mapping", "PUT", "/admin/v1/vim_accounts/<VIMSDN3>", headers_json,
522 {"config": {"sdn-port-mapping": self.port_mapping}}, 204, None, None)
523 engine.test("VIMSDN5", "Edit VIM remove port-mapping", "PUT", "/admin/v1/vim_accounts/<VIMSDN3>", headers_json,
524 {"config": {"sdn-port-mapping": None}}, 204, None, None)
525
526 if not test_osm:
527 # delete with FORCE
528 engine.test("VIMSDN6", "Delete VIM remove port-mapping", "DELETE",
529 "/admin/v1/vim_accounts/<VIMSDN3>?FORCE=True", headers_json, None, 202, None, 0)
530 engine.test("VIMSDN7", "Delete SDNC", "DELETE", "/admin/v1/sdns/<VIMSDN1>?FORCE=True", headers_json, None,
531 202, None, 0)
532
533 engine.test("VIMSDN8", "Check VIM is deleted", "GET", "/admin/v1/vim_accounts/<VIMSDN3>", headers_yaml,
534 None, 404, r_header_yaml, "yaml")
535 engine.test("VIMSDN9", "Check SDN is deleted", "GET", "/admin/v1/sdns/<VIMSDN1>", headers_yaml, None,
536 404, r_header_yaml, "yaml")
537 else:
538 # delete and wait until is really deleted
539 engine.test("VIMSDN6", "Delete VIM remove port-mapping", "DELETE", "/admin/v1/vim_accounts/<VIMSDN3>",
540 headers_json, None, (202, 201, 204), None, 0)
541 engine.test("VIMSDN7", "Delete SDN", "DELETE", "/admin/v1/sdns/<VIMSDN1>", headers_json, None,
542 (202, 201, 204), None, 0)
543 wait = timeout
544 while wait >= 0:
545 r = engine.test("VIMSDN8", "Check VIM is deleted", "GET", "/admin/v1/vim_accounts/<VIMSDN3>",
546 headers_yaml, None, None, r_header_yaml, "yaml")
547 if r.status_code == 404:
548 break
549 elif r.status_code == 200:
550 wait -= 5
551 sleep(5)
552 else:
553 raise TestException("Vim created at 'VIMSDN3' is not delete after {} seconds".format(timeout))
554 while wait >= 0:
555 r = engine.test("VIMSDN9", "Check SDNC is deleted", "GET", "/admin/v1/sdns/<VIMSDN1>",
556 headers_yaml, None, None, r_header_yaml, "yaml")
557 if r.status_code == 404:
558 break
559 elif r.status_code == 200:
560 wait -= 5
561 sleep(5)
562 else:
563 raise TestException("SDNC created at 'VIMSDN1' is not delete after {} seconds".format(timeout))
564
565
566 class TestDeploy:
567 description = "Base class for downloading descriptors from ETSI, onboard and deploy in real VIM"
568
569 def __init__(self):
570 self.step = 0
571 self.nsd_id = None
572 self.vim_id = None
573 self.nsd_test = None
574 self.ns_test = None
575 self.ns_id = None
576 self.vnfds_test = []
577 self.descriptor_url = "https://osm-download.etsi.org/ftp/osm-3.0-three/2nd-hackfest/packages/"
578 self.vnfd_filenames = ("cirros_vnf.tar.gz",)
579 self.nsd_filename = "cirros_2vnf_ns.tar.gz"
580 self.uses_configuration = False
581 self.uss = {}
582 self.passwds = {}
583 self.cmds = {}
584 self.keys = {}
585 self.timeout = 120
586 self.passed_tests = 0
587 self.total_tests = 0
588
589 def create_descriptors(self, engine):
590 temp_dir = os.path.dirname(os.path.abspath(__file__)) + "/temp/"
591 if not os.path.exists(temp_dir):
592 os.makedirs(temp_dir)
593 for vnfd_filename in self.vnfd_filenames:
594 if "/" in vnfd_filename:
595 vnfd_filename_path = vnfd_filename
596 if not os.path.exists(vnfd_filename_path):
597 raise TestException("File '{}' does not exist".format(vnfd_filename_path))
598 else:
599 vnfd_filename_path = temp_dir + vnfd_filename
600 if not os.path.exists(vnfd_filename_path):
601 with open(vnfd_filename_path, "wb") as file:
602 response = requests.get(self.descriptor_url + vnfd_filename)
603 if response.status_code >= 300:
604 raise TestException("Error downloading descriptor from '{}': {}".format(
605 self.descriptor_url + vnfd_filename, response.status_code))
606 file.write(response.content)
607 if vnfd_filename_path.endswith(".yaml"):
608 headers = headers_yaml
609 else:
610 headers = headers_zip_yaml
611 if self.step % 2 == 0:
612 # vnfd CREATE AND UPLOAD in one step:
613 engine.test("DEPLOY{}".format(self.step), "Onboard VNFD in one step", "POST",
614 "/vnfpkgm/v1/vnf_packages_content", headers, "@b" + vnfd_filename_path, 201,
615 {"Location": "/vnfpkgm/v1/vnf_packages_content/", "Content-Type": "application/yaml"}, yaml)
616 self.vnfds_test.append("DEPLOY" + str(self.step))
617 self.step += 1
618 else:
619 # vnfd CREATE AND UPLOAD ZIP
620 engine.test("DEPLOY{}".format(self.step), "Onboard VNFD step 1", "POST", "/vnfpkgm/v1/vnf_packages",
621 headers_json, None, 201,
622 {"Location": "/vnfpkgm/v1/vnf_packages/", "Content-Type": "application/json"}, "json")
623 self.vnfds_test.append("DEPLOY" + str(self.step))
624 self.step += 1
625 # location = r.headers["Location"]
626 # vnfd_id = location[location.rfind("/")+1:]
627 engine.test("DEPLOY{}".format(self.step), "Onboard VNFD step 2 as ZIP", "PUT",
628 "/vnfpkgm/v1/vnf_packages/<>/package_content",
629 headers, "@b" + vnfd_filename_path, 204, None, 0)
630 self.step += 2
631
632 if "/" in self.nsd_filename:
633 nsd_filename_path = self.nsd_filename
634 if not os.path.exists(nsd_filename_path):
635 raise TestException("File '{}' does not exist".format(nsd_filename_path))
636 else:
637 nsd_filename_path = temp_dir + self.nsd_filename
638 if not os.path.exists(nsd_filename_path):
639 with open(nsd_filename_path, "wb") as file:
640 response = requests.get(self.descriptor_url + self.nsd_filename)
641 if response.status_code >= 300:
642 raise TestException("Error downloading descriptor from '{}': {}".format(
643 self.descriptor_url + self.nsd_filename, response.status_code))
644 file.write(response.content)
645 if nsd_filename_path.endswith(".yaml"):
646 headers = headers_yaml
647 else:
648 headers = headers_zip_yaml
649
650 self.nsd_test = "DEPLOY" + str(self.step)
651 if self.step % 2 == 0:
652 # nsd CREATE AND UPLOAD in one step:
653 engine.test("DEPLOY{}".format(self.step), "Onboard NSD in one step", "POST",
654 "/nsd/v1/ns_descriptors_content", headers, "@b" + nsd_filename_path, 201,
655 {"Location": "/nsd/v1/ns_descriptors_content/", "Content-Type": "application/yaml"}, yaml)
656 self.step += 1
657 else:
658 # nsd CREATE AND UPLOAD ZIP
659 engine.test("DEPLOY{}".format(self.step), "Onboard NSD step 1", "POST", "/nsd/v1/ns_descriptors",
660 headers_json, None, 201,
661 {"Location": "/nsd/v1/ns_descriptors/", "Content-Type": "application/json"}, "json")
662 self.step += 1
663 # location = r.headers["Location"]
664 # vnfd_id = location[location.rfind("/")+1:]
665 engine.test("DEPLOY{}".format(self.step), "Onboard NSD step 2 as ZIP", "PUT",
666 "/nsd/v1/ns_descriptors/<>/nsd_content",
667 headers, "@b" + nsd_filename_path, 204, None, 0)
668 self.step += 2
669 self.nsd_id = engine.test_ids[self.nsd_test]
670
671 def delete_descriptors(self, engine):
672 # delete descriptors
673 engine.test("DEPLOY{}".format(self.step), "Delete NSSD SOL005", "DELETE",
674 "/nsd/v1/ns_descriptors/<{}>".format(self.nsd_test),
675 headers_yaml, None, 204, None, 0)
676 self.step += 1
677 for vnfd_test in self.vnfds_test:
678 engine.test("DEPLOY{}".format(self.step), "Delete VNFD SOL005", "DELETE",
679 "/vnfpkgm/v1/vnf_packages/<{}>".format(vnfd_test), headers_yaml, None, 204, None, 0)
680 self.step += 1
681
682 def instantiate(self, engine, ns_data):
683 ns_data_text = yaml.safe_dump(ns_data, default_flow_style=True, width=256)
684 # create NS Two steps
685 r = engine.test("DEPLOY{}".format(self.step), "Create NS step 1", "POST", "/nslcm/v1/ns_instances",
686 headers_yaml, ns_data_text, 201,
687 {"Location": "nslcm/v1/ns_instances/", "Content-Type": "application/yaml"}, "yaml")
688 self.ns_test = "DEPLOY{}".format(self.step)
689 self.ns_id = engine.test_ids[self.ns_test]
690 engine.test_ids[self.ns_test]
691 self.step += 1
692 r = engine.test("DEPLOY{}".format(self.step), "Instantiate NS step 2", "POST",
693 "/nslcm/v1/ns_instances/<{}>/instantiate".format(self.ns_test), headers_yaml, ns_data_text,
694 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
695 nslcmop_test = "DEPLOY{}".format(self.step)
696 self.step += 1
697
698 if test_osm:
699 # Wait until status is Ok
700 wait = timeout_configure if self.uses_configuration else timeout_deploy
701 while wait >= 0:
702 r = engine.test("DEPLOY{}".format(self.step), "Wait until NS is deployed and configured", "GET",
703 "/nslcm/v1/ns_lcm_op_occs/<{}>".format(nslcmop_test), headers_json, None,
704 200, r_header_json, "json")
705 nslcmop = r.json()
706 if "COMPLETED" in nslcmop["operationState"]:
707 break
708 elif "FAILED" in nslcmop["operationState"]:
709 raise TestException("NS instantiate has failed: {}".format(nslcmop["detailed-status"]))
710 wait -= 5
711 sleep(5)
712 else:
713 raise TestException("NS instantiate is not done after {} seconds".format(timeout_deploy))
714 self.step += 1
715
716 def _wait_nslcmop_ready(self, engine, nslcmop_test, timeout_deploy):
717 wait = timeout
718 while wait >= 0:
719 r = engine.test("DEPLOY{}".format(self.step), "Wait to ns lcm operation complete", "GET",
720 "/nslcm/v1/ns_lcm_op_occs/<{}>".format(nslcmop_test), headers_json, None,
721 200, r_header_json, "json")
722 nslcmop = r.json()
723 if "COMPLETED" in nslcmop["operationState"]:
724 break
725 elif "FAILED" in nslcmop["operationState"]:
726 raise TestException("NS terminate has failed: {}".format(nslcmop["detailed-status"]))
727 wait -= 5
728 sleep(5)
729 else:
730 raise TestException("NS instantiate is not terminate after {} seconds".format(timeout))
731
732 def terminate(self, engine):
733 # remove deployment
734 if test_osm:
735 r = engine.test("DEPLOY{}".format(self.step), "Terminate NS", "POST",
736 "/nslcm/v1/ns_instances/<{}>/terminate".format(self.ns_test), headers_yaml, None,
737 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
738 nslcmop2_test = "DEPLOY{}".format(self.step)
739 self.step += 1
740 # Wait until status is Ok
741 self._wait_nslcmop_ready(engine, nslcmop2_test, timeout_deploy)
742
743 r = engine.test("DEPLOY{}".format(self.step), "Delete NS", "DELETE",
744 "/nslcm/v1/ns_instances/<{}>".format(self.ns_test), headers_yaml, None,
745 204, None, 0)
746 self.step += 1
747 else:
748 r = engine.test("DEPLOY{}".format(self.step), "Delete NS with FORCE", "DELETE",
749 "/nslcm/v1/ns_instances/<{}>?FORCE=True".format(self.ns_test), headers_yaml, None,
750 204, None, 0)
751 self.step += 1
752
753 # check all it is deleted
754 r = engine.test("DEPLOY{}".format(self.step), "Check NS is deleted", "GET",
755 "/nslcm/v1/ns_instances/<{}>".format(self.ns_test), headers_yaml, None,
756 404, None, "yaml")
757 self.step += 1
758 r = engine.test("DEPLOY{}".format(self.step), "Check NSLCMOPs are deleted", "GET",
759 "/nslcm/v1/ns_lcm_op_occs?nsInstanceId=<{}>".format(self.ns_test), headers_json, None,
760 200, None, "json")
761 nslcmops = r.json()
762 if not isinstance(nslcmops, list) or nslcmops:
763 raise TestException("NS {} deleted but with ns_lcm_op_occ active: {}".format(self.ns_test, nslcmops))
764
765 def test_ns(self, engine, test_osm, commands=None, users=None, passwds=None, keys=None, timeout=0):
766
767 n = 0
768 r = engine.test("TEST_NS{}".format(n), "GET VNFR_IDs", "GET",
769 "/nslcm/v1/ns_instances/{}".format(self.ns_id), headers_json, None,
770 200, r_header_json, "json")
771 n += 1
772 ns_data = r.json()
773
774 vnfr_list = ns_data['constituent-vnfr-ref']
775 time = 0
776
777 for vnfr_id in vnfr_list:
778 self.total_tests += 1
779 r = engine.test("TEST_NS{}".format(n), "GET IP_ADDRESS OF VNFR", "GET",
780 "/nslcm/v1/vnfrs/{}".format(vnfr_id), headers_json, None,
781 200, r_header_json, "json")
782 n += 1
783 vnfr_data = r.json()
784
785 if vnfr_data.get("ip-address"):
786 name = "TEST_NS{}".format(n)
787 description = "Run tests in VNFR with IP {}".format(vnfr_data['ip-address'])
788 n += 1
789 test_description = "Test {} {}".format(name, description)
790 logger.warning(test_description)
791 vnf_index = str(vnfr_data["member-vnf-index-ref"])
792 while timeout >= time:
793 result, message = self.do_checks([vnfr_data["ip-address"]],
794 vnf_index=vnfr_data["member-vnf-index-ref"],
795 commands=commands.get(vnf_index), user=users.get(vnf_index),
796 passwd=passwds.get(vnf_index), key=keys.get(vnf_index))
797 if result == 1:
798 logger.warning(message)
799 break
800 elif result == 0:
801 time += 20
802 sleep(20)
803 elif result == -1:
804 logger.critical(message)
805 break
806 else:
807 time -= 20
808 logger.critical(message)
809 else:
810 logger.critical("VNFR {} has not mgmt address. Check failed".format(vnfr_id))
811
812 def do_checks(self, ip, vnf_index, commands=[], user=None, passwd=None, key=None):
813 try:
814 import urllib3
815 from pssh.clients import ParallelSSHClient
816 from pssh.utils import load_private_key
817 from ssh2 import exceptions as ssh2Exception
818 except ImportError as e:
819 logger.critical("package <pssh> or/and <urllib3> is not installed. Please add it with 'pip3 install "
820 "parallel-ssh' and/or 'pip3 install urllib3': {}".format(e))
821 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
822 try:
823 p_host = os.environ.get("PROXY_HOST")
824 p_user = os.environ.get("PROXY_USER")
825 p_password = os.environ.get("PROXY_PASSWD")
826
827 if key:
828 pkey = load_private_key(key)
829 else:
830 pkey = None
831
832 client = ParallelSSHClient(ip, user=user, password=passwd, pkey=pkey, proxy_host=p_host,
833 proxy_user=p_user, proxy_password=p_password, timeout=10, num_retries=0)
834 for cmd in commands:
835 output = client.run_command(cmd)
836 client.join(output)
837 if output[ip[0]].exit_code:
838 return -1, " VNFR {} could not be checked: {}".format(ip[0], output[ip[0]].stderr)
839 else:
840 self.passed_tests += 1
841 return 1, " Test successful"
842 except (ssh2Exception.ChannelFailure, ssh2Exception.SocketDisconnectError, ssh2Exception.SocketTimeout,
843 ssh2Exception.SocketRecvError) as e:
844 return 0, "Timeout accessing the VNFR {}: {}".format(ip[0], str(e))
845 except Exception as e:
846 return -1, "ERROR checking the VNFR {}: {}".format(ip[0], str(e))
847
848 def aditional_operations(self, engine, test_osm, manual_check):
849 pass
850
851 def run(self, engine, test_osm, manual_check, test_params=None):
852 engine.get_autorization()
853 nsname = os.environ.get("OSMNBITEST_NS_NAME", "OSMNBITEST")
854 if test_params:
855 if "vnfd-files" in test_params:
856 self.vnfd_filenames = test_params["vnfd-files"].split(",")
857 if "nsd-file" in test_params:
858 self.nsd_filename = test_params["nsd-file"]
859 if test_params.get("ns-name"):
860 nsname = test_params["ns-name"]
861 self.create_descriptors(engine)
862
863 # create real VIM if not exist
864 self.vim_id = engine.get_create_vim(test_osm)
865 ns_data = {"nsDescription": "default description", "nsName": nsname, "nsdId": self.nsd_id,
866 "vimAccountId": self.vim_id}
867 if test_params and test_params.get("ns-config"):
868 if isinstance(test_params["ns-config"], str):
869 ns_data.update(yaml.load(test_params["ns-config"]))
870 else:
871 ns_data.update(test_params["ns-config"])
872 self.instantiate(engine, ns_data)
873
874 if manual_check:
875 input('NS has been deployed. Perform manual check and press enter to resume')
876 else:
877 self.test_ns(engine, test_osm, self.cmds, self.uss, self.pss, self.keys, self.timeout)
878 self.aditional_operations(engine, test_osm, manual_check)
879 self.terminate(engine)
880 self.delete_descriptors(engine)
881 self.print_results()
882
883 def print_results(self):
884 print("\n\n\n--------------------------------------------")
885 print("TEST RESULTS:\n PASSED TESTS: {} - TOTAL TESTS: {}".format(self.total_tests, self.passed_tests))
886 print("--------------------------------------------")
887
888
889 class TestDeployHackfestCirros(TestDeploy):
890 description = "Load and deploy Hackfest cirros_2vnf_ns example"
891
892 def __init__(self):
893 super().__init__()
894 self.vnfd_filenames = ("cirros_vnf.tar.gz",)
895 self.nsd_filename = "cirros_2vnf_ns.tar.gz"
896 self.cmds = {'1': ['ls -lrt', ], '2': ['ls -lrt', ]}
897 self.uss = {'1': "cirros", '2': "cirros"}
898 self.pss = {'1': "cubswin:)", '2': "cubswin:)"}
899
900
901 class TestDeployIpMac(TestDeploy):
902 description = "Load and deploy descriptor examples setting mac, ip address at descriptor and instantiate params"
903
904 def __init__(self):
905 super().__init__()
906 self.vnfd_filenames = ("vnfd_2vdu_set_ip_mac2.yaml", "vnfd_2vdu_set_ip_mac.yaml")
907 self.nsd_filename = "scenario_2vdu_set_ip_mac.yaml"
908 self.descriptor_url = \
909 "https://osm.etsi.org/gitweb/?p=osm/RO.git;a=blob_plain;f=test/RO_tests/v3_2vdu_set_ip_mac/"
910 self.cmds = {'1': ['ls -lrt', ], '2': ['ls -lrt', ]}
911 self.uss = {'1': "osm", '2': "osm"}
912 self.pss = {'1': "osm4u", '2': "osm4u"}
913 self.timeout = 360
914
915 def run(self, engine, test_osm, manual_check, test_params=None):
916 # super().run(engine, test_osm, manual_check, test_params)
917 # run again setting IPs with instantiate parameters
918 instantiation_params = {
919 "vnf": [
920 {
921 "member-vnf-index": "1",
922 "internal-vld": [
923 {
924 "name": "internal_vld1", # net_internal
925 "ip-profile": {
926 "ip-version": "ipv4",
927 "subnet-address": "10.9.8.0/24",
928 "dhcp-params": {"count": 100, "start-address": "10.9.8.100"}
929 },
930 "internal-connection-point": [
931 {
932 "id-ref": "eth2",
933 "ip-address": "10.9.8.2",
934 },
935 {
936 "id-ref": "eth3",
937 "ip-address": "10.9.8.3",
938 }
939 ]
940 },
941 ],
942
943 "vdu": [
944 {
945 "id": "VM1",
946 "interface": [
947 # {
948 # "name": "iface11",
949 # "floating-ip-required": True,
950 # },
951 {
952 "name": "iface13",
953 "mac-address": "52:33:44:55:66:13"
954 },
955 ],
956 },
957 {
958 "id": "VM2",
959 "interface": [
960 {
961 "name": "iface21",
962 "ip-address": "10.31.31.22",
963 "mac-address": "52:33:44:55:66:21"
964 },
965 ],
966 },
967 ]
968 },
969 ]
970 }
971
972 super().run(engine, test_osm, manual_check, test_params={"ns-config": instantiation_params})
973
974
975 class TestDeployHackfest4(TestDeploy):
976 description = "Load and deploy Hackfest 4 example."
977
978 def __init__(self):
979 super().__init__()
980 self.vnfd_filenames = ("hackfest_4_vnfd.tar.gz",)
981 self.nsd_filename = "hackfest_4_nsd.tar.gz"
982 self.uses_configuration = True
983 self.cmds = {'1': ['ls -lrt', ], '2': ['ls -lrt', ]}
984 self.uss = {'1': "ubuntu", '2': "ubuntu"}
985 self.pss = {'1': "osm4u", '2': "osm4u"}
986
987 def create_descriptors(self, engine):
988 super().create_descriptors(engine)
989 # Modify VNFD to add scaling
990 payload = """
991 scaling-group-descriptor:
992 - name: "scale_dataVM"
993 max-instance-count: 10
994 scaling-policy:
995 - name: "auto_cpu_util_above_threshold"
996 scaling-type: "automatic"
997 threshold-time: 0
998 cooldown-time: 60
999 scaling-criteria:
1000 - name: "cpu_util_above_threshold"
1001 scale-in-threshold: 15
1002 scale-in-relational-operation: "LE"
1003 scale-out-threshold: 60
1004 scale-out-relational-operation: "GE"
1005 vnf-monitoring-param-ref: "all_aaa_cpu_util"
1006 vdu:
1007 - vdu-id-ref: dataVM
1008 count: 1
1009 scaling-config-action:
1010 - trigger: post-scale-out
1011 vnf-config-primitive-name-ref: touch
1012 - trigger: pre-scale-in
1013 vnf-config-primitive-name-ref: touch
1014 vnf-configuration:
1015 config-primitive:
1016 - name: touch
1017 parameter:
1018 - name: filename
1019 data-type: STRING
1020 default-value: '/home/ubuntu/touched'
1021 """
1022 engine.test("DEPLOY{}".format(self.step), "Edit VNFD ", "PATCH",
1023 "/vnfpkgm/v1/vnf_packages/<{}>".format(self.vnfds_test[0]), headers_yaml, payload, 204, None, None)
1024 self.step += 1
1025
1026
1027 class TestDeployHackfest3Charmed(TestDeploy):
1028 description = "Load and deploy Hackfest 3charmed_ns example. Modifies it for adding scaling and performs " \
1029 "primitive actions and scaling"
1030
1031 def __init__(self):
1032 super().__init__()
1033 self.vnfd_filenames = ("hackfest_3charmed_vnfd.tar.gz",)
1034 self.nsd_filename = "hackfest_3charmed_nsd.tar.gz"
1035 self.uses_configuration = True
1036 self.cmds = {'1': [''], '2': ['ls -lrt /home/ubuntu/first-touch', ]}
1037 self.uss = {'1': "ubuntu", '2': "ubuntu"}
1038 self.pss = {'1': "osm4u", '2': "osm4u"}
1039
1040 # def create_descriptors(self, engine):
1041 # super().create_descriptors(engine)
1042 # # Modify VNFD to add scaling
1043 # payload = """
1044 # scaling-group-descriptor:
1045 # - name: "scale_dataVM"
1046 # max-instance-count: 10
1047 # scaling-policy:
1048 # - name: "auto_cpu_util_above_threshold"
1049 # scaling-type: "automatic"
1050 # threshold-time: 0
1051 # cooldown-time: 60
1052 # scaling-criteria:
1053 # - name: "cpu_util_above_threshold"
1054 # scale-in-threshold: 15
1055 # scale-in-relational-operation: "LE"
1056 # scale-out-threshold: 60
1057 # scale-out-relational-operation: "GE"
1058 # vnf-monitoring-param-ref: "all_aaa_cpu_util"
1059 # vdu:
1060 # - vdu-id-ref: dataVM
1061 # count: 1
1062 # scaling-config-action:
1063 # - trigger: post-scale-out
1064 # vnf-config-primitive-name-ref: touch
1065 # - trigger: pre-scale-in
1066 # vnf-config-primitive-name-ref: touch
1067 # vnf-configuration:
1068 # config-primitive:
1069 # - name: touch
1070 # parameter:
1071 # - name: filename
1072 # data-type: STRING
1073 # default-value: '/home/ubuntu/touched'
1074 # """
1075 # engine.test("DEPLOY{}".format(self.step), "Edit VNFD ", "PATCH",
1076 # "/vnfpkgm/v1/vnf_packages/<{}>".format(self.vnfds_test[0]),
1077 # headers_yaml, payload, 200,
1078 # r_header_yaml, yaml)
1079 # self.vnfds_test.append("DEPLOY" + str(self.step))
1080 # self.step += 1
1081
1082 def aditional_operations(self, engine, test_osm, manual_check):
1083 if not test_osm:
1084 return
1085 # 1 perform action
1086 payload = '{member_vnf_index: "2", primitive: touch, primitive_params: { filename: /home/ubuntu/OSMTESTNBI }}'
1087 engine.test("DEPLOY{}".format(self.step), "Executer service primitive over NS", "POST",
1088 "/nslcm/v1/ns_instances/<{}>/action".format(self.ns_test), headers_yaml, payload,
1089 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
1090 nslcmop2_action = "DEPLOY{}".format(self.step)
1091 self.step += 1
1092 # Wait until status is Ok
1093 self._wait_nslcmop_ready(engine, nslcmop2_action, timeout_deploy)
1094 if manual_check:
1095 input('NS service primitive has been executed. Check that file /home/ubuntu/OSMTESTNBI is present at '
1096 'TODO_PUT_IP')
1097 else:
1098 cmds = {'1': [''], '2': ['ls -lrt /home/ubuntu/OSMTESTNBI', ]}
1099 uss = {'1': "ubuntu", '2': "ubuntu"}
1100 pss = {'1': "osm4u", '2': "osm4u"}
1101 self.test_ns(engine, test_osm, cmds, uss, pss, self.keys, self.timeout)
1102
1103 # # 2 perform scale out
1104 # payload = '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_OUT, scaleByStepData: ' \
1105 # '{scaling-group-descriptor: scale_dataVM, member-vnf-index: "1"}}}'
1106 # engine.test("DEPLOY{}".format(self.step), "Execute scale action over NS", "POST",
1107 # "/nslcm/v1/ns_instances/<{}>/scale".format(self.ns_test), headers_yaml, payload,
1108 # 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
1109 # nslcmop2_scale_out = "DEPLOY{}".format(self.step)
1110 # self._wait_nslcmop_ready(engine, nslcmop2_scale_out, timeout_deploy)
1111 # if manual_check:
1112 # input('NS scale out done. Check that file /home/ubuntu/touched is present and new VM is created')
1113 # # TODO check automatic
1114 #
1115 # # 2 perform scale in
1116 # payload = '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_IN, scaleByStepData: ' \
1117 # '{scaling-group-descriptor: scale_dataVM, member-vnf-index: "1"}}}'
1118 # engine.test("DEPLOY{}".format(self.step), "Execute scale action over NS", "POST",
1119 # "/nslcm/v1/ns_instances/<{}>/scale".format(self.ns_test), headers_yaml, payload,
1120 # 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
1121 # nslcmop2_scale_in = "DEPLOY{}".format(self.step)
1122 # self._wait_nslcmop_ready(engine, nslcmop2_scale_in, timeout_deploy)
1123 # if manual_check:
1124 # input('NS scale in done. Check that file /home/ubuntu/touched is updated and new VM is deleted')
1125 # # TODO check automatic
1126
1127
1128 if __name__ == "__main__":
1129 global logger
1130 test = ""
1131
1132 # Disable warnings from self-signed certificates.
1133 requests.packages.urllib3.disable_warnings()
1134 try:
1135 logging.basicConfig(format="%(levelname)s %(message)s", level=logging.ERROR)
1136 logger = logging.getLogger('NBI')
1137 # load parameters and configuration
1138 opts, args = getopt.getopt(sys.argv[1:], "hvu:p:",
1139 ["url=", "user=", "password=", "help", "version", "verbose", "no-verbose",
1140 "project=", "insecure", "timeout", "timeout-deploy", "timeout-configure",
1141 "test=", "list", "test-osm", "manual-check", "params="])
1142 url = "https://localhost:9999/osm"
1143 user = password = project = "admin"
1144 test_osm = False
1145 manual_check = False
1146 verbose = 0
1147 verify = True
1148 test_classes = {
1149 "NonAuthorized": TestNonAuthorized,
1150 "FakeVIM": TestFakeVim,
1151 "TestUsersProjects": TestUsersProjects,
1152 "VIM-SDN": TestVIMSDN,
1153 "Deploy-Custom": TestDeploy,
1154 "Deploy-Hackfest-Cirros": TestDeployHackfestCirros,
1155 "Deploy-Hackfest-3Charmed": TestDeployHackfest3Charmed,
1156 "Deploy-Hackfest-4": TestDeployHackfest4,
1157 "Deploy-CirrosMacIp": TestDeployIpMac,
1158 # "Deploy-MultiVIM": TestDeployMultiVIM,
1159 }
1160 test_to_do = []
1161 test_params = {}
1162
1163 for o, a in opts:
1164 # print("parameter:", o, a)
1165 if o == "--version":
1166 print("test version " + __version__ + ' ' + version_date)
1167 exit()
1168 elif o == "--list":
1169 for test, test_class in test_classes.items():
1170 print("{:20} {}".format(test + ":", test_class.description))
1171 exit()
1172 elif o in ("-v", "--verbose"):
1173 verbose += 1
1174 elif o == "no-verbose":
1175 verbose = -1
1176 elif o in ("-h", "--help"):
1177 usage()
1178 sys.exit()
1179 elif o == "--test-osm":
1180 test_osm = True
1181 elif o == "--manual-check":
1182 manual_check = True
1183 elif o == "--url":
1184 url = a
1185 elif o in ("-u", "--user"):
1186 user = a
1187 elif o in ("-p", "--password"):
1188 password = a
1189 elif o == "--project":
1190 project = a
1191 elif o == "--test":
1192 # print("asdfadf", o, a, a.split(","))
1193 for _test in a.split(","):
1194 if _test not in test_classes:
1195 print("Invalid test name '{}'. Use option '--list' to show available tests".format(_test),
1196 file=sys.stderr)
1197 exit(1)
1198 test_to_do.append(_test)
1199 elif o == "--params":
1200 param_key, _, param_value = a.partition("=")
1201 text_index = len(test_to_do)
1202 if text_index not in test_params:
1203 test_params[text_index] = {}
1204 test_params[text_index][param_key] = param_value
1205 elif o == "--insecure":
1206 verify = False
1207 elif o == "--timeout":
1208 timeout = int(a)
1209 elif o == "--timeout-deploy":
1210 timeout_deploy = int(a)
1211 elif o == "--timeout-configure":
1212 timeout_configure = int(a)
1213 else:
1214 assert False, "Unhandled option"
1215 if verbose == 0:
1216 logger.setLevel(logging.WARNING)
1217 elif verbose > 1:
1218 logger.setLevel(logging.DEBUG)
1219 else:
1220 logger.setLevel(logging.ERROR)
1221
1222 test_rest = TestRest(url, user=user, password=password, project=project)
1223 # print("tests to do:", test_to_do)
1224 if test_to_do:
1225 text_index = 0
1226 for test in test_to_do:
1227 text_index += 1
1228 test_class = test_classes[test]
1229 test_class().run(test_rest, test_osm, manual_check, test_params.get(text_index))
1230 else:
1231 for test, test_class in test_classes.items():
1232 test_class().run(test_rest, test_osm, manual_check, test_params.get(0))
1233 exit(0)
1234
1235 # get token
1236
1237 # # tests once authorized
1238 # for t in test_authorized_list:
1239 # test_rest.test(*t)
1240 #
1241 # # tests admin
1242 # for t in test_admin_list1:
1243 # test_rest.test(*t)
1244 #
1245 # # vnfd CREATE
1246 # r = test_rest.test("VNFD1", "Onboard VNFD step 1", "POST", "/vnfpkgm/v1/vnf_packages", headers_json, None,
1247 # 201, {"Location": "/vnfpkgm/v1/vnf_packages/", "Content-Type": "application/json"}, "json")
1248 # location = r.headers["Location"]
1249 # vnfd_id = location[location.rfind("/")+1:]
1250 # # print(location, vnfd_id)
1251 #
1252 # # vnfd UPLOAD test
1253 # r = test_rest.test("VNFD2", "Onboard VNFD step 2 as TEXT", "PUT",
1254 # "/vnfpkgm/v1/vnf_packages/{}/package_content".format(vnfd_id),
1255 # r_header_text, "@./cirros_vnf/cirros_vnfd.yaml", 204, None, 0)
1256 #
1257 # # vnfd SHOW OSM format
1258 # r = test_rest.test("VNFD3", "Show VNFD OSM format", "GET",
1259 # "/vnfpkgm/v1/vnf_packages_content/{}".format(vnfd_id),
1260 # headers_json, None, 200, r_header_json, "json")
1261 #
1262 # # vnfd SHOW text
1263 # r = test_rest.test("VNFD4", "Show VNFD SOL005 text", "GET",
1264 # "/vnfpkgm/v1/vnf_packages/{}/package_content".format(vnfd_id),
1265 # headers_text, None, 200, r_header_text, "text")
1266 #
1267 # # vnfd UPLOAD ZIP
1268 # makedirs("temp", exist_ok=True)
1269 # tar = tarfile.open("temp/cirros_vnf.tar.gz", "w:gz")
1270 # tar.add("cirros_vnf")
1271 # tar.close()
1272 # r = test_rest.test("VNFD5", "Onboard VNFD step 3 replace with ZIP", "PUT",
1273 # "/vnfpkgm/v1/vnf_packages/{}/package_content".format(vnfd_id),
1274 # r_header_zip, "@b./temp/cirros_vnf.tar.gz", 204, None, 0)
1275 #
1276 # # vnfd SHOW OSM format
1277 # r = test_rest.test("VNFD6", "Show VNFD OSM format", "GET",
1278 # "/vnfpkgm/v1/vnf_packages_content/{}".format(vnfd_id),
1279 # headers_json, None, 200, r_header_json, "json")
1280 #
1281 # # vnfd SHOW zip
1282 # r = test_rest.test("VNFD7", "Show VNFD SOL005 zip", "GET",
1283 # "/vnfpkgm/v1/vnf_packages/{}/package_content".format(vnfd_id),
1284 # headers_zip, None, 200, r_header_zip, "zip")
1285 # # vnfd SHOW descriptor
1286 # r = test_rest.test("VNFD8", "Show VNFD descriptor", "GET",
1287 # "/vnfpkgm/v1/vnf_packages/{}/vnfd".format(vnfd_id),
1288 # headers_text, None, 200, r_header_text, "text")
1289 # # vnfd SHOW actifact
1290 # r = test_rest.test("VNFD9", "Show VNFD artifact", "GET",
1291 # "/vnfpkgm/v1/vnf_packages/{}/artifacts/icons/cirros-64.png".format(vnfd_id),
1292 # headers_text, None, 200, r_header_octect, "text")
1293 #
1294 # # # vnfd DELETE
1295 # # r = test_rest.test("VNFD10", "Delete VNFD SOL005 text", "DELETE",
1296 # # "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_id),
1297 # # headers_yaml, None, 204, None, 0)
1298 #
1299 # # nsd CREATE
1300 # r = test_rest.test("NSD1", "Onboard NSD step 1", "POST", "/nsd/v1/ns_descriptors", headers_json, None,
1301 # 201, {"Location": "/nsd/v1/ns_descriptors/", "Content-Type": "application/json"}, "json")
1302 # location = r.headers["Location"]
1303 # nsd_id = location[location.rfind("/")+1:]
1304 # # print(location, nsd_id)
1305 #
1306 # # nsd UPLOAD test
1307 # r = test_rest.test("NSD2", "Onboard NSD with missing vnfd", "PUT",
1308 # "/nsd/v1/ns_descriptors/<>/nsd_content?constituent-vnfd.0.vnfd-id-ref"
1309 # "=NONEXISTING-VNFD".format(nsd_id),
1310 # r_header_text, "@./cirros_ns/cirros_nsd.yaml", 409, r_header_yaml, "yaml")
1311 #
1312 # # # VNF_CREATE
1313 # # r = test_rest.test("VNFD5", "Onboard VNFD step 3 replace with ZIP", "PUT",
1314 # # "/vnfpkgm/v1/vnf_packages/{}/package_content".format(vnfd_id),
1315 # # r_header_zip, "@b./temp/cirros_vnf.tar.gz", 204, None, 0)
1316 #
1317 # r = test_rest.test("NSD2", "Onboard NSD step 2 as TEXT", "PUT",
1318 # "/nsd/v1/ns_descriptors/{}/nsd_content".format(nsd_id),
1319 # r_header_text, "@./cirros_ns/cirros_nsd.yaml", 204, None, 0)
1320 #
1321 # # nsd SHOW OSM format
1322 # r = test_rest.test("NSD3", "Show NSD OSM format", "GET", "/nsd/v1/ns_descriptors_content/{}".format(nsd_id),
1323 # headers_json, None, 200, r_header_json, "json")
1324 #
1325 # # nsd SHOW text
1326 # r = test_rest.test("NSD4", "Show NSD SOL005 text", "GET",
1327 # "/nsd/v1/ns_descriptors/{}/nsd_content".format(nsd_id),
1328 # headers_text, None, 200, r_header_text, "text")
1329 #
1330 # # nsd UPLOAD ZIP
1331 # makedirs("temp", exist_ok=True)
1332 # tar = tarfile.open("temp/cirros_ns.tar.gz", "w:gz")
1333 # tar.add("cirros_ns")
1334 # tar.close()
1335 # r = test_rest.test("NSD5", "Onboard NSD step 3 replace with ZIP", "PUT",
1336 # "/nsd/v1/ns_descriptors/{}/nsd_content".format(nsd_id),
1337 # r_header_zip, "@b./temp/cirros_ns.tar.gz", 204, None, 0)
1338 #
1339 # # nsd SHOW OSM format
1340 # r = test_rest.test("NSD6", "Show NSD OSM format", "GET", "/nsd/v1/ns_descriptors_content/{}".format(nsd_id),
1341 # headers_json, None, 200, r_header_json, "json")
1342 #
1343 # # nsd SHOW zip
1344 # r = test_rest.test("NSD7","Show NSD SOL005 zip","GET", "/nsd/v1/ns_descriptors/{}/nsd_content".format(nsd_id),
1345 # headers_zip, None, 200, r_header_zip, "zip")
1346 #
1347 # # nsd SHOW descriptor
1348 # r = test_rest.test("NSD8", "Show NSD descriptor", "GET", "/nsd/v1/ns_descriptors/{}/nsd".format(nsd_id),
1349 # headers_text, None, 200, r_header_text, "text")
1350 # # nsd SHOW actifact
1351 # r = test_rest.test("NSD9", "Show NSD artifact", "GET",
1352 # "/nsd/v1/ns_descriptors/{}/artifacts/icons/osm_2x.png".format(nsd_id),
1353 # headers_text, None, 200, r_header_octect, "text")
1354 #
1355 # # vnfd DELETE
1356 # r = test_rest.test("VNFD10", "Delete VNFD conflict", "DELETE", "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_id),
1357 # headers_yaml, None, 409, r_header_yaml, "yaml")
1358 #
1359 # # nsd DELETE
1360 # r = test_rest.test("NSD10", "Delete NSD SOL005 text", "DELETE", "/nsd/v1/ns_descriptors/{}".format(nsd_id),
1361 # headers_yaml, None, 204, None, 0)
1362 #
1363 # # vnfd DELETE
1364 # r = test_rest.test("VNFD10","Delete VNFD SOL005 text","DELETE", "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_id),
1365 # headers_yaml, None, 204, None, 0)
1366 print("PASS")
1367
1368 except TestException as e:
1369 logger.error(test + "Test {} Exception: {}".format(test, str(e)))
1370 exit(1)
1371 except getopt.GetoptError as e:
1372 logger.error(e)
1373 print(e, file=sys.stderr)
1374 exit(1)
1375 except Exception as e:
1376 logger.critical(test + " Exception: " + str(e), exc_info=True)