2 # -*- coding: utf-8 -*-
12 from time
import sleep
15 __author__
= "Alfonso Tierno, alfonso.tiernosepulveda@telefonica.com"
16 __date__
= "$2018-03-01$"
18 version_date
= "Jul 2018"
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")
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 "
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\"")
59 r_header_json
= {"Content-type": "application/json"}
61 "Content-type": "application/json",
62 "Accept": "application/json",
64 r_header_yaml
= {"Content-type": "application/yaml"}
66 "Content-type": "application/yaml",
67 "Accept": "application/yaml",
69 r_header_text
= {"Content-type": "text/plain"}
70 r_header_octect
= {"Content-type": "application/octet-stream"}
72 "Accept": "text/plain",
74 r_header_zip
= {"Content-type": "application/zip"}
76 "Accept": "application/zip",
79 "Accept": "application/yaml", "Content-type": "application/zip"
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"),
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
97 class TestException(Exception):
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
= {}
107 self
.header_base
= header_base
.copy()
108 self
.s
= requests
.session()
109 self
.s
.headers
= self
.header_base
113 self
.password
= password
114 self
.project
= project
116 # contains ID of tests obtained from Location response header. "" key contains last obtained id
118 self
.old_test_description
= ""
120 def set_header(self
, header
):
121 self
.s
.headers
.update(header
)
123 def unset_header(self
, key
):
124 if key
in self
.s
.headers
:
125 del self
.s
.headers
[key
]
127 def test(self
, name
, description
, method
, url
, headers
, payload
, expected_codes
, expected_headers
,
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
146 self
.s
= requests
.session()
150 elif not url
.startswith("http"):
151 url
= self
.url_base
+ url
153 var_start
= url
.find("<") + 1
155 var_end
= url
.find(">", var_start
)
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
164 if isinstance(payload
, str):
165 if payload
.startswith("@"):
167 file_name
= payload
[1:]
168 if payload
.startswith("@b"):
170 file_name
= payload
[2:]
171 with
open(file_name
, mode
) as f
:
173 elif isinstance(payload
, dict):
174 payload
= json
.dumps(payload
)
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
)
181 # if expected_payload == "zip":
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
))
188 if isinstance(expected_codes
, int):
189 expected_codes
= (expected_codes
,)
190 if r
.status_code
not in expected_codes
:
192 "Got status {}. Expected {}. {}".format(r
.status_code
, expected_codes
, r
.text
))
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
]))
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":
208 except Exception as e
:
209 raise TestException("Expected json response payload, but got Exception {}".format(e
))
210 elif expected_payload
== "yaml":
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")
219 # tar = tarfile.open(None, 'r:gz', fileobj=r.raw)
220 # for tarinfo in tar:
221 # tarname = tarinfo.name
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")
229 location
= r
.headers
.get("Location")
231 _id
= location
[location
.rfind("/") + 1:]
233 self
.test_ids
[name
] = str(_id
)
234 self
.test_ids
[""] = str(_id
) # last id
236 except TestException
as e
:
240 r_status_code
= r
.status_code
242 logger
.error("{} \nRX code{}: {}".format(e
, r_status_code
, r_text
))
245 logger
.error("Cannot open file {}".format(e
))
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:
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")
258 self
.token
= response
["id"]
259 self
.set_header({"Authorization": "Bearer {}".format(self
.token
)})
261 def remove_authorization(self
):
263 self
.test("TOKEN_DEL", "Delete token", "DELETE", "/admin/v1/tokens/{}".format(self
.token
), headers_json
,
264 None, (200, 201, 204), None, None)
266 self
.unset_header("Authorization")
268 def get_create_vim(self
, test_osm
):
271 self
.get_autorization()
273 vim_name
= os
.environ
.get("OSMNBITEST_VIM_NAME")
276 "Needed to define OSMNBITEST_VIM_XXX variables to create a real VIM for deployment")
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")
284 return vims
[0]["_id"]
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"))
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:]
310 class TestNonAuthorized
:
311 description
= "test invalid URLs. methods and no authorization"
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"),
321 for t
in test_not_authorized_list
:
325 class TestUsersProjects
:
326 description
= "test project and user creation"
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")
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
)
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
))
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
)
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
))
370 engine
.test("PU9", "Edit user U1, change password", "PATCH", "/admin/v1/users/U1", headers_json
,
371 {"password": "pw1_new"}, 204, None, None)
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")
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"])})
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")
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"])})
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)
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)
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)
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,
422 description
= "Creates/edit/delete fake VIMs and SDN controllers"
426 "schema_version": "1.0",
427 "schema_type": "No idea",
429 "description": "Descriptor name",
430 "vim_type": "openstack",
431 "vim_url": "http://localhost:/vim",
432 "vim_tenant_name": "vimTenant",
434 "vim_password": "password",
435 "config": {"config_param": 1}
439 "description": "sdn-description",
440 "dpid": "50:50:52:54:00:94:21:21",
441 "ip": "192.168.15.17",
443 "type": "opendaylight",
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"}
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"}
459 def run(self
, engine
, test_osm
, manual_check
, test_params
=None):
461 vim_bad
= self
.vim
.copy()
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
,
473 engine
.test("FVIM5", "Show VIM", "GET", "/admin/v1/vim_accounts/<FVIM1>", headers_yaml
, None, 200,
474 r_header_yaml
, "yaml")
477 engine
.test("FVIM6", "Delete VIM", "DELETE", "/admin/v1/vim_accounts/<FVIM1>?FORCE=True", headers_yaml
,
479 engine
.test("FVIM7", "Check VIM is deleted", "GET", "/admin/v1/vim_accounts/<FVIM1>", headers_yaml
, None,
480 404, r_header_yaml
, "yaml")
482 # delete and wait until is really deleted
483 engine
.test("FVIM6", "Delete VIM", "DELETE", "/admin/v1/vim_accounts/<FVIM1>", headers_yaml
, None, 202,
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:
491 elif r
.status_code
== 200:
495 raise TestException("Vim created at 'FVIM1' is not delete after {} seconds".format(timeout
))
498 class TestVIMSDN(TestFakeVim
):
499 description
= "Creates VIM with SDN editing SDN controllers and port_mapping"
502 TestFakeVim
.__init
__(self
)
504 def run(self
, engine
, test_osm
, manual_check
, test_params
=None):
505 engine
.get_autorization()
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")
511 engine
.test("VIMSDN2", "Edit SDN", "PATCH", "/admin/v1/sdns/<VIMSDN1>", headers_json
, {"name": "new_sdn_name"},
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"),
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)
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,
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")
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)
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:
549 elif r
.status_code
== 200:
553 raise TestException("Vim created at 'VIMSDN3' is not delete after {} seconds".format(timeout
))
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:
559 elif r
.status_code
== 200:
563 raise TestException("SDNC created at 'VIMSDN1' is not delete after {} seconds".format(timeout
))
567 description
= "Base class for downloading descriptors from ETSI, onboard and deploy in real VIM"
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
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
))
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
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
))
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
))
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)
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
))
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
640 headers
= headers_zip_yaml
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
)
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")
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)
661 self
.nsd_id
= engine
.test_ids
[self
.nsd_test
]
663 def delete_descriptors(self
, engine
):
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)
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)
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
]
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
)
690 # Wait until status is Ok
691 wait
= timeout_configure
if self
.uses_configuration
else timeout_deploy
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")
697 if "COMPLETED" in nslcmop
["operationState"]:
699 elif "FAILED" in nslcmop
["operationState"]:
700 raise TestException("NS instantiate has failed: {}".format(nslcmop
["detailed-status"]))
704 raise TestException("NS instantiate is not done after {} seconds".format(timeout_deploy
))
707 def _wait_nslcmop_ready(self
, engine
, nslcmop_test
, timeout_deploy
):
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")
714 if "COMPLETED" in nslcmop
["operationState"]:
716 elif "FAILED" in nslcmop
["operationState"]:
717 raise TestException("NS terminate has failed: {}".format(nslcmop
["detailed-status"]))
721 raise TestException("NS instantiate is not terminate after {} seconds".format(timeout
))
723 def terminate(self
, engine
):
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
)
731 # Wait until status is Ok
732 self
._wait
_nslcmop
_ready
(engine
, nslcmop2_test
, timeout_deploy
)
734 r
= engine
.test("DEPLOY{}".format(self
.step
), "Delete NS", "DELETE",
735 "/nslcm/v1/ns_instances/<{}>".format(self
.ns_test
), headers_yaml
, None,
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,
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,
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,
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
))
756 def test_ns(self
, engine
, test_osm
):
759 def aditional_operations(self
, engine
, test_osm
, manual_check
):
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")
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
)
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"]))
782 ns_data
.update(test_params
["ns-config"])
783 self
.instantiate(engine
, ns_data
)
786 input('NS has been deployed. Perform manual check and press enter to resume')
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
)
794 class TestDeployHackfestCirros(TestDeploy
):
795 description
= "Load and deploy Hackfest cirros_2vnf_ns example"
799 self
.vnfd_filenames
= ("cirros_vnf.tar.gz",)
800 self
.nsd_filename
= "cirros_2vnf_ns.tar.gz"
802 def run(self
, engine
, test_osm
, manual_check
, test_params
=None):
803 super().run(engine
, test_osm
, manual_check
, test_params
)
806 class TestDeployIpMac(TestDeploy
):
807 description
= "Load and deploy descriptor examples setting mac, ip address at descriptor and instantiate params"
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/"
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
= {
822 "member-vnf-index": "1",
825 "name": "internal_vld1", # net_internal
827 "ip-version": "ipv4",
828 "subnet-address": "10.9.8.0/24",
829 "dhcp-params": {"count": 100, "start-address": "10.9.8.100"}
831 "internal-connection-point": [
834 "ip-address": "10.9.8.2",
838 "ip-address": "10.9.8.3",
850 # "floating-ip-required": True,
854 "mac-address": "52:33:44:55:66:13"
863 "ip-address": "10.31.31.21",
864 "mac-address": "52:33:44:55:66:21"
872 super().run(engine
, test_osm
, manual_check
, test_params
={"ns-config": instantiation_params
})
875 class TestDeployHackfest4(TestDeploy
):
876 description
= "Load and deploy Hackfest 4 example."
880 self
.vnfd_filenames
= ("hackfest_4_vnfd.tar.gz",)
881 self
.nsd_filename
= "hackfest_4_nsd.tar.gz"
882 self
.uses_configuration
= True
884 def create_descriptors(self
, engine
):
885 super().create_descriptors(engine
)
886 # Modify VNFD to add scaling
888 scaling-group-descriptor:
889 - name: "scale_dataVM"
890 max-instance-count: 10
892 - name: "auto_cpu_util_above_threshold"
893 scaling-type: "automatic"
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"
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
917 default-value: '/home/ubuntu/touched'
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)
923 def run(self
, engine
, test_osm
, manual_check
, test_params
=None):
924 super().run(engine
, test_osm
, manual_check
, test_params
)
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"
933 self
.vnfd_filenames
= ("hackfest_3charmed_vnfd.tar.gz",)
934 self
.nsd_filename
= "hackfest_3charmed_nsd.tar.gz"
935 self
.uses_configuration
= True
937 # def create_descriptors(self, engine):
938 # super().create_descriptors(engine)
939 # # Modify VNFD to add scaling
941 # scaling-group-descriptor:
942 # - name: "scale_dataVM"
943 # max-instance-count: 10
945 # - name: "auto_cpu_util_above_threshold"
946 # scaling-type: "automatic"
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"
957 # - vdu-id-ref: dataVM
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
970 # default-value: '/home/ubuntu/touched'
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))
979 def aditional_operations(self
, engine
, test_osm
, manual_check
):
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
)
989 # Wait until status is Ok
990 self
._wait
_nslcmop
_ready
(engine
, nslcmop2_action
, timeout_deploy
)
992 input('NS service primitive has been executed. Check that file /home/ubuntu/OSMTESTNBI is present at '
994 # TODO check automatic
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)
1005 # input('NS scale out done. Check that file /home/ubuntu/touched is present and new VM is created')
1006 # # TODO check automatic
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)
1017 # input('NS scale in done. Check that file /home/ubuntu/touched is updated and new VM is deleted')
1018 # # TODO check automatic
1020 def run(self
, engine
, test_osm
, manual_check
, test_params
=None):
1021 super().run(engine
, test_osm
, manual_check
, test_params
)
1024 if __name__
== "__main__":
1028 # Disable warnings from self-signed certificates.
1029 requests
.packages
.urllib3
.disable_warnings()
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"
1041 manual_check
= False
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
,
1059 # print("parameter:", o, a)
1060 if o
== "--version":
1061 print("test version " + __version__
+ ' ' + version_date
)
1064 for test
, test_class
in test_classes
.items():
1065 print("{:20} {}".format(test
+ ":", test_class
.description
))
1067 elif o
in ("-v", "--verbose"):
1069 elif o
== "no-verbose":
1071 elif o
in ("-h", "--help"):
1074 elif o
== "--test-osm":
1076 elif o
== "--manual-check":
1080 elif o
in ("-u", "--user"):
1082 elif o
in ("-p", "--password"):
1084 elif o
== "--project":
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
),
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":
1102 elif o
== "--timeout":
1104 elif o
== "--timeout-deploy":
1105 timeout_deploy
= int(a
)
1106 elif o
== "--timeout-configure":
1107 timeout_configure
= int(a
)
1109 assert False, "Unhandled option"
1111 logger
.setLevel(logging
.WARNING
)
1113 logger
.setLevel(logging
.DEBUG
)
1115 logger
.setLevel(logging
.ERROR
)
1117 test_rest
= TestRest(url
, user
=user
, password
=password
, project
=project
)
1118 # print("tests to do:", test_to_do)
1121 for test
in test_to_do
:
1123 test_class
= test_classes
[test
]
1124 test_class().run(test_rest
, test_osm
, manual_check
, test_params
.get(text_index
))
1126 for test
, test_class
in test_classes
.items():
1127 test_class().run(test_rest
, test_osm
, manual_check
, test_params
.get(0))
1132 # # tests once authorized
1133 # for t in test_authorized_list:
1134 # test_rest.test(*t)
1137 # for t in test_admin_list1:
1138 # test_rest.test(*t)
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)
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)
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")
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")
1163 # makedirs("temp", exist_ok=True)
1164 # tar = tarfile.open("temp/cirros_vnf.tar.gz", "w:gz")
1165 # tar.add("cirros_vnf")
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)
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")
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")
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)
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)
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")
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)
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)
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")
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")
1226 # makedirs("temp", exist_ok=True)
1227 # tar = tarfile.open("temp/cirros_ns.tar.gz", "w:gz")
1228 # tar.add("cirros_ns")
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)
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")
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")
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")
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")
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)
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)
1263 except TestException
as e
:
1264 logger
.error(test
+ "Test {} Exception: {}".format(test
, str(e
)))
1266 except getopt
.GetoptError
as e
:
1268 print(e
, file=sys
.stderr
)
1270 except Exception as e
:
1271 logger
.critical(test
+ " Exception: " + str(e
), exc_info
=True)