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
= "Oct 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"
577 self
.descriptor_url
= "https://osm-download.etsi.org/ftp/osm-3.0-three/2nd-hackfest/packages/"
578 self
.vnfd_filenames
= ("cirros_vnf.tar.gz",)
579 self
.nsd_filename
= "cirros_2vnf_ns.tar.gz"
580 self
.uses_configuration
= False
586 self
.passed_tests
= 0
589 def create_descriptors(self
, engine
):
590 temp_dir
= os
.path
.dirname(os
.path
.abspath(__file__
)) + "/temp/"
591 if not os
.path
.exists(temp_dir
):
592 os
.makedirs(temp_dir
)
593 for vnfd_filename
in self
.vnfd_filenames
:
594 if "/" in vnfd_filename
:
595 vnfd_filename_path
= vnfd_filename
596 if not os
.path
.exists(vnfd_filename_path
):
597 raise TestException("File '{}' does not exist".format(vnfd_filename_path
))
599 vnfd_filename_path
= temp_dir
+ vnfd_filename
600 if not os
.path
.exists(vnfd_filename_path
):
601 with
open(vnfd_filename_path
, "wb") as file:
602 response
= requests
.get(self
.descriptor_url
+ vnfd_filename
)
603 if response
.status_code
>= 300:
604 raise TestException("Error downloading descriptor from '{}': {}".format(
605 self
.descriptor_url
+ vnfd_filename
, response
.status_code
))
606 file.write(response
.content
)
607 if vnfd_filename_path
.endswith(".yaml"):
608 headers
= headers_yaml
610 headers
= headers_zip_yaml
611 if self
.step
% 2 == 0:
612 # vnfd CREATE AND UPLOAD in one step:
613 engine
.test("DEPLOY{}".format(self
.step
), "Onboard VNFD in one step", "POST",
614 "/vnfpkgm/v1/vnf_packages_content", headers
, "@b" + vnfd_filename_path
, 201,
615 {"Location": "/vnfpkgm/v1/vnf_packages_content/", "Content-Type": "application/yaml"}, yaml
)
616 self
.vnfds_test
.append("DEPLOY" + str(self
.step
))
619 # vnfd CREATE AND UPLOAD ZIP
620 engine
.test("DEPLOY{}".format(self
.step
), "Onboard VNFD step 1", "POST", "/vnfpkgm/v1/vnf_packages",
621 headers_json
, None, 201,
622 {"Location": "/vnfpkgm/v1/vnf_packages/", "Content-Type": "application/json"}, "json")
623 self
.vnfds_test
.append("DEPLOY" + str(self
.step
))
625 # location = r.headers["Location"]
626 # vnfd_id = location[location.rfind("/")+1:]
627 engine
.test("DEPLOY{}".format(self
.step
), "Onboard VNFD step 2 as ZIP", "PUT",
628 "/vnfpkgm/v1/vnf_packages/<>/package_content",
629 headers
, "@b" + vnfd_filename_path
, 204, None, 0)
632 if "/" in self
.nsd_filename
:
633 nsd_filename_path
= self
.nsd_filename
634 if not os
.path
.exists(nsd_filename_path
):
635 raise TestException("File '{}' does not exist".format(nsd_filename_path
))
637 nsd_filename_path
= temp_dir
+ self
.nsd_filename
638 if not os
.path
.exists(nsd_filename_path
):
639 with
open(nsd_filename_path
, "wb") as file:
640 response
= requests
.get(self
.descriptor_url
+ self
.nsd_filename
)
641 if response
.status_code
>= 300:
642 raise TestException("Error downloading descriptor from '{}': {}".format(
643 self
.descriptor_url
+ self
.nsd_filename
, response
.status_code
))
644 file.write(response
.content
)
645 if nsd_filename_path
.endswith(".yaml"):
646 headers
= headers_yaml
648 headers
= headers_zip_yaml
650 self
.nsd_test
= "DEPLOY" + str(self
.step
)
651 if self
.step
% 2 == 0:
652 # nsd CREATE AND UPLOAD in one step:
653 engine
.test("DEPLOY{}".format(self
.step
), "Onboard NSD in one step", "POST",
654 "/nsd/v1/ns_descriptors_content", headers
, "@b" + nsd_filename_path
, 201,
655 {"Location": "/nsd/v1/ns_descriptors_content/", "Content-Type": "application/yaml"}, yaml
)
658 # nsd CREATE AND UPLOAD ZIP
659 engine
.test("DEPLOY{}".format(self
.step
), "Onboard NSD step 1", "POST", "/nsd/v1/ns_descriptors",
660 headers_json
, None, 201,
661 {"Location": "/nsd/v1/ns_descriptors/", "Content-Type": "application/json"}, "json")
663 # location = r.headers["Location"]
664 # vnfd_id = location[location.rfind("/")+1:]
665 engine
.test("DEPLOY{}".format(self
.step
), "Onboard NSD step 2 as ZIP", "PUT",
666 "/nsd/v1/ns_descriptors/<>/nsd_content",
667 headers
, "@b" + nsd_filename_path
, 204, None, 0)
669 self
.nsd_id
= engine
.test_ids
[self
.nsd_test
]
671 def delete_descriptors(self
, engine
):
673 engine
.test("DEPLOY{}".format(self
.step
), "Delete NSSD SOL005", "DELETE",
674 "/nsd/v1/ns_descriptors/<{}>".format(self
.nsd_test
),
675 headers_yaml
, None, 204, None, 0)
677 for vnfd_test
in self
.vnfds_test
:
678 engine
.test("DEPLOY{}".format(self
.step
), "Delete VNFD SOL005", "DELETE",
679 "/vnfpkgm/v1/vnf_packages/<{}>".format(vnfd_test
), headers_yaml
, None, 204, None, 0)
682 def instantiate(self
, engine
, ns_data
):
683 ns_data_text
= yaml
.safe_dump(ns_data
, default_flow_style
=True, width
=256)
684 # create NS Two steps
685 r
= engine
.test("DEPLOY{}".format(self
.step
), "Create NS step 1", "POST", "/nslcm/v1/ns_instances",
686 headers_yaml
, ns_data_text
, 201,
687 {"Location": "nslcm/v1/ns_instances/", "Content-Type": "application/yaml"}, "yaml")
688 self
.ns_test
= "DEPLOY{}".format(self
.step
)
689 self
.ns_id
= engine
.test_ids
[self
.ns_test
]
690 engine
.test_ids
[self
.ns_test
]
692 r
= engine
.test("DEPLOY{}".format(self
.step
), "Instantiate NS step 2", "POST",
693 "/nslcm/v1/ns_instances/<{}>/instantiate".format(self
.ns_test
), headers_yaml
, ns_data_text
,
694 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
695 nslcmop_test
= "DEPLOY{}".format(self
.step
)
699 # Wait until status is Ok
700 wait
= timeout_configure
if self
.uses_configuration
else timeout_deploy
702 r
= engine
.test("DEPLOY{}".format(self
.step
), "Wait until NS is deployed and configured", "GET",
703 "/nslcm/v1/ns_lcm_op_occs/<{}>".format(nslcmop_test
), headers_json
, None,
704 200, r_header_json
, "json")
706 if "COMPLETED" in nslcmop
["operationState"]:
708 elif "FAILED" in nslcmop
["operationState"]:
709 raise TestException("NS instantiate has failed: {}".format(nslcmop
["detailed-status"]))
713 raise TestException("NS instantiate is not done after {} seconds".format(timeout_deploy
))
716 def _wait_nslcmop_ready(self
, engine
, nslcmop_test
, timeout_deploy
):
719 r
= engine
.test("DEPLOY{}".format(self
.step
), "Wait to ns lcm operation complete", "GET",
720 "/nslcm/v1/ns_lcm_op_occs/<{}>".format(nslcmop_test
), headers_json
, None,
721 200, r_header_json
, "json")
723 if "COMPLETED" in nslcmop
["operationState"]:
725 elif "FAILED" in nslcmop
["operationState"]:
726 raise TestException("NS terminate has failed: {}".format(nslcmop
["detailed-status"]))
730 raise TestException("NS instantiate is not terminate after {} seconds".format(timeout
))
732 def terminate(self
, engine
):
735 r
= engine
.test("DEPLOY{}".format(self
.step
), "Terminate NS", "POST",
736 "/nslcm/v1/ns_instances/<{}>/terminate".format(self
.ns_test
), headers_yaml
, None,
737 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
738 nslcmop2_test
= "DEPLOY{}".format(self
.step
)
740 # Wait until status is Ok
741 self
._wait
_nslcmop
_ready
(engine
, nslcmop2_test
, timeout_deploy
)
743 r
= engine
.test("DEPLOY{}".format(self
.step
), "Delete NS", "DELETE",
744 "/nslcm/v1/ns_instances/<{}>".format(self
.ns_test
), headers_yaml
, None,
748 r
= engine
.test("DEPLOY{}".format(self
.step
), "Delete NS with FORCE", "DELETE",
749 "/nslcm/v1/ns_instances/<{}>?FORCE=True".format(self
.ns_test
), headers_yaml
, None,
753 # check all it is deleted
754 r
= engine
.test("DEPLOY{}".format(self
.step
), "Check NS is deleted", "GET",
755 "/nslcm/v1/ns_instances/<{}>".format(self
.ns_test
), headers_yaml
, None,
758 r
= engine
.test("DEPLOY{}".format(self
.step
), "Check NSLCMOPs are deleted", "GET",
759 "/nslcm/v1/ns_lcm_op_occs?nsInstanceId=<{}>".format(self
.ns_test
), headers_json
, None,
762 if not isinstance(nslcmops
, list) or nslcmops
:
763 raise TestException("NS {} deleted but with ns_lcm_op_occ active: {}".format(self
.ns_test
, nslcmops
))
765 def test_ns(self
, engine
, test_osm
, commands
=None, users
=None, passwds
=None, keys
=None, timeout
=0):
768 r
= engine
.test("TEST_NS{}".format(n
), "GET VNFR_IDs", "GET",
769 "/nslcm/v1/ns_instances/{}".format(self
.ns_id
), headers_json
, None,
770 200, r_header_json
, "json")
774 vnfr_list
= ns_data
['constituent-vnfr-ref']
777 for vnfr_id
in vnfr_list
:
778 self
.total_tests
+= 1
779 r
= engine
.test("TEST_NS{}".format(n
), "GET IP_ADDRESS OF VNFR", "GET",
780 "/nslcm/v1/vnfrs/{}".format(vnfr_id
), headers_json
, None,
781 200, r_header_json
, "json")
785 if vnfr_data
.get("ip-address"):
786 name
= "TEST_NS{}".format(n
)
787 description
= "Run tests in VNFR with IP {}".format(vnfr_data
['ip-address'])
789 test_description
= "Test {} {}".format(name
, description
)
790 logger
.warning(test_description
)
791 vnf_index
= str(vnfr_data
["member-vnf-index-ref"])
792 while timeout
>= time
:
793 result
, message
= self
.do_checks([vnfr_data
["ip-address"]],
794 vnf_index
=vnfr_data
["member-vnf-index-ref"],
795 commands
=commands
.get(vnf_index
), user
=users
.get(vnf_index
),
796 passwd
=passwds
.get(vnf_index
), key
=keys
.get(vnf_index
))
798 logger
.warning(message
)
804 logger
.critical(message
)
808 logger
.critical(message
)
810 logger
.critical("VNFR {} has not mgmt address. Check failed".format(vnfr_id
))
812 def do_checks(self
, ip
, vnf_index
, commands
=[], user
=None, passwd
=None, key
=None):
815 from pssh
.clients
import ParallelSSHClient
816 from pssh
.utils
import load_private_key
817 from ssh2
import exceptions
as ssh2Exception
818 except ImportError as e
:
819 logger
.critical("package <pssh> or/and <urllib3> is not installed. Please add it with 'pip3 install "
820 "parallel-ssh' and/or 'pip3 install urllib3': {}".format(e
))
821 urllib3
.disable_warnings(urllib3
.exceptions
.InsecureRequestWarning
)
823 p_host
= os
.environ
.get("PROXY_HOST")
824 p_user
= os
.environ
.get("PROXY_USER")
825 p_password
= os
.environ
.get("PROXY_PASSWD")
828 pkey
= load_private_key(key
)
832 client
= ParallelSSHClient(ip
, user
=user
, password
=passwd
, pkey
=pkey
, proxy_host
=p_host
,
833 proxy_user
=p_user
, proxy_password
=p_password
, timeout
=10, num_retries
=0)
835 output
= client
.run_command(cmd
)
837 if output
[ip
[0]].exit_code
:
838 return -1, " VNFR {} could not be checked: {}".format(ip
[0], output
[ip
[0]].stderr
)
840 self
.passed_tests
+= 1
841 return 1, " Test successful"
842 except (ssh2Exception
.ChannelFailure
, ssh2Exception
.SocketDisconnectError
, ssh2Exception
.SocketTimeout
,
843 ssh2Exception
.SocketRecvError
) as e
:
844 return 0, "Timeout accessing the VNFR {}: {}".format(ip
[0], str(e
))
845 except Exception as e
:
846 return -1, "ERROR checking the VNFR {}: {}".format(ip
[0], str(e
))
848 def aditional_operations(self
, engine
, test_osm
, manual_check
):
851 def run(self
, engine
, test_osm
, manual_check
, test_params
=None):
852 engine
.get_autorization()
853 nsname
= os
.environ
.get("OSMNBITEST_NS_NAME", "OSMNBITEST")
855 if "vnfd-files" in test_params
:
856 self
.vnfd_filenames
= test_params
["vnfd-files"].split(",")
857 if "nsd-file" in test_params
:
858 self
.nsd_filename
= test_params
["nsd-file"]
859 if test_params
.get("ns-name"):
860 nsname
= test_params
["ns-name"]
861 self
.create_descriptors(engine
)
863 # create real VIM if not exist
864 self
.vim_id
= engine
.get_create_vim(test_osm
)
865 ns_data
= {"nsDescription": "default description", "nsName": nsname
, "nsdId": self
.nsd_id
,
866 "vimAccountId": self
.vim_id
}
867 if test_params
and test_params
.get("ns-config"):
868 if isinstance(test_params
["ns-config"], str):
869 ns_data
.update(yaml
.load(test_params
["ns-config"]))
871 ns_data
.update(test_params
["ns-config"])
872 self
.instantiate(engine
, ns_data
)
875 input('NS has been deployed. Perform manual check and press enter to resume')
877 self
.test_ns(engine
, test_osm
, self
.cmds
, self
.uss
, self
.pss
, self
.keys
, self
.timeout
)
878 self
.aditional_operations(engine
, test_osm
, manual_check
)
879 self
.terminate(engine
)
880 self
.delete_descriptors(engine
)
883 def print_results(self
):
884 print("\n\n\n--------------------------------------------")
885 print("TEST RESULTS:\n PASSED TESTS: {} - TOTAL TESTS: {}".format(self
.total_tests
, self
.passed_tests
))
886 print("--------------------------------------------")
889 class TestDeployHackfestCirros(TestDeploy
):
890 description
= "Load and deploy Hackfest cirros_2vnf_ns example"
894 self
.vnfd_filenames
= ("cirros_vnf.tar.gz",)
895 self
.nsd_filename
= "cirros_2vnf_ns.tar.gz"
896 self
.cmds
= {'1': ['ls -lrt', ], '2': ['ls -lrt', ]}
897 self
.uss
= {'1': "cirros", '2': "cirros"}
898 self
.pss
= {'1': "cubswin:)", '2': "cubswin:)"}
901 class TestDeployIpMac(TestDeploy
):
902 description
= "Load and deploy descriptor examples setting mac, ip address at descriptor and instantiate params"
906 self
.vnfd_filenames
= ("vnfd_2vdu_set_ip_mac2.yaml", "vnfd_2vdu_set_ip_mac.yaml")
907 self
.nsd_filename
= "scenario_2vdu_set_ip_mac.yaml"
908 self
.descriptor_url
= \
909 "https://osm.etsi.org/gitweb/?p=osm/RO.git;a=blob_plain;f=test/RO_tests/v3_2vdu_set_ip_mac/"
910 self
.cmds
= {'1': ['ls -lrt', ], '2': ['ls -lrt', ]}
911 self
.uss
= {'1': "osm", '2': "osm"}
912 self
.pss
= {'1': "osm4u", '2': "osm4u"}
915 def run(self
, engine
, test_osm
, manual_check
, test_params
=None):
916 # super().run(engine, test_osm, manual_check, test_params)
917 # run again setting IPs with instantiate parameters
918 instantiation_params
= {
921 "member-vnf-index": "1",
924 "name": "internal_vld1", # net_internal
926 "ip-version": "ipv4",
927 "subnet-address": "10.9.8.0/24",
928 "dhcp-params": {"count": 100, "start-address": "10.9.8.100"}
930 "internal-connection-point": [
933 "ip-address": "10.9.8.2",
937 "ip-address": "10.9.8.3",
949 # "floating-ip-required": True,
953 "mac-address": "52:33:44:55:66:13"
962 "ip-address": "10.31.31.22",
963 "mac-address": "52:33:44:55:66:21"
972 super().run(engine
, test_osm
, manual_check
, test_params
={"ns-config": instantiation_params
})
975 class TestDeployHackfest4(TestDeploy
):
976 description
= "Load and deploy Hackfest 4 example."
980 self
.vnfd_filenames
= ("hackfest_4_vnfd.tar.gz",)
981 self
.nsd_filename
= "hackfest_4_nsd.tar.gz"
982 self
.uses_configuration
= True
983 self
.cmds
= {'1': ['ls -lrt', ], '2': ['ls -lrt', ]}
984 self
.uss
= {'1': "ubuntu", '2': "ubuntu"}
985 self
.pss
= {'1': "osm4u", '2': "osm4u"}
987 def create_descriptors(self
, engine
):
988 super().create_descriptors(engine
)
989 # Modify VNFD to add scaling
991 scaling-group-descriptor:
992 - name: "scale_dataVM"
993 max-instance-count: 10
995 - name: "auto_cpu_util_above_threshold"
996 scaling-type: "automatic"
1000 - name: "cpu_util_above_threshold"
1001 scale-in-threshold: 15
1002 scale-in-relational-operation: "LE"
1003 scale-out-threshold: 60
1004 scale-out-relational-operation: "GE"
1005 vnf-monitoring-param-ref: "all_aaa_cpu_util"
1007 - vdu-id-ref: dataVM
1009 scaling-config-action:
1010 - trigger: post-scale-out
1011 vnf-config-primitive-name-ref: touch
1012 - trigger: pre-scale-in
1013 vnf-config-primitive-name-ref: touch
1020 default-value: '/home/ubuntu/touched'
1022 engine
.test("DEPLOY{}".format(self
.step
), "Edit VNFD ", "PATCH",
1023 "/vnfpkgm/v1/vnf_packages/<{}>".format(self
.vnfds_test
[0]), headers_yaml
, payload
, 204, None, None)
1027 class TestDeployHackfest3Charmed(TestDeploy
):
1028 description
= "Load and deploy Hackfest 3charmed_ns example. Modifies it for adding scaling and performs " \
1029 "primitive actions and scaling"
1033 self
.vnfd_filenames
= ("hackfest_3charmed_vnfd.tar.gz",)
1034 self
.nsd_filename
= "hackfest_3charmed_nsd.tar.gz"
1035 self
.uses_configuration
= True
1036 self
.cmds
= {'1': [''], '2': ['ls -lrt /home/ubuntu/first-touch', ]}
1037 self
.uss
= {'1': "ubuntu", '2': "ubuntu"}
1038 self
.pss
= {'1': "osm4u", '2': "osm4u"}
1040 # def create_descriptors(self, engine):
1041 # super().create_descriptors(engine)
1042 # # Modify VNFD to add scaling
1044 # scaling-group-descriptor:
1045 # - name: "scale_dataVM"
1046 # max-instance-count: 10
1048 # - name: "auto_cpu_util_above_threshold"
1049 # scaling-type: "automatic"
1053 # - name: "cpu_util_above_threshold"
1054 # scale-in-threshold: 15
1055 # scale-in-relational-operation: "LE"
1056 # scale-out-threshold: 60
1057 # scale-out-relational-operation: "GE"
1058 # vnf-monitoring-param-ref: "all_aaa_cpu_util"
1060 # - vdu-id-ref: dataVM
1062 # scaling-config-action:
1063 # - trigger: post-scale-out
1064 # vnf-config-primitive-name-ref: touch
1065 # - trigger: pre-scale-in
1066 # vnf-config-primitive-name-ref: touch
1067 # vnf-configuration:
1073 # default-value: '/home/ubuntu/touched'
1075 # engine.test("DEPLOY{}".format(self.step), "Edit VNFD ", "PATCH",
1076 # "/vnfpkgm/v1/vnf_packages/<{}>".format(self.vnfds_test[0]),
1077 # headers_yaml, payload, 200,
1078 # r_header_yaml, yaml)
1079 # self.vnfds_test.append("DEPLOY" + str(self.step))
1082 def aditional_operations(self
, engine
, test_osm
, manual_check
):
1086 payload
= '{member_vnf_index: "2", primitive: touch, primitive_params: { filename: /home/ubuntu/OSMTESTNBI }}'
1087 engine
.test("DEPLOY{}".format(self
.step
), "Executer service primitive over NS", "POST",
1088 "/nslcm/v1/ns_instances/<{}>/action".format(self
.ns_test
), headers_yaml
, payload
,
1089 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
1090 nslcmop2_action
= "DEPLOY{}".format(self
.step
)
1092 # Wait until status is Ok
1093 self
._wait
_nslcmop
_ready
(engine
, nslcmop2_action
, timeout_deploy
)
1095 input('NS service primitive has been executed. Check that file /home/ubuntu/OSMTESTNBI is present at '
1098 cmds
= {'1': [''], '2': ['ls -lrt /home/ubuntu/OSMTESTNBI', ]}
1099 uss
= {'1': "ubuntu", '2': "ubuntu"}
1100 pss
= {'1': "osm4u", '2': "osm4u"}
1101 self
.test_ns(engine
, test_osm
, cmds
, uss
, pss
, self
.keys
, self
.timeout
)
1103 # # 2 perform scale out
1104 # payload = '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_OUT, scaleByStepData: ' \
1105 # '{scaling-group-descriptor: scale_dataVM, member-vnf-index: "1"}}}'
1106 # engine.test("DEPLOY{}".format(self.step), "Execute scale action over NS", "POST",
1107 # "/nslcm/v1/ns_instances/<{}>/scale".format(self.ns_test), headers_yaml, payload,
1108 # 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
1109 # nslcmop2_scale_out = "DEPLOY{}".format(self.step)
1110 # self._wait_nslcmop_ready(engine, nslcmop2_scale_out, timeout_deploy)
1112 # input('NS scale out done. Check that file /home/ubuntu/touched is present and new VM is created')
1113 # # TODO check automatic
1115 # # 2 perform scale in
1116 # payload = '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_IN, scaleByStepData: ' \
1117 # '{scaling-group-descriptor: scale_dataVM, member-vnf-index: "1"}}}'
1118 # engine.test("DEPLOY{}".format(self.step), "Execute scale action over NS", "POST",
1119 # "/nslcm/v1/ns_instances/<{}>/scale".format(self.ns_test), headers_yaml, payload,
1120 # 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
1121 # nslcmop2_scale_in = "DEPLOY{}".format(self.step)
1122 # self._wait_nslcmop_ready(engine, nslcmop2_scale_in, timeout_deploy)
1124 # input('NS scale in done. Check that file /home/ubuntu/touched is updated and new VM is deleted')
1125 # # TODO check automatic
1128 if __name__
== "__main__":
1132 # Disable warnings from self-signed certificates.
1133 requests
.packages
.urllib3
.disable_warnings()
1135 logging
.basicConfig(format
="%(levelname)s %(message)s", level
=logging
.ERROR
)
1136 logger
= logging
.getLogger('NBI')
1137 # load parameters and configuration
1138 opts
, args
= getopt
.getopt(sys
.argv
[1:], "hvu:p:",
1139 ["url=", "user=", "password=", "help", "version", "verbose", "no-verbose",
1140 "project=", "insecure", "timeout", "timeout-deploy", "timeout-configure",
1141 "test=", "list", "test-osm", "manual-check", "params="])
1142 url
= "https://localhost:9999/osm"
1143 user
= password
= project
= "admin"
1145 manual_check
= False
1149 "NonAuthorized": TestNonAuthorized
,
1150 "FakeVIM": TestFakeVim
,
1151 "TestUsersProjects": TestUsersProjects
,
1152 "VIM-SDN": TestVIMSDN
,
1153 "Deploy-Custom": TestDeploy
,
1154 "Deploy-Hackfest-Cirros": TestDeployHackfestCirros
,
1155 "Deploy-Hackfest-3Charmed": TestDeployHackfest3Charmed
,
1156 "Deploy-Hackfest-4": TestDeployHackfest4
,
1157 "Deploy-CirrosMacIp": TestDeployIpMac
,
1158 # "Deploy-MultiVIM": TestDeployMultiVIM,
1164 # print("parameter:", o, a)
1165 if o
== "--version":
1166 print("test version " + __version__
+ ' ' + version_date
)
1169 for test
, test_class
in test_classes
.items():
1170 print("{:20} {}".format(test
+ ":", test_class
.description
))
1172 elif o
in ("-v", "--verbose"):
1174 elif o
== "no-verbose":
1176 elif o
in ("-h", "--help"):
1179 elif o
== "--test-osm":
1181 elif o
== "--manual-check":
1185 elif o
in ("-u", "--user"):
1187 elif o
in ("-p", "--password"):
1189 elif o
== "--project":
1192 # print("asdfadf", o, a, a.split(","))
1193 for _test
in a
.split(","):
1194 if _test
not in test_classes
:
1195 print("Invalid test name '{}'. Use option '--list' to show available tests".format(_test
),
1198 test_to_do
.append(_test
)
1199 elif o
== "--params":
1200 param_key
, _
, param_value
= a
.partition("=")
1201 text_index
= len(test_to_do
)
1202 if text_index
not in test_params
:
1203 test_params
[text_index
] = {}
1204 test_params
[text_index
][param_key
] = param_value
1205 elif o
== "--insecure":
1207 elif o
== "--timeout":
1209 elif o
== "--timeout-deploy":
1210 timeout_deploy
= int(a
)
1211 elif o
== "--timeout-configure":
1212 timeout_configure
= int(a
)
1214 assert False, "Unhandled option"
1216 logger
.setLevel(logging
.WARNING
)
1218 logger
.setLevel(logging
.DEBUG
)
1220 logger
.setLevel(logging
.ERROR
)
1222 test_rest
= TestRest(url
, user
=user
, password
=password
, project
=project
)
1223 # print("tests to do:", test_to_do)
1226 for test
in test_to_do
:
1228 test_class
= test_classes
[test
]
1229 test_class().run(test_rest
, test_osm
, manual_check
, test_params
.get(text_index
))
1231 for test
, test_class
in test_classes
.items():
1232 test_class().run(test_rest
, test_osm
, manual_check
, test_params
.get(0))
1237 # # tests once authorized
1238 # for t in test_authorized_list:
1239 # test_rest.test(*t)
1242 # for t in test_admin_list1:
1243 # test_rest.test(*t)
1246 # r = test_rest.test("VNFD1", "Onboard VNFD step 1", "POST", "/vnfpkgm/v1/vnf_packages", headers_json, None,
1247 # 201, {"Location": "/vnfpkgm/v1/vnf_packages/", "Content-Type": "application/json"}, "json")
1248 # location = r.headers["Location"]
1249 # vnfd_id = location[location.rfind("/")+1:]
1250 # # print(location, vnfd_id)
1252 # # vnfd UPLOAD test
1253 # r = test_rest.test("VNFD2", "Onboard VNFD step 2 as TEXT", "PUT",
1254 # "/vnfpkgm/v1/vnf_packages/{}/package_content".format(vnfd_id),
1255 # r_header_text, "@./cirros_vnf/cirros_vnfd.yaml", 204, None, 0)
1257 # # vnfd SHOW OSM format
1258 # r = test_rest.test("VNFD3", "Show VNFD OSM format", "GET",
1259 # "/vnfpkgm/v1/vnf_packages_content/{}".format(vnfd_id),
1260 # headers_json, None, 200, r_header_json, "json")
1263 # r = test_rest.test("VNFD4", "Show VNFD SOL005 text", "GET",
1264 # "/vnfpkgm/v1/vnf_packages/{}/package_content".format(vnfd_id),
1265 # headers_text, None, 200, r_header_text, "text")
1268 # makedirs("temp", exist_ok=True)
1269 # tar = tarfile.open("temp/cirros_vnf.tar.gz", "w:gz")
1270 # tar.add("cirros_vnf")
1272 # r = test_rest.test("VNFD5", "Onboard VNFD step 3 replace with ZIP", "PUT",
1273 # "/vnfpkgm/v1/vnf_packages/{}/package_content".format(vnfd_id),
1274 # r_header_zip, "@b./temp/cirros_vnf.tar.gz", 204, None, 0)
1276 # # vnfd SHOW OSM format
1277 # r = test_rest.test("VNFD6", "Show VNFD OSM format", "GET",
1278 # "/vnfpkgm/v1/vnf_packages_content/{}".format(vnfd_id),
1279 # headers_json, None, 200, r_header_json, "json")
1282 # r = test_rest.test("VNFD7", "Show VNFD SOL005 zip", "GET",
1283 # "/vnfpkgm/v1/vnf_packages/{}/package_content".format(vnfd_id),
1284 # headers_zip, None, 200, r_header_zip, "zip")
1285 # # vnfd SHOW descriptor
1286 # r = test_rest.test("VNFD8", "Show VNFD descriptor", "GET",
1287 # "/vnfpkgm/v1/vnf_packages/{}/vnfd".format(vnfd_id),
1288 # headers_text, None, 200, r_header_text, "text")
1289 # # vnfd SHOW actifact
1290 # r = test_rest.test("VNFD9", "Show VNFD artifact", "GET",
1291 # "/vnfpkgm/v1/vnf_packages/{}/artifacts/icons/cirros-64.png".format(vnfd_id),
1292 # headers_text, None, 200, r_header_octect, "text")
1295 # # r = test_rest.test("VNFD10", "Delete VNFD SOL005 text", "DELETE",
1296 # # "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_id),
1297 # # headers_yaml, None, 204, None, 0)
1300 # r = test_rest.test("NSD1", "Onboard NSD step 1", "POST", "/nsd/v1/ns_descriptors", headers_json, None,
1301 # 201, {"Location": "/nsd/v1/ns_descriptors/", "Content-Type": "application/json"}, "json")
1302 # location = r.headers["Location"]
1303 # nsd_id = location[location.rfind("/")+1:]
1304 # # print(location, nsd_id)
1307 # r = test_rest.test("NSD2", "Onboard NSD with missing vnfd", "PUT",
1308 # "/nsd/v1/ns_descriptors/<>/nsd_content?constituent-vnfd.0.vnfd-id-ref"
1309 # "=NONEXISTING-VNFD".format(nsd_id),
1310 # r_header_text, "@./cirros_ns/cirros_nsd.yaml", 409, r_header_yaml, "yaml")
1313 # # r = test_rest.test("VNFD5", "Onboard VNFD step 3 replace with ZIP", "PUT",
1314 # # "/vnfpkgm/v1/vnf_packages/{}/package_content".format(vnfd_id),
1315 # # r_header_zip, "@b./temp/cirros_vnf.tar.gz", 204, None, 0)
1317 # r = test_rest.test("NSD2", "Onboard NSD step 2 as TEXT", "PUT",
1318 # "/nsd/v1/ns_descriptors/{}/nsd_content".format(nsd_id),
1319 # r_header_text, "@./cirros_ns/cirros_nsd.yaml", 204, None, 0)
1321 # # nsd SHOW OSM format
1322 # r = test_rest.test("NSD3", "Show NSD OSM format", "GET", "/nsd/v1/ns_descriptors_content/{}".format(nsd_id),
1323 # headers_json, None, 200, r_header_json, "json")
1326 # r = test_rest.test("NSD4", "Show NSD SOL005 text", "GET",
1327 # "/nsd/v1/ns_descriptors/{}/nsd_content".format(nsd_id),
1328 # headers_text, None, 200, r_header_text, "text")
1331 # makedirs("temp", exist_ok=True)
1332 # tar = tarfile.open("temp/cirros_ns.tar.gz", "w:gz")
1333 # tar.add("cirros_ns")
1335 # r = test_rest.test("NSD5", "Onboard NSD step 3 replace with ZIP", "PUT",
1336 # "/nsd/v1/ns_descriptors/{}/nsd_content".format(nsd_id),
1337 # r_header_zip, "@b./temp/cirros_ns.tar.gz", 204, None, 0)
1339 # # nsd SHOW OSM format
1340 # r = test_rest.test("NSD6", "Show NSD OSM format", "GET", "/nsd/v1/ns_descriptors_content/{}".format(nsd_id),
1341 # headers_json, None, 200, r_header_json, "json")
1344 # r = test_rest.test("NSD7","Show NSD SOL005 zip","GET", "/nsd/v1/ns_descriptors/{}/nsd_content".format(nsd_id),
1345 # headers_zip, None, 200, r_header_zip, "zip")
1347 # # nsd SHOW descriptor
1348 # r = test_rest.test("NSD8", "Show NSD descriptor", "GET", "/nsd/v1/ns_descriptors/{}/nsd".format(nsd_id),
1349 # headers_text, None, 200, r_header_text, "text")
1350 # # nsd SHOW actifact
1351 # r = test_rest.test("NSD9", "Show NSD artifact", "GET",
1352 # "/nsd/v1/ns_descriptors/{}/artifacts/icons/osm_2x.png".format(nsd_id),
1353 # headers_text, None, 200, r_header_octect, "text")
1356 # r = test_rest.test("VNFD10", "Delete VNFD conflict", "DELETE", "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_id),
1357 # headers_yaml, None, 409, r_header_yaml, "yaml")
1360 # r = test_rest.test("NSD10", "Delete NSD SOL005 text", "DELETE", "/nsd/v1/ns_descriptors/{}".format(nsd_id),
1361 # headers_yaml, None, 204, None, 0)
1364 # r = test_rest.test("VNFD10","Delete VNFD SOL005 text","DELETE", "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_id),
1365 # headers_yaml, None, 204, None, 0)
1368 except TestException
as e
:
1369 logger
.error(test
+ "Test {} Exception: {}".format(test
, str(e
)))
1371 except getopt
.GetoptError
as e
:
1373 print(e
, file=sys
.stderr
)
1375 except Exception as e
:
1376 logger
.critical(test
+ " Exception: " + str(e
), exc_info
=True)