projects / osm / NBI.git / blob
? search:
summary | shortlog | log | commit | commitdiff | tree
history | raw | HEAD
Allow user and project edition
[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.2"
18 version_date = "Jul 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.vnfds_test = []
576 self.descriptor_url = "https://osm-download.etsi.org/ftp/osm-3.0-three/2nd-hackfest/packages/"
577 self.vnfd_filenames = ("cirros_vnf.tar.gz",)
578 self.nsd_filename = "cirros_2vnf_ns.tar.gz"
579 self.uses_configuration = False
580
581 def create_descriptors(self, engine):
582 temp_dir = os.path.dirname(__file__) + "/temp/"
583 if not os.path.exists(temp_dir):
584 os.makedirs(temp_dir)
585 for vnfd_filename in self.vnfd_filenames:
586 if "/" in vnfd_filename:
587 vnfd_filename_path = vnfd_filename
588 if not os.path.exists(vnfd_filename_path):
589 raise TestException("File '{}' does not exist".format(vnfd_filename_path))
590 else:
591 vnfd_filename_path = temp_dir + vnfd_filename
592 if not os.path.exists(vnfd_filename_path):
593 with open(vnfd_filename_path, "wb") as file:
594 response = requests.get(self.descriptor_url + vnfd_filename)
595 if response.status_code >= 300:
596 raise TestException("Error downloading descriptor from '{}': {}".format(
597 self.descriptor_url + vnfd_filename, response.status_code))
598 file.write(response.content)
599 if vnfd_filename_path.endswith(".yaml"):
600 headers = headers_yaml
601 else:
602 headers = headers_zip_yaml
603 if self.step % 2 == 0:
604 # vnfd CREATE AND UPLOAD in one step:
605 engine.test("DEPLOY{}".format(self.step), "Onboard VNFD in one step", "POST",
606 "/vnfpkgm/v1/vnf_packages_content", headers, "@b" + vnfd_filename_path, 201,
607 {"Location": "/vnfpkgm/v1/vnf_packages_content/", "Content-Type": "application/yaml"}, yaml)
608 self.vnfds_test.append("DEPLOY" + str(self.step))
609 self.step += 1
610 else:
611 # vnfd CREATE AND UPLOAD ZIP
612 engine.test("DEPLOY{}".format(self.step), "Onboard VNFD step 1", "POST", "/vnfpkgm/v1/vnf_packages",
613 headers_json, None, 201,
614 {"Location": "/vnfpkgm/v1/vnf_packages/", "Content-Type": "application/json"}, "json")
615 self.vnfds_test.append("DEPLOY" + str(self.step))
616 self.step += 1
617 # location = r.headers["Location"]
618 # vnfd_id = location[location.rfind("/")+1:]
619 engine.test("DEPLOY{}".format(self.step), "Onboard VNFD step 2 as ZIP", "PUT",
620 "/vnfpkgm/v1/vnf_packages/<>/package_content",
621 headers, "@b" + vnfd_filename_path, 204, None, 0)
622 self.step += 2
623
624 if "/" in self.nsd_filename:
625 nsd_filename_path = self.nsd_filename
626 if not os.path.exists(nsd_filename_path):
627 raise TestException("File '{}' does not exist".format(nsd_filename_path))
628 else:
629 nsd_filename_path = temp_dir + self.nsd_filename
630 if not os.path.exists(nsd_filename_path):
631 with open(nsd_filename_path, "wb") as file:
632 response = requests.get(self.descriptor_url + self.nsd_filename)
633 if response.status_code >= 300:
634 raise TestException("Error downloading descriptor from '{}': {}".format(
635 self.descriptor_url + self.nsd_filename, response.status_code))
636 file.write(response.content)
637 if nsd_filename_path.endswith(".yaml"):
638 headers = headers_yaml
639 else:
640 headers = headers_zip_yaml
641
642 self.nsd_test = "DEPLOY" + str(self.step)
643 if self.step % 2 == 0:
644 # nsd CREATE AND UPLOAD in one step:
645 engine.test("DEPLOY{}".format(self.step), "Onboard NSD in one step", "POST",
646 "/nsd/v1/ns_descriptors_content", headers, "@b" + nsd_filename_path, 201,
647 {"Location": "/nsd/v1/ns_descriptors_content/", "Content-Type": "application/yaml"}, yaml)
648 self.step += 1
649 else:
650 # nsd CREATE AND UPLOAD ZIP
651 engine.test("DEPLOY{}".format(self.step), "Onboard NSD step 1", "POST", "/nsd/v1/ns_descriptors",
652 headers_json, None, 201,
653 {"Location": "/nsd/v1/ns_descriptors/", "Content-Type": "application/json"}, "json")
654 self.step += 1
655 # location = r.headers["Location"]
656 # vnfd_id = location[location.rfind("/")+1:]
657 engine.test("DEPLOY{}".format(self.step), "Onboard NSD step 2 as ZIP", "PUT",
658 "/nsd/v1/ns_descriptors/<>/nsd_content",
659 headers, "@b" + nsd_filename_path, 204, None, 0)
660 self.step += 2
661 self.nsd_id = engine.test_ids[self.nsd_test]
662
663 def delete_descriptors(self, engine):
664 # delete descriptors
665 engine.test("DEPLOY{}".format(self.step), "Delete NSSD SOL005", "DELETE",
666 "/nsd/v1/ns_descriptors/<{}>".format(self.nsd_test),
667 headers_yaml, None, 204, None, 0)
668 self.step += 1
669 for vnfd_test in self.vnfds_test:
670 engine.test("DEPLOY{}".format(self.step), "Delete VNFD SOL005", "DELETE",
671 "/vnfpkgm/v1/vnf_packages/<{}>".format(vnfd_test), headers_yaml, None, 204, None, 0)
672 self.step += 1
673
674 def instantiate(self, engine, ns_data):
675 ns_data_text = yaml.safe_dump(ns_data, default_flow_style=True, width=256)
676 # create NS Two steps
677 r = engine.test("DEPLOY{}".format(self.step), "Create NS step 1", "POST", "/nslcm/v1/ns_instances",
678 headers_yaml, ns_data_text, 201,
679 {"Location": "nslcm/v1/ns_instances/", "Content-Type": "application/yaml"}, "yaml")
680 self.ns_test = "DEPLOY{}".format(self.step)
681 engine.test_ids[self.ns_test]
682 self.step += 1
683 r = engine.test("DEPLOY{}".format(self.step), "Instantiate NS step 2", "POST",
684 "/nslcm/v1/ns_instances/<{}>/instantiate".format(self.ns_test), headers_yaml, ns_data_text,
685 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
686 nslcmop_test = "DEPLOY{}".format(self.step)
687 self.step += 1
688
689 if test_osm:
690 # Wait until status is Ok
691 wait = timeout_configure if self.uses_configuration else timeout_deploy
692 while wait >= 0:
693 r = engine.test("DEPLOY{}".format(self.step), "Wait until NS is deployed and configured", "GET",
694 "/nslcm/v1/ns_lcm_op_occs/<{}>".format(nslcmop_test), headers_json, None,
695 200, r_header_json, "json")
696 nslcmop = r.json()
697 if "COMPLETED" in nslcmop["operationState"]:
698 break
699 elif "FAILED" in nslcmop["operationState"]:
700 raise TestException("NS instantiate has failed: {}".format(nslcmop["detailed-status"]))
701 wait -= 5
702 sleep(5)
703 else:
704 raise TestException("NS instantiate is not done after {} seconds".format(timeout_deploy))
705 self.step += 1
706
707 def _wait_nslcmop_ready(self, engine, nslcmop_test, timeout_deploy):
708 wait = timeout
709 while wait >= 0:
710 r = engine.test("DEPLOY{}".format(self.step), "Wait to ns lcm operation complete", "GET",
711 "/nslcm/v1/ns_lcm_op_occs/<{}>".format(nslcmop_test), headers_json, None,
712 200, r_header_json, "json")
713 nslcmop = r.json()
714 if "COMPLETED" in nslcmop["operationState"]:
715 break
716 elif "FAILED" in nslcmop["operationState"]:
717 raise TestException("NS terminate has failed: {}".format(nslcmop["detailed-status"]))
718 wait -= 5
719 sleep(5)
720 else:
721 raise TestException("NS instantiate is not terminate after {} seconds".format(timeout))
722
723 def terminate(self, engine):
724 # remove deployment
725 if test_osm:
726 r = engine.test("DEPLOY{}".format(self.step), "Terminate NS", "POST",
727 "/nslcm/v1/ns_instances/<{}>/terminate".format(self.ns_test), headers_yaml, None,
728 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
729 nslcmop2_test = "DEPLOY{}".format(self.step)
730 self.step += 1
731 # Wait until status is Ok
732 self._wait_nslcmop_ready(engine, nslcmop2_test, timeout_deploy)
733
734 r = engine.test("DEPLOY{}".format(self.step), "Delete NS", "DELETE",
735 "/nslcm/v1/ns_instances/<{}>".format(self.ns_test), headers_yaml, None,
736 204, None, 0)
737 self.step += 1
738 else:
739 r = engine.test("DEPLOY{}".format(self.step), "Delete NS with FORCE", "DELETE",
740 "/nslcm/v1/ns_instances/<{}>?FORCE=True".format(self.ns_test), headers_yaml, None,
741 204, None, 0)
742 self.step += 1
743
744 # check all it is deleted
745 r = engine.test("DEPLOY{}".format(self.step), "Check NS is deleted", "GET",
746 "/nslcm/v1/ns_instances/<{}>".format(self.ns_test), headers_yaml, None,
747 404, None, "yaml")
748 self.step += 1
749 r = engine.test("DEPLOY{}".format(self.step), "Check NSLCMOPs are deleted", "GET",
750 "/nslcm/v1/ns_lcm_op_occs?nsInstanceId=<{}>".format(self.ns_test), headers_json, None,
751 200, None, "json")
752 nslcmops = r.json()
753 if not isinstance(nslcmops, list) or nslcmops:
754 raise TestException("NS {} deleted but with ns_lcm_op_occ active: {}".format(self.ns_test, nslcmops))
755
756 def test_ns(self, engine, test_osm):
757 pass
758
759 def aditional_operations(self, engine, test_osm, manual_check):
760 pass
761
762 def run(self, engine, test_osm, manual_check, test_params=None):
763 engine.get_autorization()
764 nsname = os.environ.get("OSMNBITEST_NS_NAME", "OSMNBITEST")
765 if test_params:
766 if "vnfd-files" in test_params:
767 self.vnfd_filenames = test_params["vnfd-files"].split(",")
768 if "nsd-file" in test_params:
769 self.nsd_filename = test_params["nsd-file"]
770 if test_params.get("ns-name"):
771 nsname = test_params["ns-name"]
772 self.create_descriptors(engine)
773
774 # create real VIM if not exist
775 self.vim_id = engine.get_create_vim(test_osm)
776 ns_data = {"nsDescription": "default description", "nsName": nsname, "nsdId": self.nsd_id,
777 "vimAccountId": self.vim_id}
778 if test_params and test_params.get("ns-config"):
779 if isinstance(test_params["ns-config"], str):
780 ns_data.update(yaml.load(test_params["ns-config"]))
781 else:
782 ns_data.update(test_params["ns-config"])
783 self.instantiate(engine, ns_data)
784
785 if manual_check:
786 input('NS has been deployed. Perform manual check and press enter to resume')
787 else:
788 self.test_ns(engine, test_osm)
789 self.aditional_operations(engine, test_osm, manual_check)
790 self.terminate(engine)
791 self.delete_descriptors(engine)
792
793
794 class TestDeployHackfestCirros(TestDeploy):
795 description = "Load and deploy Hackfest cirros_2vnf_ns example"
796
797 def __init__(self):
798 super().__init__()
799 self.vnfd_filenames = ("cirros_vnf.tar.gz",)
800 self.nsd_filename = "cirros_2vnf_ns.tar.gz"
801
802 def run(self, engine, test_osm, manual_check, test_params=None):
803 super().run(engine, test_osm, manual_check, test_params)
804
805
806 class TestDeployIpMac(TestDeploy):
807 description = "Load and deploy descriptor examples setting mac, ip address at descriptor and instantiate params"
808
809 def __init__(self):
810 super().__init__()
811 self.vnfd_filenames = ("vnfd_2vdu_set_ip_mac2.yaml", "vnfd_2vdu_set_ip_mac.yaml")
812 self.nsd_filename = "scenario_2vdu_set_ip_mac.yaml"
813 self.descriptor_url = \
814 "https://osm.etsi.org/gitweb/?p=osm/RO.git;a=blob_plain;f=test/RO_tests/v3_2vdu_set_ip_mac/"
815
816 def run(self, engine, test_osm, manual_check, test_params=None):
817 # super().run(engine, test_osm, manual_check, test_params)
818 # run again setting IPs with instantiate parameters
819 instantiation_params = {
820 "vnf": [
821 {
822 "member-vnf-index": "1",
823 "internal-vld": [
824 {
825 "name": "internal_vld1", # net_internal
826 "ip-profile": {
827 "ip-version": "ipv4",
828 "subnet-address": "10.9.8.0/24",
829 "dhcp-params": {"count": 100, "start-address": "10.9.8.100"}
830 },
831 "internal-connection-point": [
832 {
833 "id-ref": "eth2",
834 "ip-address": "10.9.8.2",
835 },
836 {
837 "id-ref": "eth3",
838 "ip-address": "10.9.8.3",
839 }
840 ]
841 },
842 ],
843
844 "vdu": [
845 {
846 "id": "VM1",
847 "interface": [
848 # {
849 # "name": "iface11",
850 # "floating-ip-required": True,
851 # },
852 {
853 "name": "iface13",
854 "mac-address": "52:33:44:55:66:13"
855 },
856 ],
857 },
858 {
859 "id": "VM2",
860 "interface": [
861 {
862 "name": "iface21",
863 "ip-address": "10.31.31.21",
864 "mac-address": "52:33:44:55:66:21"
865 },
866 ],
867 },
868 ]
869 },
870 ]
871 }
872 super().run(engine, test_osm, manual_check, test_params={"ns-config": instantiation_params})
873
874
875 class TestDeployHackfest4(TestDeploy):
876 description = "Load and deploy Hackfest 4 example."
877
878 def __init__(self):
879 super().__init__()
880 self.vnfd_filenames = ("hackfest_4_vnfd.tar.gz",)
881 self.nsd_filename = "hackfest_4_nsd.tar.gz"
882 self.uses_configuration = True
883
884 def create_descriptors(self, engine):
885 super().create_descriptors(engine)
886 # Modify VNFD to add scaling
887 payload = """
888 scaling-group-descriptor:
889 - name: "scale_dataVM"
890 max-instance-count: 10
891 scaling-policy:
892 - name: "auto_cpu_util_above_threshold"
893 scaling-type: "automatic"
894 threshold-time: 0
895 cooldown-time: 60
896 scaling-criteria:
897 - name: "cpu_util_above_threshold"
898 scale-in-threshold: 15
899 scale-in-relational-operation: "LE"
900 scale-out-threshold: 60
901 scale-out-relational-operation: "GE"
902 vnf-monitoring-param-ref: "all_aaa_cpu_util"
903 vdu:
904 - vdu-id-ref: dataVM
905 count: 1
906 scaling-config-action:
907 - trigger: post-scale-out
908 vnf-config-primitive-name-ref: touch
909 - trigger: pre-scale-in
910 vnf-config-primitive-name-ref: touch
911 vnf-configuration:
912 config-primitive:
913 - name: touch
914 parameter:
915 - name: filename
916 data-type: STRING
917 default-value: '/home/ubuntu/touched'
918 """
919 engine.test("DEPLOY{}".format(self.step), "Edit VNFD ", "PATCH",
920 "/vnfpkgm/v1/vnf_packages/<{}>".format(self.vnfds_test[0]), headers_yaml, payload, 204, None, None)
921 self.step += 1
922
923 def run(self, engine, test_osm, manual_check, test_params=None):
924 super().run(engine, test_osm, manual_check, test_params)
925
926
927 class TestDeployHackfest3Charmed(TestDeploy):
928 description = "Load and deploy Hackfest 3charmed_ns example. Modifies it for adding scaling and performs " \
929 "primitive actions and scaling"
930
931 def __init__(self):
932 super().__init__()
933 self.vnfd_filenames = ("hackfest_3charmed_vnfd.tar.gz",)
934 self.nsd_filename = "hackfest_3charmed_nsd.tar.gz"
935 self.uses_configuration = True
936
937 # def create_descriptors(self, engine):
938 # super().create_descriptors(engine)
939 # # Modify VNFD to add scaling
940 # payload = """
941 # scaling-group-descriptor:
942 # - name: "scale_dataVM"
943 # max-instance-count: 10
944 # scaling-policy:
945 # - name: "auto_cpu_util_above_threshold"
946 # scaling-type: "automatic"
947 # threshold-time: 0
948 # cooldown-time: 60
949 # scaling-criteria:
950 # - name: "cpu_util_above_threshold"
951 # scale-in-threshold: 15
952 # scale-in-relational-operation: "LE"
953 # scale-out-threshold: 60
954 # scale-out-relational-operation: "GE"
955 # vnf-monitoring-param-ref: "all_aaa_cpu_util"
956 # vdu:
957 # - vdu-id-ref: dataVM
958 # count: 1
959 # scaling-config-action:
960 # - trigger: post-scale-out
961 # vnf-config-primitive-name-ref: touch
962 # - trigger: pre-scale-in
963 # vnf-config-primitive-name-ref: touch
964 # vnf-configuration:
965 # config-primitive:
966 # - name: touch
967 # parameter:
968 # - name: filename
969 # data-type: STRING
970 # default-value: '/home/ubuntu/touched'
971 # """
972 # engine.test("DEPLOY{}".format(self.step), "Edit VNFD ", "PATCH",
973 # "/vnfpkgm/v1/vnf_packages/<{}>".format(self.vnfds_test[0]),
974 # headers_yaml, payload, 200,
975 # r_header_yaml, yaml)
976 # self.vnfds_test.append("DEPLOY" + str(self.step))
977 # self.step += 1
978
979 def aditional_operations(self, engine, test_osm, manual_check):
980 if not test_osm:
981 return
982 # 1 perform action
983 payload = '{member_vnf_index: "2", primitive: touch, primitive_params: { filename: /home/ubuntu/OSMTESTNBI }}'
984 engine.test("DEPLOY{}".format(self.step), "Executer service primitive over NS", "POST",
985 "/nslcm/v1/ns_instances/<{}>/action".format(self.ns_test), headers_yaml, payload,
986 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
987 nslcmop2_action = "DEPLOY{}".format(self.step)
988 self.step += 1
989 # Wait until status is Ok
990 self._wait_nslcmop_ready(engine, nslcmop2_action, timeout_deploy)
991 if manual_check:
992 input('NS service primitive has been executed. Check that file /home/ubuntu/OSMTESTNBI is present at '
993 'TODO_PUT_IP')
994 # TODO check automatic
995
996 # # 2 perform scale out
997 # payload = '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_OUT, scaleByStepData: ' \
998 # '{scaling-group-descriptor: scale_dataVM, member-vnf-index: "1"}}}'
999 # engine.test("DEPLOY{}".format(self.step), "Execute scale action over NS", "POST",
1000 # "/nslcm/v1/ns_instances/<{}>/scale".format(self.ns_test), headers_yaml, payload,
1001 # 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
1002 # nslcmop2_scale_out = "DEPLOY{}".format(self.step)
1003 # self._wait_nslcmop_ready(engine, nslcmop2_scale_out, timeout_deploy)
1004 # if manual_check:
1005 # input('NS scale out done. Check that file /home/ubuntu/touched is present and new VM is created')
1006 # # TODO check automatic
1007 #
1008 # # 2 perform scale in
1009 # payload = '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_IN, scaleByStepData: ' \
1010 # '{scaling-group-descriptor: scale_dataVM, member-vnf-index: "1"}}}'
1011 # engine.test("DEPLOY{}".format(self.step), "Execute scale action over NS", "POST",
1012 # "/nslcm/v1/ns_instances/<{}>/scale".format(self.ns_test), headers_yaml, payload,
1013 # 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
1014 # nslcmop2_scale_in = "DEPLOY{}".format(self.step)
1015 # self._wait_nslcmop_ready(engine, nslcmop2_scale_in, timeout_deploy)
1016 # if manual_check:
1017 # input('NS scale in done. Check that file /home/ubuntu/touched is updated and new VM is deleted')
1018 # # TODO check automatic
1019
1020 def run(self, engine, test_osm, manual_check, test_params=None):
1021 super().run(engine, test_osm, manual_check, test_params)
1022
1023
1024 if __name__ == "__main__":
1025 global logger
1026 test = ""
1027
1028 # Disable warnings from self-signed certificates.
1029 requests.packages.urllib3.disable_warnings()
1030 try:
1031 logging.basicConfig(format="%(levelname)s %(message)s", level=logging.ERROR)
1032 logger = logging.getLogger('NBI')
1033 # load parameters and configuration
1034 opts, args = getopt.getopt(sys.argv[1:], "hvu:p:",
1035 ["url=", "user=", "password=", "help", "version", "verbose", "no-verbose",
1036 "project=", "insecure", "timeout", "timeout-deploy", "timeout-configure",
1037 "test=", "list", "test-osm", "manual-check", "params="])
1038 url = "https://localhost:9999/osm"
1039 user = password = project = "admin"
1040 test_osm = False
1041 manual_check = False
1042 verbose = 0
1043 verify = True
1044 test_classes = {
1045 "NonAuthorized": TestNonAuthorized,
1046 "FakeVIM": TestFakeVim,
1047 "TestUsersProjects": TestUsersProjects,
1048 "VIM-SDN": TestVIMSDN,
1049 "Deploy-Custom": TestDeploy,
1050 "Deploy-Hackfest-Cirros": TestDeployHackfestCirros,
1051 "Deploy-Hackfest-3Charmed": TestDeployHackfest3Charmed,
1052 "Deploy-Hackfest-4": TestDeployHackfest4,
1053 "Deploy-CirrosMacIp": TestDeployIpMac,
1054 }
1055 test_to_do = []
1056 test_params = {}
1057
1058 for o, a in opts:
1059 # print("parameter:", o, a)
1060 if o == "--version":
1061 print("test version " + __version__ + ' ' + version_date)
1062 exit()
1063 elif o == "--list":
1064 for test, test_class in test_classes.items():
1065 print("{:20} {}".format(test + ":", test_class.description))
1066 exit()
1067 elif o in ("-v", "--verbose"):
1068 verbose += 1
1069 elif o == "no-verbose":
1070 verbose = -1
1071 elif o in ("-h", "--help"):
1072 usage()
1073 sys.exit()
1074 elif o == "--test-osm":
1075 test_osm = True
1076 elif o == "--manual-check":
1077 manual_check = True
1078 elif o == "--url":
1079 url = a
1080 elif o in ("-u", "--user"):
1081 user = a
1082 elif o in ("-p", "--password"):
1083 password = a
1084 elif o == "--project":
1085 project = a
1086 elif o == "--test":
1087 # print("asdfadf", o, a, a.split(","))
1088 for _test in a.split(","):
1089 if _test not in test_classes:
1090 print("Invalid test name '{}'. Use option '--list' to show available tests".format(_test),
1091 file=sys.stderr)
1092 exit(1)
1093 test_to_do.append(_test)
1094 elif o == "--params":
1095 param_key, _, param_value = a.partition("=")
1096 text_index = len(test_to_do)
1097 if text_index not in test_params:
1098 test_params[text_index] = {}
1099 test_params[text_index][param_key] = param_value
1100 elif o == "--insecure":
1101 verify = False
1102 elif o == "--timeout":
1103 timeout = int(a)
1104 elif o == "--timeout-deploy":
1105 timeout_deploy = int(a)
1106 elif o == "--timeout-configure":
1107 timeout_configure = int(a)
1108 else:
1109 assert False, "Unhandled option"
1110 if verbose == 0:
1111 logger.setLevel(logging.WARNING)
1112 elif verbose > 1:
1113 logger.setLevel(logging.DEBUG)
1114 else:
1115 logger.setLevel(logging.ERROR)
1116
1117 test_rest = TestRest(url, user=user, password=password, project=project)
1118 # print("tests to do:", test_to_do)
1119 if test_to_do:
1120 text_index = 0
1121 for test in test_to_do:
1122 text_index += 1
1123 test_class = test_classes[test]
1124 test_class().run(test_rest, test_osm, manual_check, test_params.get(text_index))
1125 else:
1126 for test, test_class in test_classes.items():
1127 test_class().run(test_rest, test_osm, manual_check, test_params.get(0))
1128 exit(0)
1129
1130 # get token
1131
1132 # # tests once authorized
1133 # for t in test_authorized_list:
1134 # test_rest.test(*t)
1135 #
1136 # # tests admin
1137 # for t in test_admin_list1:
1138 # test_rest.test(*t)
1139 #
1140 # # vnfd CREATE
1141 # r = test_rest.test("VNFD1", "Onboard VNFD step 1", "POST", "/vnfpkgm/v1/vnf_packages", headers_json, None,
1142 # 201, {"Location": "/vnfpkgm/v1/vnf_packages/", "Content-Type": "application/json"}, "json")
1143 # location = r.headers["Location"]
1144 # vnfd_id = location[location.rfind("/")+1:]
1145 # # print(location, vnfd_id)
1146 #
1147 # # vnfd UPLOAD test
1148 # r = test_rest.test("VNFD2", "Onboard VNFD step 2 as TEXT", "PUT",
1149 # "/vnfpkgm/v1/vnf_packages/{}/package_content".format(vnfd_id),
1150 # r_header_text, "@./cirros_vnf/cirros_vnfd.yaml", 204, None, 0)
1151 #
1152 # # vnfd SHOW OSM format
1153 # r = test_rest.test("VNFD3", "Show VNFD OSM format", "GET",
1154 # "/vnfpkgm/v1/vnf_packages_content/{}".format(vnfd_id),
1155 # headers_json, None, 200, r_header_json, "json")
1156 #
1157 # # vnfd SHOW text
1158 # r = test_rest.test("VNFD4", "Show VNFD SOL005 text", "GET",
1159 # "/vnfpkgm/v1/vnf_packages/{}/package_content".format(vnfd_id),
1160 # headers_text, None, 200, r_header_text, "text")
1161 #
1162 # # vnfd UPLOAD ZIP
1163 # makedirs("temp", exist_ok=True)
1164 # tar = tarfile.open("temp/cirros_vnf.tar.gz", "w:gz")
1165 # tar.add("cirros_vnf")
1166 # tar.close()
1167 # r = test_rest.test("VNFD5", "Onboard VNFD step 3 replace with ZIP", "PUT",
1168 # "/vnfpkgm/v1/vnf_packages/{}/package_content".format(vnfd_id),
1169 # r_header_zip, "@b./temp/cirros_vnf.tar.gz", 204, None, 0)
1170 #
1171 # # vnfd SHOW OSM format
1172 # r = test_rest.test("VNFD6", "Show VNFD OSM format", "GET",
1173 # "/vnfpkgm/v1/vnf_packages_content/{}".format(vnfd_id),
1174 # headers_json, None, 200, r_header_json, "json")
1175 #
1176 # # vnfd SHOW zip
1177 # r = test_rest.test("VNFD7", "Show VNFD SOL005 zip", "GET",
1178 # "/vnfpkgm/v1/vnf_packages/{}/package_content".format(vnfd_id),
1179 # headers_zip, None, 200, r_header_zip, "zip")
1180 # # vnfd SHOW descriptor
1181 # r = test_rest.test("VNFD8", "Show VNFD descriptor", "GET",
1182 # "/vnfpkgm/v1/vnf_packages/{}/vnfd".format(vnfd_id),
1183 # headers_text, None, 200, r_header_text, "text")
1184 # # vnfd SHOW actifact
1185 # r = test_rest.test("VNFD9", "Show VNFD artifact", "GET",
1186 # "/vnfpkgm/v1/vnf_packages/{}/artifacts/icons/cirros-64.png".format(vnfd_id),
1187 # headers_text, None, 200, r_header_octect, "text")
1188 #
1189 # # # vnfd DELETE
1190 # # r = test_rest.test("VNFD10", "Delete VNFD SOL005 text", "DELETE",
1191 # # "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_id),
1192 # # headers_yaml, None, 204, None, 0)
1193 #
1194 # # nsd CREATE
1195 # r = test_rest.test("NSD1", "Onboard NSD step 1", "POST", "/nsd/v1/ns_descriptors", headers_json, None,
1196 # 201, {"Location": "/nsd/v1/ns_descriptors/", "Content-Type": "application/json"}, "json")
1197 # location = r.headers["Location"]
1198 # nsd_id = location[location.rfind("/")+1:]
1199 # # print(location, nsd_id)
1200 #
1201 # # nsd UPLOAD test
1202 # r = test_rest.test("NSD2", "Onboard NSD with missing vnfd", "PUT",
1203 # "/nsd/v1/ns_descriptors/<>/nsd_content?constituent-vnfd.0.vnfd-id-ref"
1204 # "=NONEXISTING-VNFD".format(nsd_id),
1205 # r_header_text, "@./cirros_ns/cirros_nsd.yaml", 409, r_header_yaml, "yaml")
1206 #
1207 # # # VNF_CREATE
1208 # # r = test_rest.test("VNFD5", "Onboard VNFD step 3 replace with ZIP", "PUT",
1209 # # "/vnfpkgm/v1/vnf_packages/{}/package_content".format(vnfd_id),
1210 # # r_header_zip, "@b./temp/cirros_vnf.tar.gz", 204, None, 0)
1211 #
1212 # r = test_rest.test("NSD2", "Onboard NSD step 2 as TEXT", "PUT",
1213 # "/nsd/v1/ns_descriptors/{}/nsd_content".format(nsd_id),
1214 # r_header_text, "@./cirros_ns/cirros_nsd.yaml", 204, None, 0)
1215 #
1216 # # nsd SHOW OSM format
1217 # r = test_rest.test("NSD3", "Show NSD OSM format", "GET", "/nsd/v1/ns_descriptors_content/{}".format(nsd_id),
1218 # headers_json, None, 200, r_header_json, "json")
1219 #
1220 # # nsd SHOW text
1221 # r = test_rest.test("NSD4", "Show NSD SOL005 text", "GET",
1222 # "/nsd/v1/ns_descriptors/{}/nsd_content".format(nsd_id),
1223 # headers_text, None, 200, r_header_text, "text")
1224 #
1225 # # nsd UPLOAD ZIP
1226 # makedirs("temp", exist_ok=True)
1227 # tar = tarfile.open("temp/cirros_ns.tar.gz", "w:gz")
1228 # tar.add("cirros_ns")
1229 # tar.close()
1230 # r = test_rest.test("NSD5", "Onboard NSD step 3 replace with ZIP", "PUT",
1231 # "/nsd/v1/ns_descriptors/{}/nsd_content".format(nsd_id),
1232 # r_header_zip, "@b./temp/cirros_ns.tar.gz", 204, None, 0)
1233 #
1234 # # nsd SHOW OSM format
1235 # r = test_rest.test("NSD6", "Show NSD OSM format", "GET", "/nsd/v1/ns_descriptors_content/{}".format(nsd_id),
1236 # headers_json, None, 200, r_header_json, "json")
1237 #
1238 # # nsd SHOW zip
1239 # r = test_rest.test("NSD7","Show NSD SOL005 zip","GET", "/nsd/v1/ns_descriptors/{}/nsd_content".format(nsd_id),
1240 # headers_zip, None, 200, r_header_zip, "zip")
1241 #
1242 # # nsd SHOW descriptor
1243 # r = test_rest.test("NSD8", "Show NSD descriptor", "GET", "/nsd/v1/ns_descriptors/{}/nsd".format(nsd_id),
1244 # headers_text, None, 200, r_header_text, "text")
1245 # # nsd SHOW actifact
1246 # r = test_rest.test("NSD9", "Show NSD artifact", "GET",
1247 # "/nsd/v1/ns_descriptors/{}/artifacts/icons/osm_2x.png".format(nsd_id),
1248 # headers_text, None, 200, r_header_octect, "text")
1249 #
1250 # # vnfd DELETE
1251 # r = test_rest.test("VNFD10", "Delete VNFD conflict", "DELETE", "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_id),
1252 # headers_yaml, None, 409, r_header_yaml, "yaml")
1253 #
1254 # # nsd DELETE
1255 # r = test_rest.test("NSD10", "Delete NSD SOL005 text", "DELETE", "/nsd/v1/ns_descriptors/{}".format(nsd_id),
1256 # headers_yaml, None, 204, None, 0)
1257 #
1258 # # vnfd DELETE
1259 # r = test_rest.test("VNFD10","Delete VNFD SOL005 text","DELETE", "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_id),
1260 # headers_yaml, None, 204, None, 0)
1261 print("PASS")
1262
1263 except TestException as e:
1264 logger.error(test + "Test {} Exception: {}".format(test, str(e)))
1265 exit(1)
1266 except getopt.GetoptError as e:
1267 logger.error(e)
1268 print(e, file=sys.stderr)
1269 exit(1)
1270 except Exception as e:
1271 logger.critical(test + " Exception: " + str(e), exc_info=True)