projects / osm / NBI.git / blob
? search:
summary | shortlog | log | commit | commitdiff | tree
history | raw | HEAD
Reformat NBI to standardized format
[osm/NBI.git] / osm_nbi / tests / run_test.py
1 #! /usr/bin/python3
2 # -*- coding: utf-8 -*-
3
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
7 #
8 # http://www.apache.org/licenses/LICENSE-2.0
9 #
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
13 # implied.
14 # See the License for the specific language governing permissions and
15 # limitations under the License.
16
17 import getopt
18 import sys
19 import requests
20 import json
21 import logging
22 import yaml
23
24 # import json
25 # import tarfile
26 from time import sleep
27 from random import randint
28 import os
29 from sys import stderr
30 from uuid import uuid4
31 import re
32
33 __author__ = "Alfonso Tierno, alfonso.tiernosepulveda@telefonica.com"
34 __date__ = "$2018-03-01$"
35 __version__ = "0.3"
36 version_date = "Oct 2018"
37
38
39 def usage():
40 print("Usage: ", sys.argv[0], "[options]")
41 print(
42 " Performs system tests over running NBI. It can be used for real OSM test using option '--test-osm'"
43 )
44 print(
45 " If this is the case env variables 'OSMNBITEST_VIM_NAME' must be supplied to create a VIM if not exist "
46 "where deployment is done"
47 )
48 print("OPTIONS")
49 print(" -h|--help: shows this help")
50 print(" --insecure: Allows non trusted https NBI server")
51 print(" --list: list available tests")
52 print(
53 " --manual-check: Deployment tests stop after deployed to allow manual inspection. Only make sense with "
54 "'--test-osm'"
55 )
56 print(" -p|--password PASSWORD: NBI access password. 'admin' by default")
57 print(" ---project PROJECT: NBI access project. 'admin' by default")
58 print(
59 " --test TEST[,...]: Execute only a test or a comma separated list of tests"
60 )
61 print(
62 " --params key=val: params to the previous test. key can be vnfd-files, nsd-file, ns-name, ns-config"
63 )
64 print(
65 " --test-osm: If missing this test is intended for NBI only, no other OSM components are expected. Use "
66 "this flag to test the system. LCM and RO components are expected to be up and running"
67 )
68 print(
69 " --timeout TIMEOUT: General NBI timeout, by default {}s".format(timeout)
70 )
71 print(
72 " --timeout-deploy TIMEOUT: Timeout used for getting NS deployed, by default {}s".format(
73 timeout_deploy
74 )
75 )
76 print(
77 " --timeout-configure TIMEOUT: Timeout used for getting NS deployed and configured,"
78 " by default {}s".format(timeout_configure)
79 )
80 print(" -u|--user USERNAME: NBI access username. 'admin' by default")
81 print(
82 " --url URL: complete NBI server URL. 'https//localhost:9999/osm' by default"
83 )
84 print(" -v|--verbose print debug information, can be used several times")
85 print(" --no-verbose remove verbosity")
86 print(" --version: prints current version")
87 print("ENV variables used for real deployment tests with option osm-test.")
88 print(" export OSMNBITEST_VIM_NAME=vim-name")
89 print(" export OSMNBITEST_VIM_URL=vim-url")
90 print(" export OSMNBITEST_VIM_TYPE=vim-type")
91 print(" export OSMNBITEST_VIM_TENANT=vim-tenant")
92 print(" export OSMNBITEST_VIM_USER=vim-user")
93 print(" export OSMNBITEST_VIM_PASSWORD=vim-password")
94 print(' export OSMNBITEST_VIM_CONFIG="vim-config"')
95 print(' export OSMNBITEST_NS_NAME="vim-config"')
96 return
97
98
99 r_header_json = {"Content-type": "application/json"}
100 headers_json = {"Content-type": "application/json", "Accept": "application/json"}
101 r_header_yaml = {"Content-type": "application/yaml"}
102 headers_yaml = {"Content-type": "application/yaml", "Accept": "application/yaml"}
103 r_header_text = {"Content-type": "text/plain"}
104 r_header_octect = {"Content-type": "application/octet-stream"}
105 headers_text = {"Accept": "text/plain,application/yaml"}
106 r_header_zip = {"Content-type": "application/zip"}
107 headers_zip = {"Accept": "application/zip,application/yaml"}
108 headers_zip_yaml = {"Accept": "application/yaml", "Content-type": "application/zip"}
109 headers_zip_json = {"Accept": "application/json", "Content-type": "application/zip"}
110 headers_txt_json = {"Accept": "application/json", "Content-type": "text/plain"}
111 r_headers_yaml_location_vnfd = {
112 "Location": "/vnfpkgm/v1/vnf_packages_content/",
113 "Content-Type": "application/yaml",
114 }
115 r_headers_yaml_location_nsd = {
116 "Location": "/nsd/v1/ns_descriptors_content/",
117 "Content-Type": "application/yaml",
118 }
119 r_headers_yaml_location_nst = {
120 "Location": "/nst/v1/netslice_templates_content",
121 "Content-Type": "application/yaml",
122 }
123 r_headers_yaml_location_nslcmop = {
124 "Location": "nslcm/v1/ns_lcm_op_occs/",
125 "Content-Type": "application/yaml",
126 }
127 r_headers_yaml_location_nsilcmop = {
128 "Location": "/osm/nsilcm/v1/nsi_lcm_op_occs/",
129 "Content-Type": "application/yaml",
130 }
131
132 # test ones authorized
133 test_authorized_list = (
134 (
135 "AU1",
136 "Invalid vnfd id",
137 "GET",
138 "/vnfpkgm/v1/vnf_packages/non-existing-id",
139 headers_json,
140 None,
141 404,
142 r_header_json,
143 "json",
144 ),
145 (
146 "AU2",
147 "Invalid nsd id",
148 "GET",
149 "/nsd/v1/ns_descriptors/non-existing-id",
150 headers_yaml,
151 None,
152 404,
153 r_header_yaml,
154 "yaml",
155 ),
156 (
157 "AU3",
158 "Invalid nsd id",
159 "DELETE",
160 "/nsd/v1/ns_descriptors_content/non-existing-id",
161 headers_yaml,
162 None,
163 404,
164 r_header_yaml,
165 "yaml",
166 ),
167 )
168 timeout = 120 # general timeout
169 timeout_deploy = 60 * 10 # timeout for NS deploying without charms
170 timeout_configure = 60 * 20 # timeout for NS deploying and configuring
171
172
173 class TestException(Exception):
174 pass
175
176
177 class TestRest:
178 def __init__(
179 self,
180 url_base,
181 header_base=None,
182 verify=False,
183 user="admin",
184 password="admin",
185 project="admin",
186 ):
187 self.url_base = url_base
188 if header_base is None:
189 self.header_base = {}
190 else:
191 self.header_base = header_base.copy()
192 self.s = requests.session()
193 self.s.headers = self.header_base
194 self.verify = verify
195 self.token = False
196 self.user = user
197 self.password = password
198 self.project = project
199 self.vim_id = None
200 # contains ID of tests obtained from Location response header. "" key contains last obtained id
201 self.last_id = ""
202 self.test_name = None
203 self.step = 0 # number of subtest under test
204 self.passed_tests = 0
205 self.failed_tests = 0
206
207 def set_test_name(self, test_name):
208 self.test_name = test_name
209 self.step = 0
210 self.last_id = ""
211
212 def set_header(self, header):
213 self.s.headers.update(header)
214
215 def set_tet_name(self, test_name):
216 self.test_name = test_name
217
218 def unset_header(self, key):
219 if key in self.s.headers:
220 del self.s.headers[key]
221
222 def test(
223 self,
224 description,
225 method,
226 url,
227 headers,
228 payload,
229 expected_codes,
230 expected_headers,
231 expected_payload,
232 store_file=None,
233 pooling=False,
234 ):
235 """
236 Performs an http request and check http code response. Exit if different than allowed. It get the returned id
237 that can be used by following test in the URL with {name} where name is the name of the test
238 :param description: description of the test
239 :param method: HTTP method: GET,PUT,POST,DELETE,...
240 :param url: complete URL or relative URL
241 :param headers: request headers to add to the base headers
242 :param payload: Can be a dict, transformed to json, a text or a file if starts with '@'
243 :param expected_codes: expected response codes, can be int, int tuple or int range
244 :param expected_headers: expected response headers, dict with key values
245 :param expected_payload: expected payload, 0 if empty, 'yaml', 'json', 'text', 'zip', 'octet-stream'
246 :param store_file: filename to store content
247 :param pooling: if True do not count neither log this test. Because a pooling is done with many equal requests
248 :return: requests response
249 """
250 r = None
251 try:
252 if not self.s:
253 self.s = requests.session()
254 # URL
255 if not url:
256 url = self.url_base
257 elif not url.startswith("http"):
258 url = self.url_base + url
259
260 # replace url <> with the last ID
261 url = url.replace("<>", self.last_id)
262 if payload:
263 if isinstance(payload, str):
264 if payload.startswith("@"):
265 mode = "r"
266 file_name = payload[1:]
267 if payload.startswith("@b"):
268 mode = "rb"
269 file_name = payload[2:]
270 with open(file_name, mode) as f:
271 payload = f.read()
272 elif isinstance(payload, dict):
273 payload = json.dumps(payload)
274
275 if not pooling:
276 test_description = "Test {}{} {} {} {}".format(
277 self.test_name, self.step, description, method, url
278 )
279 logger.warning(test_description)
280 self.step += 1
281 stream = False
282 if expected_payload in ("zip", "octet-string") or store_file:
283 stream = True
284 __retry = 0
285 while True:
286 try:
287 r = getattr(self.s, method.lower())(
288 url,
289 data=payload,
290 headers=headers,
291 verify=self.verify,
292 stream=stream,
293 )
294 break
295 except requests.exceptions.ConnectionError as e:
296 if __retry == 2:
297 raise
298 logger.error("Exception {}. Retrying".format(e))
299 __retry += 1
300
301 if expected_payload in ("zip", "octet-string") or store_file:
302 logger.debug("RX {}".format(r.status_code))
303 else:
304 logger.debug("RX {}: {}".format(r.status_code, r.text))
305
306 # check response
307 if expected_codes:
308 if isinstance(expected_codes, int):
309 expected_codes = (expected_codes,)
310 if r.status_code not in expected_codes:
311 raise TestException(
312 "Got status {}. Expected {}. {}".format(
313 r.status_code, expected_codes, r.text
314 )
315 )
316
317 if expected_headers:
318 for header_key, header_val in expected_headers.items():
319 if header_key.lower() not in r.headers:
320 raise TestException("Header {} not present".format(header_key))
321 if header_val and header_val.lower() not in r.headers[header_key]:
322 raise TestException(
323 "Header {} does not contain {} but {}".format(
324 header_key, header_val, r.headers[header_key]
325 )
326 )
327
328 if expected_payload is not None:
329 if expected_payload == 0 and len(r.content) > 0:
330 raise TestException("Expected empty payload")
331 elif expected_payload == "json":
332 try:
333 r.json()
334 except Exception as e:
335 raise TestException(
336 "Expected json response payload, but got Exception {}".format(
337 e
338 )
339 )
340 elif expected_payload == "yaml":
341 try:
342 yaml.safe_load(r.text)
343 except Exception as e:
344 raise TestException(
345 "Expected yaml response payload, but got Exception {}".format(
346 e
347 )
348 )
349 elif expected_payload in ("zip", "octet-string"):
350 if len(r.content) == 0:
351 raise TestException(
352 "Expected some response payload, but got empty"
353 )
354 # try:
355 # tar = tarfile.open(None, 'r:gz', fileobj=r.raw)
356 # for tarinfo in tar:
357 # tarname = tarinfo.name
358 # print(tarname)
359 # except Exception as e:
360 # raise TestException("Expected zip response payload, but got Exception {}".format(e))
361 elif expected_payload == "text":
362 if len(r.content) == 0:
363 raise TestException(
364 "Expected some response payload, but got empty"
365 )
366 # r.text
367 if store_file:
368 with open(store_file, "wb") as fd:
369 for chunk in r.iter_content(chunk_size=128):
370 fd.write(chunk)
371
372 location = r.headers.get("Location")
373 if location:
374 _id = location[location.rfind("/") + 1 :]
375 if _id:
376 self.last_id = str(_id)
377 if not pooling:
378 self.passed_tests += 1
379 return r
380 except TestException as e:
381 self.failed_tests += 1
382 r_status_code = None
383 r_text = None
384 if r:
385 r_status_code = r.status_code
386 r_text = r.text
387 logger.error("{} \nRX code{}: {}".format(e, r_status_code, r_text))
388 return None
389 # exit(1)
390 except IOError as e:
391 if store_file:
392 logger.error("Cannot open file {}: {}".format(store_file, e))
393 else:
394 logger.error("Exception: {}".format(e), exc_info=True)
395 self.failed_tests += 1
396 return None
397 # exit(1)
398 except requests.exceptions.RequestException as e:
399 logger.error("Exception: {}".format(e))
400
401 def get_autorization(self): # user=None, password=None, project=None):
402 if (
403 self.token
404 ): # and self.user == user and self.password == password and self.project == project:
405 return
406 # self.user = user
407 # self.password = password
408 # self.project = project
409 r = self.test(
410 "Obtain token",
411 "POST",
412 "/admin/v1/tokens",
413 headers_json,
414 {
415 "username": self.user,
416 "password": self.password,
417 "project_id": self.project,
418 },
419 (200, 201),
420 r_header_json,
421 "json",
422 )
423 if not r:
424 return
425 response = r.json()
426 self.token = response["id"]
427 self.set_header({"Authorization": "Bearer {}".format(self.token)})
428
429 def remove_authorization(self):
430 if self.token:
431 self.test(
432 "Delete token",
433 "DELETE",
434 "/admin/v1/tokens/{}".format(self.token),
435 headers_json,
436 None,
437 (200, 201, 204),
438 None,
439 None,
440 )
441 self.token = None
442 self.unset_header("Authorization")
443
444 def get_create_vim(self, test_osm):
445 if self.vim_id:
446 return self.vim_id
447 self.get_autorization()
448 if test_osm:
449 vim_name = os.environ.get("OSMNBITEST_VIM_NAME")
450 if not vim_name:
451 raise TestException(
452 "Needed to define OSMNBITEST_VIM_XXX variables to create a real VIM for deployment"
453 )
454 else:
455 vim_name = "fakeVim"
456 # Get VIM
457 r = self.test(
458 "Get VIM ID",
459 "GET",
460 "/admin/v1/vim_accounts?name={}".format(vim_name),
461 headers_json,
462 None,
463 200,
464 r_header_json,
465 "json",
466 )
467 if not r:
468 return
469 vims = r.json()
470 if vims:
471 return vims[0]["_id"]
472 # Add VIM
473 if test_osm:
474 # check needed environ parameters:
475 if not os.environ.get("OSMNBITEST_VIM_URL") or not os.environ.get(
476 "OSMNBITEST_VIM_TENANT"
477 ):
478 raise TestException(
479 "Env OSMNBITEST_VIM_URL and OSMNBITEST_VIM_TENANT are needed for create a real VIM"
480 " to deploy on whit the --test-osm option"
481 )
482 vim_data = "{{schema_version: '1.0', name: '{}', vim_type: {}, vim_url: '{}', vim_tenant_name: '{}', " "vim_user: {}, vim_password: {}".format(
483 vim_name,
484 os.environ.get("OSMNBITEST_VIM_TYPE", "openstack"),
485 os.environ.get("OSMNBITEST_VIM_URL"),
486 os.environ.get("OSMNBITEST_VIM_TENANT"),
487 os.environ.get("OSMNBITEST_VIM_USER"),
488 os.environ.get("OSMNBITEST_VIM_PASSWORD"),
489 )
490 if os.environ.get("OSMNBITEST_VIM_CONFIG"):
491 vim_data += " ,config: {}".format(
492 os.environ.get("OSMNBITEST_VIM_CONFIG")
493 )
494 vim_data += "}"
495 else:
496 vim_data = (
497 "{schema_version: '1.0', name: fakeVim, vim_type: openstack, vim_url: 'http://10.11.12.13/fake'"
498 ", vim_tenant_name: 'vimtenant', vim_user: vimuser, vim_password: vimpassword}"
499 )
500 self.test(
501 "Create VIM",
502 "POST",
503 "/admin/v1/vim_accounts",
504 headers_yaml,
505 vim_data,
506 (201, 202),
507 {"Location": "/admin/v1/vim_accounts/", "Content-Type": "application/yaml"},
508 "yaml",
509 )
510 return self.last_id
511
512 def print_results(self):
513 print("\n\n\n--------------------------------------------")
514 print(
515 "TEST RESULTS: Total: {}, Passed: {}, Failed: {}".format(
516 self.passed_tests + self.failed_tests,
517 self.passed_tests,
518 self.failed_tests,
519 )
520 )
521 print("--------------------------------------------")
522
523 def wait_until_delete(self, url_op, timeout_delete):
524 """
525 Make a pooling until topic is not present, because of deleted
526 :param url_op:
527 :param timeout_delete:
528 :return:
529 """
530 description = "Wait to topic being deleted"
531 test_description = "Test {}{} {} {} {}".format(
532 self.test_name, self.step, description, "GET", url_op
533 )
534 logger.warning(test_description)
535 self.step += 1
536
537 wait = timeout_delete
538 while wait >= 0:
539 r = self.test(
540 description,
541 "GET",
542 url_op,
543 headers_yaml,
544 None,
545 (200, 404),
546 None,
547 r_header_yaml,
548 "yaml",
549 pooling=True,
550 )
551 if not r:
552 return
553 if r.status_code == 404:
554 self.passed_tests += 1
555 break
556 elif r.status_code == 200:
557 wait -= 5
558 sleep(5)
559 else:
560 raise TestException(
561 "Topic is not deleted after {} seconds".format(timeout_delete)
562 )
563 self.failed_tests += 1
564
565 def wait_operation_ready(self, ns_nsi, opp_id, timeout, expected_fail=False):
566 """
567 Wait until nslcmop or nsilcmop finished
568 :param ns_nsi: "ns" o "nsi"
569 :param opp_id: Id o fthe operation
570 :param timeout:
571 :param expected_fail:
572 :return: None. Updates passed/failed_tests
573 """
574 if ns_nsi == "ns":
575 url_op = "/nslcm/v1/ns_lcm_op_occs/{}".format(opp_id)
576 else:
577 url_op = "/nsilcm/v1/nsi_lcm_op_occs/{}".format(opp_id)
578 description = "Wait to {} lcm operation complete".format(ns_nsi)
579 test_description = "Test {}{} {} {} {}".format(
580 self.test_name, self.step, description, "GET", url_op
581 )
582 logger.warning(test_description)
583 self.step += 1
584 wait = timeout
585 while wait >= 0:
586 r = self.test(
587 description,
588 "GET",
589 url_op,
590 headers_json,
591 None,
592 200,
593 r_header_json,
594 "json",
595 pooling=True,
596 )
597 if not r:
598 return
599 nslcmop = r.json()
600 if "COMPLETED" in nslcmop["operationState"]:
601 if expected_fail:
602 logger.error(
603 "NS terminate has success, expecting failing: {}".format(
604 nslcmop["detailed-status"]
605 )
606 )
607 self.failed_tests += 1
608 else:
609 self.passed_tests += 1
610 break
611 elif "FAILED" in nslcmop["operationState"]:
612 if not expected_fail:
613 logger.error(
614 "NS terminate has failed: {}".format(nslcmop["detailed-status"])
615 )
616 self.failed_tests += 1
617 else:
618 self.passed_tests += 1
619 break
620
621 print(".", end="", file=stderr)
622 wait -= 10
623 sleep(10)
624 else:
625 self.failed_tests += 1
626 logger.error(
627 "NS instantiate is not terminate after {} seconds".format(timeout)
628 )
629 return
630 print("", file=stderr)
631
632
633 class TestNonAuthorized:
634 description = "Test invalid URLs. methods and no authorization"
635
636 @staticmethod
637 def run(engine, test_osm, manual_check, test_params=None):
638 engine.set_test_name("NonAuth")
639 engine.remove_authorization()
640 test_not_authorized_list = (
641 (
642 "Invalid token",
643 "GET",
644 "/admin/v1/users",
645 headers_json,
646 None,
647 401,
648 r_header_json,
649 "json",
650 ),
651 (
652 "Invalid URL",
653 "POST",
654 "/admin/v1/nonexist",
655 headers_yaml,
656 None,
657 405,
658 r_header_yaml,
659 "yaml",
660 ),
661 (
662 "Invalid version",
663 "DELETE",
664 "/admin/v2/users",
665 headers_yaml,
666 None,
667 405,
668 r_header_yaml,
669 "yaml",
670 ),
671 )
672 for t in test_not_authorized_list:
673 engine.test(*t)
674
675
676 class TestUsersProjects:
677 description = "test project and user creation"
678
679 @staticmethod
680 def run(engine, test_osm, manual_check, test_params=None):
681 engine.set_test_name("UserProject")
682 # backend = test_params.get("backend") if test_params else None # UNUSED
683
684 # Initialisation
685 p1 = p2 = p3 = None
686 padmin = pbad = None
687 u1 = u2 = u3 = u4 = None
688
689 engine.get_autorization()
690
691 res = engine.test(
692 "Create project non admin 1",
693 "POST",
694 "/admin/v1/projects",
695 headers_json,
696 {"name": "P1"},
697 (201, 204),
698 {"Location": "/admin/v1/projects/", "Content-Type": "application/json"},
699 "json",
700 )
701 p1 = engine.last_id if res else None
702
703 res = engine.test(
704 "Create project admin",
705 "POST",
706 "/admin/v1/projects",
707 headers_json,
708 {"name": "Padmin", "admin": True},
709 (201, 204),
710 {"Location": "/admin/v1/projects/", "Content-Type": "application/json"},
711 "json",
712 )
713 padmin = engine.last_id if res else None
714
715 res = engine.test(
716 "Create project bad format",
717 "POST",
718 "/admin/v1/projects",
719 headers_json,
720 {"name": 1},
721 (400, 422),
722 r_header_json,
723 "json",
724 )
725 pbad = engine.last_id if res else None
726
727 res = engine.test(
728 "Get project admin role",
729 "GET",
730 "/admin/v1/roles?name=project_admin",
731 headers_json,
732 {},
733 (200),
734 {"Content-Type": "application/json"},
735 "json",
736 )
737 rpa = res.json()[0]["_id"] if res else None
738 res = engine.test(
739 "Get project user role",
740 "GET",
741 "/admin/v1/roles?name=project_user",
742 headers_json,
743 {},
744 (200),
745 {"Content-Type": "application/json"},
746 "json",
747 )
748 rpu = res.json()[0]["_id"] if res else None
749 res = engine.test(
750 "Get system admin role",
751 "GET",
752 "/admin/v1/roles?name=system_admin",
753 headers_json,
754 {},
755 (200),
756 {"Content-Type": "application/json"},
757 "json",
758 )
759 rsa = res.json()[0]["_id"] if res else None
760
761 data = {"username": "U1", "password": "pw1"}
762 p2 = uuid4().hex
763 data["project_role_mappings"] = [
764 {"project": p1, "role": rpa},
765 {"project": p2, "role": rpa},
766 {"project": padmin, "role": rpu},
767 ]
768 rc = 201
769 xhd = {"Location": "/admin/v1/users/", "Content-Type": "application/json"}
770 res = engine.test(
771 "Create user with bad project and force",
772 "POST",
773 "/admin/v1/users?FORCE=True",
774 headers_json,
775 data,
776 rc,
777 xhd,
778 "json",
779 )
780 if res:
781 u1 = engine.last_id
782 else:
783 # User is created sometimes even though an exception is raised
784 res = engine.test(
785 "Get user U1",
786 "GET",
787 "/admin/v1/users?username=U1",
788 headers_json,
789 {},
790 (200),
791 {"Content-Type": "application/json"},
792 "json",
793 )
794 u1 = res.json()[0]["_id"] if res else None
795
796 data = {"username": "U2", "password": "pw2"}
797 data["project_role_mappings"] = [
798 {"project": p1, "role": rpa},
799 {"project": padmin, "role": rsa},
800 ]
801 res = engine.test(
802 "Create user 2",
803 "POST",
804 "/admin/v1/users",
805 headers_json,
806 data,
807 201,
808 {"Location": "/admin/v1/users/", "Content-Type": "application/json"},
809 "json",
810 )
811 u2 = engine.last_id if res else None
812
813 if u1:
814 ftt = "project_role_mappings"
815 xpr = [{"project": p1, "role": rpa}, {"project": padmin, "role": rpu}]
816 data = {ftt: xpr}
817 engine.test(
818 "Edit user U1, delete P2 project",
819 "PATCH",
820 "/admin/v1/users/" + u1,
821 headers_json,
822 data,
823 204,
824 None,
825 None,
826 )
827 res = engine.test(
828 "Check user U1, contains the right projects",
829 "GET",
830 "/admin/v1/users/" + u1,
831 headers_json,
832 None,
833 200,
834 None,
835 json,
836 )
837 if res:
838 rj = res.json()
839 xpr[0]["project_name"] = "P1"
840 xpr[0]["role_name"] = "project_admin"
841 xpr[1]["project_name"] = "Padmin"
842 xpr[1]["role_name"] = "project_user"
843 ok = True
844 for pr in rj[ftt]:
845 if pr not in xpr:
846 ok = False
847 for pr in xpr:
848 if pr not in rj[ftt]:
849 ok = False
850 if not ok:
851 logger.error(
852 "User {} '{}' are different than expected '{}'. Edition was not done properly".format(
853 ftt, rj[ftt], xpr
854 )
855 )
856 engine.failed_tests += 1
857
858 p2 = None # To prevent deletion attempts
859
860 # Add a test of 'default project' for Keystone?
861
862 if u2:
863 engine.test(
864 "Edit user U2, change password",
865 "PUT",
866 "/admin/v1/users/" + u2,
867 headers_json,
868 {"password": "pw2_new"},
869 204,
870 None,
871 None,
872 )
873
874 if p1:
875 engine.test(
876 "Change to project P1 non existing",
877 "POST",
878 "/admin/v1/tokens/",
879 headers_json,
880 {"project_id": p1},
881 401,
882 r_header_json,
883 "json",
884 )
885
886 if u2 and p1:
887 res = engine.test(
888 "Change to user U2 project P1",
889 "POST",
890 "/admin/v1/tokens",
891 headers_json,
892 {"username": "U2", "password": "pw2_new", "project_id": "P1"},
893 (200, 201),
894 r_header_json,
895 "json",
896 )
897 if res:
898 rj = res.json()
899 engine.set_header({"Authorization": "Bearer {}".format(rj["id"])})
900
901 engine.test(
902 "Edit user projects non admin",
903 "PUT",
904 "/admin/v1/users/U1",
905 headers_json,
906 {"remove_project_role_mappings": [{"project": "P1", "role": None}]},
907 401,
908 r_header_json,
909 "json",
910 )
911
912 res = engine.test(
913 "Add new project non admin",
914 "POST",
915 "/admin/v1/projects",
916 headers_json,
917 {"name": "P2"},
918 401,
919 r_header_json,
920 "json",
921 )
922 if res is None or res.status_code == 201:
923 # The project has been created even though it shouldn't
924 res = engine.test(
925 "Get project P2",
926 "GET",
927 "/admin/v1/projects/P2",
928 headers_json,
929 None,
930 200,
931 r_header_json,
932 "json",
933 )
934 p2 = res.json()["_id"] if res else None
935
936 if p1:
937 data = {"username": "U3", "password": "pw3"}
938 data["project_role_mappings"] = [{"project": p1, "role": rpu}]
939 res = engine.test(
940 "Add new user non admin",
941 "POST",
942 "/admin/v1/users",
943 headers_json,
944 data,
945 401,
946 r_header_json,
947 "json",
948 )
949 if res is None or res.status_code == 201:
950 # The user has been created even though it shouldn't
951 res = engine.test(
952 "Get user U3",
953 "GET",
954 "/admin/v1/users/U3",
955 headers_json,
956 None,
957 200,
958 r_header_json,
959 "json",
960 )
961 u3 = res.json()["_id"] if res else None
962 else:
963 u3 = None
964
965 if padmin:
966 res = engine.test(
967 "Change to user U2 project Padmin",
968 "POST",
969 "/admin/v1/tokens",
970 headers_json,
971 {
972 "project_id": "Padmin"
973 }, # Caused a Keystone authentication error
974 # {"username": "U2", "password": "pw2_new", "project_id": "Padmin"},
975 (200, 201),
976 r_header_json,
977 "json",
978 )
979 if res:
980 rj = res.json()
981 engine.set_header(
982 {"Authorization": "Bearer {}".format(rj["id"])}
983 )
984
985 res = engine.test(
986 "Add new project admin",
987 "POST",
988 "/admin/v1/projects",
989 headers_json,
990 {"name": "P3"},
991 (201, 204),
992 {
993 "Location": "/admin/v1/projects/",
994 "Content-Type": "application/json",
995 },
996 "json",
997 )
998 p3 = engine.last_id if res else None
999
1000 if p1:
1001 data = {"username": "U4", "password": "pw4"}
1002 data["project_role_mappings"] = [
1003 {"project": p1, "role": rpa}
1004 ]
1005 res = engine.test(
1006 "Add new user admin",
1007 "POST",
1008 "/admin/v1/users",
1009 headers_json,
1010 data,
1011 (201, 204),
1012 {
1013 "Location": "/admin/v1/users/",
1014 "Content-Type": "application/json",
1015 },
1016 "json",
1017 )
1018 u4 = engine.last_id if res else None
1019 else:
1020 u4 = None
1021
1022 if u4 and p3:
1023 data = {
1024 "project_role_mappings": [{"project": p3, "role": rpa}]
1025 }
1026 engine.test(
1027 "Edit user projects admin",
1028 "PUT",
1029 "/admin/v1/users/U4",
1030 headers_json,
1031 data,
1032 204,
1033 None,
1034 None,
1035 )
1036 # Project is deleted even though it shouldn't - PROVISIONAL?
1037 res = engine.test(
1038 "Delete project P3 conflict",
1039 "DELETE",
1040 "/admin/v1/projects/" + p3,
1041 headers_json,
1042 None,
1043 409,
1044 None,
1045 None,
1046 )
1047 if res and res.status_code in (200, 204):
1048 p3 = None
1049 if p3:
1050 res = engine.test(
1051 "Delete project P3 forcing",
1052 "DELETE",
1053 "/admin/v1/projects/" + p3 + "?FORCE=True",
1054 headers_json,
1055 None,
1056 204,
1057 None,
1058 None,
1059 )
1060 if res and res.status_code in (200, 204):
1061 p3 = None
1062
1063 if u2:
1064 res = engine.test(
1065 "Delete user U2. Conflict deleting own user",
1066 "DELETE",
1067 "/admin/v1/users/" + u2,
1068 headers_json,
1069 None,
1070 409,
1071 r_header_json,
1072 "json",
1073 )
1074 if res is None or res.status_code in (200, 204):
1075 u2 = None
1076 if u4:
1077 res = engine.test(
1078 "Delete user U4",
1079 "DELETE",
1080 "/admin/v1/users/" + u4,
1081 headers_json,
1082 None,
1083 204,
1084 None,
1085 None,
1086 )
1087 if res and res.status_code in (200, 204):
1088 u4 = None
1089 if p3:
1090 res = engine.test(
1091 "Delete project P3",
1092 "DELETE",
1093 "/admin/v1/projects/" + p3,
1094 headers_json,
1095 None,
1096 204,
1097 None,
1098 None,
1099 )
1100 if res and res.status_code in (200, 204):
1101 p3 = None
1102
1103 if u3:
1104 res = engine.test(
1105 "Delete user U3",
1106 "DELETE",
1107 "/admin/v1/users/" + u3,
1108 headers_json,
1109 None,
1110 204,
1111 None,
1112 None,
1113 )
1114 if res:
1115 u3 = None
1116
1117 # change to admin
1118 engine.remove_authorization() # To force get authorization
1119 engine.get_autorization()
1120 if u1:
1121 engine.test(
1122 "Delete user U1",
1123 "DELETE",
1124 "/admin/v1/users/" + u1,
1125 headers_json,
1126 None,
1127 204,
1128 None,
1129 None,
1130 )
1131 if u2:
1132 engine.test(
1133 "Delete user U2",
1134 "DELETE",
1135 "/admin/v1/users/" + u2,
1136 headers_json,
1137 None,
1138 204,
1139 None,
1140 None,
1141 )
1142 if u3:
1143 engine.test(
1144 "Delete user U3",
1145 "DELETE",
1146 "/admin/v1/users/" + u3,
1147 headers_json,
1148 None,
1149 204,
1150 None,
1151 None,
1152 )
1153 if u4:
1154 engine.test(
1155 "Delete user U4",
1156 "DELETE",
1157 "/admin/v1/users/" + u4,
1158 headers_json,
1159 None,
1160 204,
1161 None,
1162 None,
1163 )
1164 if p1:
1165 engine.test(
1166 "Delete project P1",
1167 "DELETE",
1168 "/admin/v1/projects/" + p1,
1169 headers_json,
1170 None,
1171 204,
1172 None,
1173 None,
1174 )
1175 if p2:
1176 engine.test(
1177 "Delete project P2",
1178 "DELETE",
1179 "/admin/v1/projects/" + p2,
1180 headers_json,
1181 None,
1182 204,
1183 None,
1184 None,
1185 )
1186 if p3:
1187 engine.test(
1188 "Delete project P3",
1189 "DELETE",
1190 "/admin/v1/projects/" + p3,
1191 headers_json,
1192 None,
1193 204,
1194 None,
1195 None,
1196 )
1197 if padmin:
1198 engine.test(
1199 "Delete project Padmin",
1200 "DELETE",
1201 "/admin/v1/projects/" + padmin,
1202 headers_json,
1203 None,
1204 204,
1205 None,
1206 None,
1207 )
1208 if pbad:
1209 engine.test(
1210 "Delete bad project",
1211 "DELETE",
1212 "/admin/v1/projects/" + pbad,
1213 headers_json,
1214 None,
1215 204,
1216 None,
1217 None,
1218 )
1219
1220 # BEGIN New Tests - Addressing Projects/Users by Name/ID
1221 pid1 = pid2 = None
1222 uid1 = uid2 = None
1223 res = engine.test(
1224 "Create new project P1",
1225 "POST",
1226 "/admin/v1/projects",
1227 headers_json,
1228 {"name": "P1"},
1229 201,
1230 {"Location": "/admin/v1/projects/", "Content-Type": "application/json"},
1231 "json",
1232 )
1233 if res:
1234 pid1 = res.json()["id"]
1235 # print("# pid =", pid1)
1236 res = engine.test(
1237 "Create new project P2",
1238 "POST",
1239 "/admin/v1/projects",
1240 headers_json,
1241 {"name": "P2"},
1242 201,
1243 {"Location": "/admin/v1/projects/", "Content-Type": "application/json"},
1244 "json",
1245 )
1246 if res:
1247 pid2 = res.json()["id"]
1248 # print("# pid =", pid2)
1249 data = {"username": "U1", "password": "pw1"}
1250 data["project_role_mappings"] = [{"project": pid1, "role": rpu}]
1251 res = engine.test(
1252 "Create new user U1",
1253 "POST",
1254 "/admin/v1/users",
1255 headers_json,
1256 data,
1257 201,
1258 {"Location": "/admin/v1/users/", "Content-Type": "application/json"},
1259 "json",
1260 )
1261 if res:
1262 uid1 = res.json()["id"]
1263 # print("# uid =", uid1)
1264 data = {"username": "U2", "password": "pw2"}
1265 data["project_role_mappings"] = [{"project": pid2, "role": rpu}]
1266 res = engine.test(
1267 "Create new user U2",
1268 "POST",
1269 "/admin/v1/users",
1270 headers_json,
1271 data,
1272 201,
1273 {"Location": "/admin/v1/users/", "Content-Type": "application/json"},
1274 "json",
1275 )
1276 if res:
1277 uid2 = res.json()["id"]
1278 # print("# uid =", uid2)
1279 if pid1:
1280 engine.test(
1281 "Get Project P1 by Name",
1282 "GET",
1283 "/admin/v1/projects/P1",
1284 headers_json,
1285 None,
1286 200,
1287 None,
1288 "json",
1289 )
1290 engine.test(
1291 "Get Project P1 by ID",
1292 "GET",
1293 "/admin/v1/projects/" + pid1,
1294 headers_json,
1295 None,
1296 200,
1297 None,
1298 "json",
1299 )
1300 if uid1:
1301 engine.test(
1302 "Get User U1 by Name",
1303 "GET",
1304 "/admin/v1/users/U1",
1305 headers_json,
1306 None,
1307 200,
1308 None,
1309 "json",
1310 )
1311 engine.test(
1312 "Get User U1 by ID",
1313 "GET",
1314 "/admin/v1/users/" + uid1,
1315 headers_json,
1316 None,
1317 200,
1318 None,
1319 "json",
1320 )
1321 if pid1:
1322 res = engine.test(
1323 "Rename Project P1 by Name",
1324 "PUT",
1325 "/admin/v1/projects/P1",
1326 headers_json,
1327 {"name": "P3"},
1328 204,
1329 None,
1330 None,
1331 )
1332 if res:
1333 engine.test(
1334 "Get Project P1 by new Name",
1335 "GET",
1336 "/admin/v1/projects/P3",
1337 headers_json,
1338 None,
1339 200,
1340 None,
1341 "json",
1342 )
1343 if pid2:
1344 res = engine.test(
1345 "Rename Project P2 by ID",
1346 "PUT",
1347 "/admin/v1/projects/" + pid2,
1348 headers_json,
1349 {"name": "P4"},
1350 204,
1351 None,
1352 None,
1353 )
1354 if res:
1355 engine.test(
1356 "Get Project P2 by new Name",
1357 "GET",
1358 "/admin/v1/projects/P4",
1359 headers_json,
1360 None,
1361 200,
1362 None,
1363 "json",
1364 )
1365
1366 if uid1:
1367 res = engine.test(
1368 "Rename User U1 by Name",
1369 "PUT",
1370 "/admin/v1/users/U1",
1371 headers_json,
1372 {"username": "U3"},
1373 204,
1374 None,
1375 None,
1376 )
1377 if res:
1378 engine.test(
1379 "Get User U1 by new Name",
1380 "GET",
1381 "/admin/v1/users/U3",
1382 headers_json,
1383 None,
1384 200,
1385 None,
1386 "json",
1387 )
1388
1389 if uid2:
1390 res = engine.test(
1391 "Rename User U2 by ID",
1392 "PUT",
1393 "/admin/v1/users/" + uid2,
1394 headers_json,
1395 {"username": "U4"},
1396 204,
1397 None,
1398 None,
1399 )
1400 if res:
1401 engine.test(
1402 "Get User U2 by new Name",
1403 "GET",
1404 "/admin/v1/users/U4",
1405 headers_json,
1406 None,
1407 200,
1408 None,
1409 "json",
1410 )
1411 if uid1:
1412 res = engine.test(
1413 "Delete User U1 by Name",
1414 "DELETE",
1415 "/admin/v1/users/U3",
1416 headers_json,
1417 None,
1418 204,
1419 None,
1420 None,
1421 )
1422 if res:
1423 uid1 = None
1424
1425 if uid2:
1426 res = engine.test(
1427 "Delete User U2 by ID",
1428 "DELETE",
1429 "/admin/v1/users/" + uid2,
1430 headers_json,
1431 None,
1432 204,
1433 None,
1434 None,
1435 )
1436 if res:
1437 uid2 = None
1438
1439 if pid1:
1440 res = engine.test(
1441 "Delete Project P1 by Name",
1442 "DELETE",
1443 "/admin/v1/projects/P3",
1444 headers_json,
1445 None,
1446 204,
1447 None,
1448 None,
1449 )
1450 if res:
1451 pid1 = None
1452
1453 if pid2:
1454 res = engine.test(
1455 "Delete Project P2 by ID",
1456 "DELETE",
1457 "/admin/v1/projects/" + pid2,
1458 headers_json,
1459 None,
1460 204,
1461 None,
1462 None,
1463 )
1464 if res:
1465 pid2 = None
1466
1467 # END New Tests - Addressing Projects/Users by Name
1468
1469 # CLEANUP
1470 if pid1:
1471 engine.test(
1472 "Delete Project P1",
1473 "DELETE",
1474 "/admin/v1/projects/" + pid1,
1475 headers_json,
1476 None,
1477 204,
1478 None,
1479 None,
1480 )
1481 if pid2:
1482 engine.test(
1483 "Delete Project P2",
1484 "DELETE",
1485 "/admin/v1/projects/" + pid2,
1486 headers_json,
1487 None,
1488 204,
1489 None,
1490 None,
1491 )
1492 if uid1:
1493 engine.test(
1494 "Delete User U1",
1495 "DELETE",
1496 "/admin/v1/users/" + uid1,
1497 headers_json,
1498 None,
1499 204,
1500 None,
1501 None,
1502 )
1503 if uid2:
1504 engine.test(
1505 "Delete User U2",
1506 "DELETE",
1507 "/admin/v1/users/" + uid2,
1508 headers_json,
1509 None,
1510 204,
1511 None,
1512 None,
1513 )
1514
1515 engine.remove_authorization() # To finish
1516
1517
1518 class TestProjectsDescriptors:
1519 description = "test descriptors visibility among projects"
1520
1521 @staticmethod
1522 def run(engine, test_osm, manual_check, test_params=None):
1523 vnfd_ids = []
1524 engine.set_test_name("ProjectDescriptors")
1525 engine.get_autorization()
1526
1527 project_admin_id = None
1528 res = engine.test(
1529 "Get my project Padmin",
1530 "GET",
1531 "/admin/v1/projects/{}".format(engine.project),
1532 headers_json,
1533 None,
1534 200,
1535 r_header_json,
1536 "json",
1537 )
1538 if res:
1539 response = res.json()
1540 project_admin_id = response["_id"]
1541 engine.test(
1542 "Create project Padmin",
1543 "POST",
1544 "/admin/v1/projects",
1545 headers_json,
1546 {"name": "Padmin", "admin": True},
1547 (201, 204),
1548 {"Location": "/admin/v1/projects/", "Content-Type": "application/json"},
1549 "json",
1550 )
1551 engine.test(
1552 "Create project P2",
1553 "POST",
1554 "/admin/v1/projects",
1555 headers_json,
1556 {"name": "P2"},
1557 (201, 204),
1558 {"Location": "/admin/v1/projects/", "Content-Type": "application/json"},
1559 "json",
1560 )
1561 engine.test(
1562 "Create project P3",
1563 "POST",
1564 "/admin/v1/projects",
1565 headers_json,
1566 {"name": "P3"},
1567 (201, 204),
1568 {"Location": "/admin/v1/projects/", "Content-Type": "application/json"},
1569 "json",
1570 )
1571
1572 engine.test(
1573 "Create user U1",
1574 "POST",
1575 "/admin/v1/users",
1576 headers_json,
1577 {
1578 "username": "U1",
1579 "password": "pw1",
1580 "project_role_mappings": [
1581 {"project": "Padmin", "role": "system_admin"},
1582 {"project": "P2", "role": "project_admin"},
1583 {"project": "P3", "role": "project_admin"},
1584 ],
1585 },
1586 201,
1587 {"Location": "/admin/v1/users/", "Content-Type": "application/json"},
1588 "json",
1589 )
1590
1591 engine.test(
1592 "Onboard VNFD id1",
1593 "POST",
1594 "/vnfpkgm/v1/vnf_packages_content?id=id1",
1595 headers_yaml,
1596 TestDescriptors.vnfd_empty,
1597 201,
1598 r_headers_yaml_location_vnfd,
1599 "yaml",
1600 )
1601 vnfd_ids.append(engine.last_id)
1602 engine.test(
1603 "Onboard VNFD id2 PUBLIC",
1604 "POST",
1605 "/vnfpkgm/v1/vnf_packages_content?id=id2&PUBLIC=TRUE",
1606 headers_yaml,
1607 TestDescriptors.vnfd_empty,
1608 201,
1609 r_headers_yaml_location_vnfd,
1610 "yaml",
1611 )
1612 vnfd_ids.append(engine.last_id)
1613 engine.test(
1614 "Onboard VNFD id3",
1615 "POST",
1616 "/vnfpkgm/v1/vnf_packages_content?id=id3&PUBLIC=FALSE",
1617 headers_yaml,
1618 TestDescriptors.vnfd_empty,
1619 201,
1620 r_headers_yaml_location_vnfd,
1621 "yaml",
1622 )
1623 vnfd_ids.append(engine.last_id)
1624
1625 res = engine.test(
1626 "Get VNFD descriptors",
1627 "GET",
1628 "/vnfpkgm/v1/vnf_packages?id=id1,id2,id3",
1629 headers_json,
1630 None,
1631 200,
1632 r_header_json,
1633 "json",
1634 )
1635 response = res.json()
1636 if len(response) != 3:
1637 logger.error(
1638 "Only 3 vnfds should be present for project admin. {} listed".format(
1639 len(response)
1640 )
1641 )
1642 engine.failed_tests += 1
1643
1644 # Change to other project Padmin
1645 res = engine.test(
1646 "Change to user U1 project Padmin",
1647 "POST",
1648 "/admin/v1/tokens",
1649 headers_json,
1650 {"username": "U1", "password": "pw1", "project_id": "Padmin"},
1651 (200, 201),
1652 r_header_json,
1653 "json",
1654 )
1655 if res:
1656 response = res.json()
1657 engine.set_header({"Authorization": "Bearer {}".format(response["id"])})
1658
1659 # list vnfds
1660 res = engine.test(
1661 "List VNFD descriptors for Padmin",
1662 "GET",
1663 "/vnfpkgm/v1/vnf_packages",
1664 headers_json,
1665 None,
1666 200,
1667 r_header_json,
1668 "json",
1669 )
1670 response = res.json()
1671 if len(response) != 0:
1672 logger.error(
1673 "Only 0 vnfds should be present for project Padmin. {} listed".format(
1674 len(response)
1675 )
1676 )
1677 engine.failed_tests += 1
1678
1679 # list Public vnfds
1680 res = engine.test(
1681 "List VNFD public descriptors",
1682 "GET",
1683 "/vnfpkgm/v1/vnf_packages?PUBLIC=True",
1684 headers_json,
1685 None,
1686 200,
1687 r_header_json,
1688 "json",
1689 )
1690 response = res.json()
1691 if len(response) != 1:
1692 logger.error(
1693 "Only 1 vnfds should be present for project Padmin. {} listed".format(
1694 len(response)
1695 )
1696 )
1697 engine.failed_tests += 1
1698
1699 # list vnfds belonging to project "admin"
1700 res = engine.test(
1701 "List VNFD of admin project",
1702 "GET",
1703 "/vnfpkgm/v1/vnf_packages?ADMIN={}".format(project_admin_id),
1704 headers_json,
1705 None,
1706 200,
1707 r_header_json,
1708 "json",
1709 )
1710 if res:
1711 response = res.json()
1712 if len(response) != 3:
1713 logger.error(
1714 "Only 3 vnfds should be present for project Padmin. {} listed".format(
1715 len(response)
1716 )
1717 )
1718 engine.failed_tests += 1
1719
1720 # Get Public vnfds
1721 engine.test(
1722 "Get VNFD public descriptors",
1723 "GET",
1724 "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_ids[1]),
1725 headers_json,
1726 None,
1727 200,
1728 r_header_json,
1729 "json",
1730 )
1731 # Edit not owned vnfd
1732 engine.test(
1733 "Edit VNFD ",
1734 "PATCH",
1735 "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_ids[0]),
1736 headers_yaml,
1737 "{name: pepe}",
1738 404,
1739 r_header_yaml,
1740 "yaml",
1741 )
1742
1743 # Add to my catalog
1744 engine.test(
1745 "Add VNFD id2 to my catalog",
1746 "PATCH",
1747 "/vnfpkgm/v1/vnf_packages/{}?SET_PROJECT".format(vnfd_ids[1]),
1748 headers_json,
1749 None,
1750 204,
1751 None,
1752 0,
1753 )
1754
1755 # Add a new vnfd
1756 engine.test(
1757 "Onboard VNFD id4",
1758 "POST",
1759 "/vnfpkgm/v1/vnf_packages_content?id=id4",
1760 headers_yaml,
1761 TestDescriptors.vnfd_empty,
1762 201,
1763 r_headers_yaml_location_vnfd,
1764 "yaml",
1765 )
1766 vnfd_ids.append(engine.last_id)
1767
1768 # list vnfds
1769 res = engine.test(
1770 "List VNFD public descriptors",
1771 "GET",
1772 "/vnfpkgm/v1/vnf_packages",
1773 headers_json,
1774 None,
1775 200,
1776 r_header_json,
1777 "json",
1778 )
1779 response = res.json()
1780 if len(response) != 2:
1781 logger.error(
1782 "Only 2 vnfds should be present for project Padmin. {} listed".format(
1783 len(response)
1784 )
1785 )
1786 engine.failed_tests += 1
1787
1788 if manual_check:
1789 input(
1790 "VNFDs have been omboarded. Perform manual check and press enter to resume"
1791 )
1792
1793 test_rest.test(
1794 "Delete VNFD id2",
1795 "DELETE",
1796 "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_ids[1]),
1797 headers_yaml,
1798 None,
1799 204,
1800 None,
1801 0,
1802 )
1803
1804 # change to admin project
1805 engine.remove_authorization() # To force get authorization
1806 engine.get_autorization()
1807 test_rest.test(
1808 "Delete VNFD id1",
1809 "DELETE",
1810 "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_ids[0]),
1811 headers_yaml,
1812 None,
1813 204,
1814 None,
1815 0,
1816 )
1817 test_rest.test(
1818 "Delete VNFD id2",
1819 "DELETE",
1820 "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_ids[1]),
1821 headers_yaml,
1822 None,
1823 204,
1824 None,
1825 0,
1826 )
1827 test_rest.test(
1828 "Delete VNFD id3",
1829 "DELETE",
1830 "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_ids[2]),
1831 headers_yaml,
1832 None,
1833 204,
1834 None,
1835 0,
1836 )
1837 test_rest.test(
1838 "Delete VNFD id4",
1839 "DELETE",
1840 "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_ids[3]),
1841 headers_yaml,
1842 None,
1843 404,
1844 r_header_yaml,
1845 "yaml",
1846 )
1847 test_rest.test(
1848 "Delete VNFD id4",
1849 "DELETE",
1850 "/vnfpkgm/v1/vnf_packages/{}?ADMIN".format(vnfd_ids[3]),
1851 headers_yaml,
1852 None,
1853 204,
1854 None,
1855 0,
1856 )
1857 # Get Public vnfds
1858 engine.test(
1859 "Get VNFD deleted id1",
1860 "GET",
1861 "/vnfpkgm/v1/vnf_packages/{}?ADMIN".format(vnfd_ids[0]),
1862 headers_json,
1863 None,
1864 404,
1865 r_header_json,
1866 "json",
1867 )
1868 engine.test(
1869 "Get VNFD deleted id2",
1870 "GET",
1871 "/vnfpkgm/v1/vnf_packages/{}?ADMIN".format(vnfd_ids[1]),
1872 headers_json,
1873 None,
1874 404,
1875 r_header_json,
1876 "json",
1877 )
1878 engine.test(
1879 "Get VNFD deleted id3",
1880 "GET",
1881 "/vnfpkgm/v1/vnf_packages/{}?ADMIN".format(vnfd_ids[2]),
1882 headers_json,
1883 None,
1884 404,
1885 r_header_json,
1886 "json",
1887 )
1888 engine.test(
1889 "Get VNFD deleted id4",
1890 "GET",
1891 "/vnfpkgm/v1/vnf_packages/{}?ADMIN".format(vnfd_ids[3]),
1892 headers_json,
1893 None,
1894 404,
1895 r_header_json,
1896 "json",
1897 )
1898
1899 engine.test(
1900 "Delete user U1",
1901 "DELETE",
1902 "/admin/v1/users/U1",
1903 headers_json,
1904 None,
1905 204,
1906 None,
1907 None,
1908 )
1909 engine.test(
1910 "Delete project Padmin",
1911 "DELETE",
1912 "/admin/v1/projects/Padmin",
1913 headers_json,
1914 None,
1915 204,
1916 None,
1917 None,
1918 )
1919 engine.test(
1920 "Delete project P2",
1921 "DELETE",
1922 "/admin/v1/projects/P2",
1923 headers_json,
1924 None,
1925 204,
1926 None,
1927 None,
1928 )
1929 engine.test(
1930 "Delete project P3",
1931 "DELETE",
1932 "/admin/v1/projects/P3",
1933 headers_json,
1934 None,
1935 204,
1936 None,
1937 None,
1938 )
1939
1940
1941 class TestFakeVim:
1942 description = "Creates/edit/delete fake VIMs and SDN controllers"
1943
1944 def __init__(self):
1945 self.vim = {
1946 "schema_version": "1.0",
1947 "schema_type": "No idea",
1948 "name": "myVim",
1949 "description": "Descriptor name",
1950 "vim_type": "openstack",
1951 "vim_url": "http://localhost:/vim",
1952 "vim_tenant_name": "vimTenant",
1953 "vim_user": "user",
1954 "vim_password": "password",
1955 "config": {"config_param": 1},
1956 }
1957 self.sdn = {
1958 "name": "sdn-name",
1959 "description": "sdn-description",
1960 "dpid": "50:50:52:54:00:94:21:21",
1961 "ip": "192.168.15.17",
1962 "port": 8080,
1963 "type": "opendaylight",
1964 "version": "3.5.6",
1965 "user": "user",
1966 "password": "passwd",
1967 }
1968 self.port_mapping = [
1969 {
1970 "compute_node": "compute node 1",
1971 "ports": [
1972 {
1973 "pci": "0000:81:00.0",
1974 "switch_port": "port-2/1",
1975 "switch_mac": "52:54:00:94:21:21",
1976 },
1977 {
1978 "pci": "0000:81:00.1",
1979 "switch_port": "port-2/2",
1980 "switch_mac": "52:54:00:94:21:22",
1981 },
1982 ],
1983 },
1984 {
1985 "compute_node": "compute node 2",
1986 "ports": [
1987 {
1988 "pci": "0000:81:00.0",
1989 "switch_port": "port-2/3",
1990 "switch_mac": "52:54:00:94:21:23",
1991 },
1992 {
1993 "pci": "0000:81:00.1",
1994 "switch_port": "port-2/4",
1995 "switch_mac": "52:54:00:94:21:24",
1996 },
1997 ],
1998 },
1999 ]
2000
2001 def run(self, engine, test_osm, manual_check, test_params=None):
2002
2003 vim_bad = self.vim.copy()
2004 vim_bad.pop("name")
2005
2006 engine.set_test_name("FakeVim")
2007 engine.get_autorization()
2008 engine.test(
2009 "Create VIM",
2010 "POST",
2011 "/admin/v1/vim_accounts",
2012 headers_json,
2013 self.vim,
2014 (201, 202),
2015 {"Location": "/admin/v1/vim_accounts/", "Content-Type": "application/json"},
2016 "json",
2017 )
2018 vim_id = engine.last_id
2019 engine.test(
2020 "Create VIM without name, bad schema",
2021 "POST",
2022 "/admin/v1/vim_accounts",
2023 headers_json,
2024 vim_bad,
2025 422,
2026 None,
2027 headers_json,
2028 )
2029 engine.test(
2030 "Create VIM name repeated",
2031 "POST",
2032 "/admin/v1/vim_accounts",
2033 headers_json,
2034 self.vim,
2035 409,
2036 None,
2037 headers_json,
2038 )
2039 engine.test(
2040 "Show VIMs",
2041 "GET",
2042 "/admin/v1/vim_accounts",
2043 headers_yaml,
2044 None,
2045 200,
2046 r_header_yaml,
2047 "yaml",
2048 )
2049 engine.test(
2050 "Show VIM",
2051 "GET",
2052 "/admin/v1/vim_accounts/{}".format(vim_id),
2053 headers_yaml,
2054 None,
2055 200,
2056 r_header_yaml,
2057 "yaml",
2058 )
2059 if not test_osm:
2060 # delete with FORCE
2061 engine.test(
2062 "Delete VIM",
2063 "DELETE",
2064 "/admin/v1/vim_accounts/{}?FORCE=True".format(vim_id),
2065 headers_yaml,
2066 None,
2067 202,
2068 None,
2069 0,
2070 )
2071 engine.test(
2072 "Check VIM is deleted",
2073 "GET",
2074 "/admin/v1/vim_accounts/{}".format(vim_id),
2075 headers_yaml,
2076 None,
2077 404,
2078 r_header_yaml,
2079 "yaml",
2080 )
2081 else:
2082 # delete and wait until is really deleted
2083 engine.test(
2084 "Delete VIM",
2085 "DELETE",
2086 "/admin/v1/vim_accounts/{}".format(vim_id),
2087 headers_yaml,
2088 None,
2089 202,
2090 None,
2091 0,
2092 )
2093 engine.wait_until_delete(
2094 "/admin/v1/vim_accounts/{}".format(vim_id), timeout
2095 )
2096
2097
2098 class TestVIMSDN(TestFakeVim):
2099 description = "Creates VIM with SDN editing SDN controllers and port_mapping"
2100
2101 def __init__(self):
2102 TestFakeVim.__init__(self)
2103 self.wim = {
2104 "schema_version": "1.0",
2105 "schema_type": "No idea",
2106 "name": "myWim",
2107 "description": "Descriptor name",
2108 "wim_type": "odl",
2109 "wim_url": "http://localhost:/wim",
2110 "user": "user",
2111 "password": "password",
2112 "config": {"config_param": 1},
2113 }
2114
2115 def run(self, engine, test_osm, manual_check, test_params=None):
2116 engine.set_test_name("VimSdn")
2117 engine.get_autorization()
2118 # Added SDN
2119 engine.test(
2120 "Create SDN",
2121 "POST",
2122 "/admin/v1/sdns",
2123 headers_json,
2124 self.sdn,
2125 (201, 202),
2126 {"Location": "/admin/v1/sdns/", "Content-Type": "application/json"},
2127 "json",
2128 )
2129 sdnc_id = engine.last_id
2130 # sleep(5)
2131 # Edit SDN
2132 engine.test(
2133 "Edit SDN",
2134 "PATCH",
2135 "/admin/v1/sdns/{}".format(sdnc_id),
2136 headers_json,
2137 {"name": "new_sdn_name"},
2138 (202, 204),
2139 None,
2140 None,
2141 )
2142 # sleep(5)
2143 # VIM with SDN
2144 self.vim["config"]["sdn-controller"] = sdnc_id
2145 self.vim["config"]["sdn-port-mapping"] = self.port_mapping
2146 engine.test(
2147 "Create VIM",
2148 "POST",
2149 "/admin/v1/vim_accounts",
2150 headers_json,
2151 self.vim,
2152 (200, 202, 201),
2153 {"Location": "/admin/v1/vim_accounts/", "Content-Type": "application/json"},
2154 "json",
2155 ),
2156
2157 vim_id = engine.last_id
2158 self.port_mapping[0]["compute_node"] = "compute node XX"
2159 engine.test(
2160 "Edit VIM change port-mapping",
2161 "PUT",
2162 "/admin/v1/vim_accounts/{}".format(vim_id),
2163 headers_json,
2164 {"config": {"sdn-port-mapping": self.port_mapping}},
2165 (202, 204),
2166 None,
2167 None,
2168 )
2169 engine.test(
2170 "Edit VIM remove port-mapping",
2171 "PUT",
2172 "/admin/v1/vim_accounts/{}".format(vim_id),
2173 headers_json,
2174 {"config": {"sdn-port-mapping": None}},
2175 (202, 204),
2176 None,
2177 None,
2178 )
2179
2180 engine.test(
2181 "Create WIM",
2182 "POST",
2183 "/admin/v1/wim_accounts",
2184 headers_json,
2185 self.wim,
2186 (200, 202, 201),
2187 {"Location": "/admin/v1/wim_accounts/", "Content-Type": "application/json"},
2188 "json",
2189 ),
2190 wim_id = engine.last_id
2191
2192 if not test_osm:
2193 # delete with FORCE
2194 engine.test(
2195 "Delete VIM remove port-mapping",
2196 "DELETE",
2197 "/admin/v1/vim_accounts/{}?FORCE=True".format(vim_id),
2198 headers_json,
2199 None,
2200 202,
2201 None,
2202 0,
2203 )
2204 engine.test(
2205 "Delete SDNC",
2206 "DELETE",
2207 "/admin/v1/sdns/{}?FORCE=True".format(sdnc_id),
2208 headers_json,
2209 None,
2210 202,
2211 None,
2212 0,
2213 )
2214
2215 engine.test(
2216 "Delete WIM",
2217 "DELETE",
2218 "/admin/v1/wim_accounts/{}?FORCE=True".format(wim_id),
2219 headers_json,
2220 None,
2221 202,
2222 None,
2223 0,
2224 )
2225 engine.test(
2226 "Check VIM is deleted",
2227 "GET",
2228 "/admin/v1/vim_accounts/{}".format(vim_id),
2229 headers_yaml,
2230 None,
2231 404,
2232 r_header_yaml,
2233 "yaml",
2234 )
2235 engine.test(
2236 "Check SDN is deleted",
2237 "GET",
2238 "/admin/v1/sdns/{}".format(sdnc_id),
2239 headers_yaml,
2240 None,
2241 404,
2242 r_header_yaml,
2243 "yaml",
2244 )
2245 engine.test(
2246 "Check WIM is deleted",
2247 "GET",
2248 "/admin/v1/wim_accounts/{}".format(wim_id),
2249 headers_yaml,
2250 None,
2251 404,
2252 r_header_yaml,
2253 "yaml",
2254 )
2255 else:
2256 if manual_check:
2257 input(
2258 "VIM, SDN, WIM has been deployed. Perform manual check and press enter to resume"
2259 )
2260 # delete and wait until is really deleted
2261 engine.test(
2262 "Delete VIM remove port-mapping",
2263 "DELETE",
2264 "/admin/v1/vim_accounts/{}".format(vim_id),
2265 headers_json,
2266 None,
2267 (202, 201, 204),
2268 None,
2269 0,
2270 )
2271 engine.test(
2272 "Delete SDN",
2273 "DELETE",
2274 "/admin/v1/sdns/{}".format(sdnc_id),
2275 headers_json,
2276 None,
2277 (202, 201, 204),
2278 None,
2279 0,
2280 )
2281 engine.test(
2282 "Delete VIM",
2283 "DELETE",
2284 "/admin/v1/wim_accounts/{}".format(wim_id),
2285 headers_json,
2286 None,
2287 (202, 201, 204),
2288 None,
2289 0,
2290 )
2291 engine.wait_until_delete(
2292 "/admin/v1/vim_accounts/{}".format(vim_id), timeout
2293 )
2294 engine.wait_until_delete("/admin/v1/sdns/{}".format(sdnc_id), timeout)
2295 engine.wait_until_delete(
2296 "/admin/v1/wim_accounts/{}".format(wim_id), timeout
2297 )
2298
2299
2300 class TestDeploy:
2301 description = "Base class for downloading descriptors from ETSI, onboard and deploy in real VIM"
2302
2303 def __init__(self):
2304 self.test_name = "DEPLOY"
2305 self.nsd_id = None
2306 self.vim_id = None
2307 self.ns_id = None
2308 self.vnfds_id = []
2309 self.descriptor_url = (
2310 "https://osm-download.etsi.org/ftp/osm-3.0-three/2nd-hackfest/packages/"
2311 )
2312 self.vnfd_filenames = ("cirros_vnf.tar.gz",)
2313 self.nsd_filename = "cirros_2vnf_ns.tar.gz"
2314 self.descriptor_edit = None
2315 self.uses_configuration = False
2316 self.users = {}
2317 self.passwords = {}
2318 self.commands = {}
2319 self.keys = {}
2320 self.timeout = 120
2321 self.qforce = ""
2322 self.ns_params = None
2323 self.vnfr_ip_list = {}
2324
2325 def create_descriptors(self, engine):
2326 temp_dir = os.path.dirname(os.path.abspath(__file__)) + "/temp/"
2327 if not os.path.exists(temp_dir):
2328 os.makedirs(temp_dir)
2329 for vnfd_index, vnfd_filename in enumerate(self.vnfd_filenames):
2330 if "/" in vnfd_filename:
2331 vnfd_filename_path = vnfd_filename
2332 if not os.path.exists(vnfd_filename_path):
2333 raise TestException(
2334 "File '{}' does not exist".format(vnfd_filename_path)
2335 )
2336 else:
2337 vnfd_filename_path = temp_dir + vnfd_filename
2338 if not os.path.exists(vnfd_filename_path):
2339 with open(vnfd_filename_path, "wb") as file:
2340 response = requests.get(self.descriptor_url + vnfd_filename)
2341 if response.status_code >= 300:
2342 raise TestException(
2343 "Error downloading descriptor from '{}': {}".format(
2344 self.descriptor_url + vnfd_filename,
2345 response.status_code,
2346 )
2347 )
2348 file.write(response.content)
2349 if vnfd_filename_path.endswith(".yaml"):
2350 headers = headers_yaml
2351 else:
2352 headers = headers_zip_yaml
2353 if randint(0, 1) == 0:
2354 # vnfd CREATE AND UPLOAD in one step:
2355 engine.test(
2356 "Onboard VNFD in one step",
2357 "POST",
2358 "/vnfpkgm/v1/vnf_packages_content" + self.qforce,
2359 headers,
2360 "@b" + vnfd_filename_path,
2361 201,
2362 r_headers_yaml_location_vnfd,
2363 "yaml",
2364 )
2365 self.vnfds_id.append(engine.last_id)
2366 else:
2367 # vnfd CREATE AND UPLOAD ZIP
2368 engine.test(
2369 "Onboard VNFD step 1",
2370 "POST",
2371 "/vnfpkgm/v1/vnf_packages",
2372 headers_json,
2373 None,
2374 201,
2375 {
2376 "Location": "/vnfpkgm/v1/vnf_packages/",
2377 "Content-Type": "application/json",
2378 },
2379 "json",
2380 )
2381 self.vnfds_id.append(engine.last_id)
2382 engine.test(
2383 "Onboard VNFD step 2 as ZIP",
2384 "PUT",
2385 "/vnfpkgm/v1/vnf_packages/<>/package_content" + self.qforce,
2386 headers,
2387 "@b" + vnfd_filename_path,
2388 204,
2389 None,
2390 0,
2391 )
2392
2393 if self.descriptor_edit:
2394 if "vnfd{}".format(vnfd_index) in self.descriptor_edit:
2395 # Modify VNFD
2396 engine.test(
2397 "Edit VNFD ",
2398 "PATCH",
2399 "/vnfpkgm/v1/vnf_packages/{}".format(self.vnfds_id[-1]),
2400 headers_yaml,
2401 self.descriptor_edit["vnfd{}".format(vnfd_index)],
2402 204,
2403 None,
2404 None,
2405 )
2406
2407 if "/" in self.nsd_filename:
2408 nsd_filename_path = self.nsd_filename
2409 if not os.path.exists(nsd_filename_path):
2410 raise TestException(
2411 "File '{}' does not exist".format(nsd_filename_path)
2412 )
2413 else:
2414 nsd_filename_path = temp_dir + self.nsd_filename
2415 if not os.path.exists(nsd_filename_path):
2416 with open(nsd_filename_path, "wb") as file:
2417 response = requests.get(self.descriptor_url + self.nsd_filename)
2418 if response.status_code >= 300:
2419 raise TestException(
2420 "Error downloading descriptor from '{}': {}".format(
2421 self.descriptor_url + self.nsd_filename,
2422 response.status_code,
2423 )
2424 )
2425 file.write(response.content)
2426 if nsd_filename_path.endswith(".yaml"):
2427 headers = headers_yaml
2428 else:
2429 headers = headers_zip_yaml
2430
2431 if randint(0, 1) == 0:
2432 # nsd CREATE AND UPLOAD in one step:
2433 engine.test(
2434 "Onboard NSD in one step",
2435 "POST",
2436 "/nsd/v1/ns_descriptors_content" + self.qforce,
2437 headers,
2438 "@b" + nsd_filename_path,
2439 201,
2440 r_headers_yaml_location_nsd,
2441 yaml,
2442 )
2443 self.nsd_id = engine.last_id
2444 else:
2445 # nsd CREATE AND UPLOAD ZIP
2446 engine.test(
2447 "Onboard NSD step 1",
2448 "POST",
2449 "/nsd/v1/ns_descriptors",
2450 headers_json,
2451 None,
2452 201,
2453 {
2454 "Location": "/nsd/v1/ns_descriptors/",
2455 "Content-Type": "application/json",
2456 },
2457 "json",
2458 )
2459 self.nsd_id = engine.last_id
2460 engine.test(
2461 "Onboard NSD step 2 as ZIP",
2462 "PUT",
2463 "/nsd/v1/ns_descriptors/<>/nsd_content" + self.qforce,
2464 headers,
2465 "@b" + nsd_filename_path,
2466 204,
2467 None,
2468 0,
2469 )
2470
2471 if self.descriptor_edit and "nsd" in self.descriptor_edit:
2472 # Modify NSD
2473 engine.test(
2474 "Edit NSD ",
2475 "PATCH",
2476 "/nsd/v1/ns_descriptors/{}".format(self.nsd_id),
2477 headers_yaml,
2478 self.descriptor_edit["nsd"],
2479 204,
2480 None,
2481 None,
2482 )
2483
2484 def delete_descriptors(self, engine):
2485 # delete descriptors
2486 engine.test(
2487 "Delete NSSD SOL005",
2488 "DELETE",
2489 "/nsd/v1/ns_descriptors/{}".format(self.nsd_id),
2490 headers_yaml,
2491 None,
2492 204,
2493 None,
2494 0,
2495 )
2496 for vnfd_id in self.vnfds_id:
2497 engine.test(
2498 "Delete VNFD SOL005",
2499 "DELETE",
2500 "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_id),
2501 headers_yaml,
2502 None,
2503 204,
2504 None,
2505 0,
2506 )
2507
2508 def instantiate(self, engine, ns_data):
2509 ns_data_text = yaml.safe_dump(ns_data, default_flow_style=True, width=256)
2510 # create NS Two steps
2511 r = engine.test(
2512 "Create NS step 1",
2513 "POST",
2514 "/nslcm/v1/ns_instances",
2515 headers_yaml,
2516 ns_data_text,
2517 (201, 202),
2518 {"Location": "nslcm/v1/ns_instances/", "Content-Type": "application/yaml"},
2519 "yaml",
2520 )
2521 if not r:
2522 return
2523 self.ns_id = engine.last_id
2524 engine.test(
2525 "Instantiate NS step 2",
2526 "POST",
2527 "/nslcm/v1/ns_instances/{}/instantiate".format(self.ns_id),
2528 headers_yaml,
2529 ns_data_text,
2530 (201, 202),
2531 r_headers_yaml_location_nslcmop,
2532 "yaml",
2533 )
2534 nslcmop_id = engine.last_id
2535
2536 if test_osm:
2537 # Wait until status is Ok
2538 timeout = timeout_configure if self.uses_configuration else timeout_deploy
2539 engine.wait_operation_ready("ns", nslcmop_id, timeout)
2540
2541 def terminate(self, engine):
2542 # remove deployment
2543 if test_osm:
2544 engine.test(
2545 "Terminate NS",
2546 "POST",
2547 "/nslcm/v1/ns_instances/{}/terminate".format(self.ns_id),
2548 headers_yaml,
2549 None,
2550 (201, 202),
2551 r_headers_yaml_location_nslcmop,
2552 "yaml",
2553 )
2554 nslcmop2_id = engine.last_id
2555 # Wait until status is Ok
2556 engine.wait_operation_ready("ns", nslcmop2_id, timeout_deploy)
2557
2558 engine.test(
2559 "Delete NS",
2560 "DELETE",
2561 "/nslcm/v1/ns_instances/{}".format(self.ns_id),
2562 headers_yaml,
2563 None,
2564 204,
2565 None,
2566 0,
2567 )
2568 else:
2569 engine.test(
2570 "Delete NS with FORCE",
2571 "DELETE",
2572 "/nslcm/v1/ns_instances/{}?FORCE=True".format(self.ns_id),
2573 headers_yaml,
2574 None,
2575 204,
2576 None,
2577 0,
2578 )
2579
2580 # check all it is deleted
2581 engine.test(
2582 "Check NS is deleted",
2583 "GET",
2584 "/nslcm/v1/ns_instances/{}".format(self.ns_id),
2585 headers_yaml,
2586 None,
2587 404,
2588 None,
2589 "yaml",
2590 )
2591 r = engine.test(
2592 "Check NSLCMOPs are deleted",
2593 "GET",
2594 "/nslcm/v1/ns_lcm_op_occs?nsInstanceId={}".format(self.ns_id),
2595 headers_json,
2596 None,
2597 200,
2598 None,
2599 "json",
2600 )
2601 if not r:
2602 return
2603 nslcmops = r.json()
2604 if not isinstance(nslcmops, list) or nslcmops:
2605 raise TestException(
2606 "NS {} deleted but with ns_lcm_op_occ active: {}".format(
2607 self.ns_id, nslcmops
2608 )
2609 )
2610
2611 def test_ns(
2612 self,
2613 engine,
2614 test_osm,
2615 commands=None,
2616 users=None,
2617 passwds=None,
2618 keys=None,
2619 timeout=0,
2620 ):
2621
2622 r = engine.test(
2623 "GET VNFR IDs",
2624 "GET",
2625 "/nslcm/v1/ns_instances/{}".format(self.ns_id),
2626 headers_json,
2627 None,
2628 200,
2629 r_header_json,
2630 "json",
2631 )
2632 if not r:
2633 return
2634 ns_data = r.json()
2635
2636 vnfr_list = ns_data["constituent-vnfr-ref"]
2637 time = 0
2638 _commands = commands if commands is not None else self.commands
2639 _users = users if users is not None else self.users
2640 _passwds = passwds if passwds is not None else self.passwords
2641 _keys = keys if keys is not None else self.keys
2642 _timeout = timeout if timeout != 0 else self.timeout
2643
2644 # vnfr_list=[d8272263-6bd3-4680-84ca-6a4be23b3f2d, 88b22e2f-994a-4b61-94fd-4a3c90de3dc4]
2645 for vnfr_id in vnfr_list:
2646 r = engine.test(
2647 "Get VNFR to get IP_ADDRESS",
2648 "GET",
2649 "/nslcm/v1/vnfrs/{}".format(vnfr_id),
2650 headers_json,
2651 None,
2652 200,
2653 r_header_json,
2654 "json",
2655 )
2656 if not r:
2657 continue
2658 vnfr_data = r.json()
2659
2660 vnf_index = str(vnfr_data["member-vnf-index-ref"])
2661
2662 ip_address = self.get_vnfr_ip(engine, vnf_index)
2663 description = "Exec command='{}' at VNFR={} IP={}".format(
2664 _commands.get(vnf_index)[0], vnf_index, ip_address
2665 )
2666 engine.step += 1
2667 test_description = "{}{} {}".format(
2668 engine.test_name, engine.step, description
2669 )
2670 logger.warning(test_description)
2671 while _timeout >= time:
2672 result, message = self.do_checks(
2673 [ip_address],
2674 vnf_index=vnfr_data["member-vnf-index-ref"],
2675 commands=_commands.get(vnf_index),
2676 user=_users.get(vnf_index),
2677 passwd=_passwds.get(vnf_index),
2678 key=_keys.get(vnf_index),
2679 )
2680 if result == 1:
2681 engine.passed_tests += 1
2682 logger.debug(message)
2683 break
2684 elif result == 0:
2685 time += 20
2686 sleep(20)
2687 elif result == -1:
2688 engine.failed_tests += 1
2689 logger.error(message)
2690 break
2691 else:
2692 time -= 20
2693 engine.failed_tests += 1
2694 logger.error(message)
2695 else:
2696 engine.failed_tests += 1
2697 logger.error(
2698 "VNFR {} has not mgmt address. Check failed".format(vnf_index)
2699 )
2700
2701 def do_checks(self, ip, vnf_index, commands=[], user=None, passwd=None, key=None):
2702 try:
2703 import urllib3
2704 from pssh.clients import ParallelSSHClient
2705 from pssh.utils import load_private_key
2706 from ssh2 import exceptions as ssh2Exception
2707 except ImportError as e:
2708 logger.critical(
2709 "Package <pssh> or/and <urllib3> is not installed. Please add them with 'pip3 install "
2710 "parallel-ssh urllib3': {}".format(e)
2711 )
2712 return -1, "install needed packages 'pip3 install parallel-ssh urllib3'"
2713 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
2714 try:
2715 p_host = os.environ.get("PROXY_HOST")
2716 p_user = os.environ.get("PROXY_USER")
2717 p_password = os.environ.get("PROXY_PASSWD")
2718
2719 if key:
2720 pkey = load_private_key(key)
2721 else:
2722 pkey = None
2723
2724 client = ParallelSSHClient(
2725 ip,
2726 user=user,
2727 password=passwd,
2728 pkey=pkey,
2729 proxy_host=p_host,
2730 proxy_user=p_user,
2731 proxy_password=p_password,
2732 timeout=10,
2733 num_retries=0,
2734 )
2735 for cmd in commands:
2736 output = client.run_command(cmd)
2737 client.join(output)
2738 if output[ip[0]].exit_code:
2739 return -1, "VNFR {} command '{}' returns error: '{}'".format(
2740 ip[0], cmd, "\n".join(output[ip[0]].stderr)
2741 )
2742 else:
2743 return 1, "VNFR {} command '{}' successful".format(ip[0], cmd)
2744 except (
2745 ssh2Exception.ChannelFailure,
2746 ssh2Exception.SocketDisconnectError,
2747 ssh2Exception.SocketTimeout,
2748 ssh2Exception.SocketRecvError,
2749 ) as e:
2750 return 0, "Timeout accessing the VNFR {}: {}".format(ip[0], str(e))
2751 except Exception as e:
2752 return -1, "ERROR checking the VNFR {}: {}".format(ip[0], str(e))
2753
2754 def additional_operations(self, engine, test_osm, manual_check):
2755 pass
2756
2757 def run(self, engine, test_osm, manual_check, test_params=None):
2758 engine.set_test_name(self.test_name)
2759 engine.get_autorization()
2760 nsname = os.environ.get("OSMNBITEST_NS_NAME", "OSMNBITEST")
2761 if test_params:
2762 if "vnfd-files" in test_params:
2763 self.vnfd_filenames = test_params["vnfd-files"].split(",")
2764 if "nsd-file" in test_params:
2765 self.nsd_filename = test_params["nsd-file"]
2766 if test_params.get("ns-name"):
2767 nsname = test_params["ns-name"]
2768 self.create_descriptors(engine)
2769
2770 # create real VIM if not exist
2771 self.vim_id = engine.get_create_vim(test_osm)
2772 ns_data = {
2773 "nsDescription": "default description",
2774 "nsName": nsname,
2775 "nsdId": self.nsd_id,
2776 "vimAccountId": self.vim_id,
2777 }
2778 if self.ns_params:
2779 ns_data.update(self.ns_params)
2780 if test_params and test_params.get("ns-config"):
2781 if isinstance(test_params["ns-config"], str):
2782 ns_data.update(yaml.load(test_params["ns-config"]), Loader=yaml.Loader)
2783 else:
2784 ns_data.update(test_params["ns-config"])
2785 self.instantiate(engine, ns_data)
2786
2787 if manual_check:
2788 input(
2789 "NS has been deployed. Perform manual check and press enter to resume"
2790 )
2791 if test_osm and self.commands:
2792 self.test_ns(engine, test_osm)
2793 self.additional_operations(engine, test_osm, manual_check)
2794 self.terminate(engine)
2795 self.delete_descriptors(engine)
2796
2797 def get_first_ip(self, ip_string):
2798 # When using a floating IP, the vnfr_data['ip-address'] contains a semicolon-separated list of IP:s.
2799 first_ip = ip_string.split(";")[0] if ip_string else ""
2800 return first_ip
2801
2802 def get_vnfr_ip(self, engine, vnfr_index_wanted):
2803 # If the IP address list has been obtained before, it has been stored in 'vnfr_ip_list'
2804 ip = self.vnfr_ip_list.get(vnfr_index_wanted, "")
2805 if ip:
2806 return self.get_first_ip(ip)
2807 r = engine.test(
2808 "Get VNFR to get IP_ADDRESS",
2809 "GET",
2810 "/nslcm/v1/vnfrs?member-vnf-index-ref={}&nsr-id-ref={}".format(
2811 vnfr_index_wanted, self.ns_id
2812 ),
2813 headers_json,
2814 None,
2815 200,
2816 r_header_json,
2817 "json",
2818 )
2819 if not r:
2820 return ""
2821 vnfr_data = r.json()
2822 if not (vnfr_data and vnfr_data[0]):
2823 return ""
2824 # Store the IP (or list of IPs) in 'vnfr_ip_list'
2825 ip_list = vnfr_data[0].get("ip-address", "")
2826 if ip_list:
2827 self.vnfr_ip_list[vnfr_index_wanted] = ip_list
2828 ip = self.get_first_ip(ip_list)
2829 return ip
2830
2831
2832 class TestDeployHackfestCirros(TestDeploy):
2833 description = "Load and deploy Hackfest cirros_2vnf_ns example"
2834
2835 def __init__(self):
2836 super().__init__()
2837 self.test_name = "CIRROS"
2838 self.vnfd_filenames = ("cirros_vnf.tar.gz",)
2839 self.nsd_filename = "cirros_2vnf_ns.tar.gz"
2840 self.commands = {
2841 "1": [
2842 "ls -lrt",
2843 ],
2844 "2": [
2845 "ls -lrt",
2846 ],
2847 }
2848 self.users = {"1": "cirros", "2": "cirros"}
2849 self.passwords = {"1": "cubswin:)", "2": "cubswin:)"}
2850
2851 def terminate(self, engine):
2852 # Make a delete in one step, overriding the normal two step of TestDeploy that launched terminate and delete
2853 if test_osm:
2854 engine.test(
2855 "Terminate and delete NS in one step",
2856 "DELETE",
2857 "/nslcm/v1/ns_instances_content/{}".format(self.ns_id),
2858 headers_yaml,
2859 None,
2860 202,
2861 None,
2862 "yaml",
2863 )
2864
2865 engine.wait_until_delete(
2866 "/nslcm/v1/ns_instances/{}".format(self.ns_id), timeout_deploy
2867 )
2868 else:
2869 engine.test(
2870 "Delete NS with FORCE",
2871 "DELETE",
2872 "/nslcm/v1/ns_instances/{}?FORCE=True".format(self.ns_id),
2873 headers_yaml,
2874 None,
2875 204,
2876 None,
2877 0,
2878 )
2879
2880 # check all it is deleted
2881 engine.test(
2882 "Check NS is deleted",
2883 "GET",
2884 "/nslcm/v1/ns_instances/{}".format(self.ns_id),
2885 headers_yaml,
2886 None,
2887 404,
2888 None,
2889 "yaml",
2890 )
2891 r = engine.test(
2892 "Check NSLCMOPs are deleted",
2893 "GET",
2894 "/nslcm/v1/ns_lcm_op_occs?nsInstanceId={}".format(self.ns_id),
2895 headers_json,
2896 None,
2897 200,
2898 None,
2899 "json",
2900 )
2901 if not r:
2902 return
2903 nslcmops = r.json()
2904 if not isinstance(nslcmops, list) or nslcmops:
2905 raise TestException(
2906 "NS {} deleted but with ns_lcm_op_occ active: {}".format(
2907 self.ns_id, nslcmops
2908 )
2909 )
2910
2911
2912 class TestDeployHackfest1(TestDeploy):
2913 description = "Load and deploy Hackfest_1_vnfd example"
2914
2915 def __init__(self):
2916 super().__init__()
2917 self.test_name = "HACKFEST1-"
2918 self.vnfd_filenames = ("hackfest_1_vnfd.tar.gz",)
2919 self.nsd_filename = "hackfest_1_nsd.tar.gz"
2920 # self.commands = {'1': ['ls -lrt', ], '2': ['ls -lrt', ]}
2921 # self.users = {'1': "cirros", '2': "cirros"}
2922 # self.passwords = {'1': "cubswin:)", '2': "cubswin:)"}
2923
2924
2925 class TestDeployHackfestCirrosScaling(TestDeploy):
2926 description = (
2927 "Load and deploy Hackfest cirros_2vnf_ns example with scaling modifications"
2928 )
2929
2930 def __init__(self):
2931 super().__init__()
2932 self.test_name = "CIRROS-SCALE"
2933 self.vnfd_filenames = ("cirros_vnf.tar.gz",)
2934 self.nsd_filename = "cirros_2vnf_ns.tar.gz"
2935 # Modify VNFD to add scaling and count=2
2936 self.descriptor_edit = {
2937 "vnfd0": {
2938 "vdu": {"$id: 'cirros_vnfd-VM'": {"count": 2}},
2939 "scaling-group-descriptor": [
2940 {
2941 "name": "scale_cirros",
2942 "max-instance-count": 2,
2943 "vdu": [{"vdu-id-ref": "cirros_vnfd-VM", "count": 2}],
2944 }
2945 ],
2946 }
2947 }
2948
2949 def additional_operations(self, engine, test_osm, manual_check):
2950 if not test_osm:
2951 return
2952 # 2 perform scale out twice
2953 payload = (
2954 "{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_OUT, scaleByStepData: "
2955 '{scaling-group-descriptor: scale_cirros, member-vnf-index: "1"}}}'
2956 )
2957 for i in range(0, 2):
2958 engine.test(
2959 "Execute scale action over NS",
2960 "POST",
2961 "/nslcm/v1/ns_instances/{}/scale".format(self.ns_id),
2962 headers_yaml,
2963 payload,
2964 (201, 202),
2965 r_headers_yaml_location_nslcmop,
2966 "yaml",
2967 )
2968 nslcmop2_scale_out = engine.last_id
2969 engine.wait_operation_ready("ns", nslcmop2_scale_out, timeout_deploy)
2970 if manual_check:
2971 input("NS scale out done. Check that two more vdus are there")
2972 # TODO check automatic
2973
2974 # 2 perform scale in
2975 payload = (
2976 "{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_IN, scaleByStepData: "
2977 '{scaling-group-descriptor: scale_cirros, member-vnf-index: "1"}}}'
2978 )
2979 for i in range(0, 2):
2980 engine.test(
2981 "Execute scale IN action over NS",
2982 "POST",
2983 "/nslcm/v1/ns_instances/{}/scale".format(self.ns_id),
2984 headers_yaml,
2985 payload,
2986 (201, 202),
2987 r_headers_yaml_location_nslcmop,
2988 "yaml",
2989 )
2990 nslcmop2_scale_in = engine.last_id
2991 engine.wait_operation_ready("ns", nslcmop2_scale_in, timeout_deploy)
2992 if manual_check:
2993 input("NS scale in done. Check that two less vdus are there")
2994 # TODO check automatic
2995
2996 # perform scale in that must fail as reached limit
2997 engine.test(
2998 "Execute scale IN out of limit action over NS",
2999 "POST",
3000 "/nslcm/v1/ns_instances/{}/scale".format(self.ns_id),
3001 headers_yaml,
3002 payload,
3003 (201, 202),
3004 r_headers_yaml_location_nslcmop,
3005 "yaml",
3006 )
3007 nslcmop2_scale_in = engine.last_id
3008 engine.wait_operation_ready(
3009 "ns", nslcmop2_scale_in, timeout_deploy, expected_fail=True
3010 )
3011
3012
3013 class TestDeployIpMac(TestDeploy):
3014 description = "Load and deploy descriptor examples setting mac, ip address at descriptor and instantiate params"
3015
3016 def __init__(self):
3017 super().__init__()
3018 self.test_name = "SetIpMac"
3019 self.vnfd_filenames = (
3020 "vnfd_2vdu_set_ip_mac2.yaml",
3021 "vnfd_2vdu_set_ip_mac.yaml",
3022 )
3023 self.nsd_filename = "scenario_2vdu_set_ip_mac.yaml"
3024 self.descriptor_url = "https://osm.etsi.org/gitweb/?p=osm/RO.git;a=blob_plain;f=test/RO_tests/v3_2vdu_set_ip_mac/"
3025 self.commands = {
3026 "1": [
3027 "ls -lrt",
3028 ],
3029 "2": [
3030 "ls -lrt",
3031 ],
3032 }
3033 self.users = {"1": "osm", "2": "osm"}
3034 self.passwords = {"1": "osm4u", "2": "osm4u"}
3035 self.timeout = 360
3036
3037 def run(self, engine, test_osm, manual_check, test_params=None):
3038 # super().run(engine, test_osm, manual_check, test_params)
3039 # run again setting IPs with instantiate parameters
3040 instantiation_params = {
3041 "vnf": [
3042 {
3043 "member-vnf-index": "1",
3044 "internal-vld": [
3045 {
3046 "name": "internal_vld1", # net_internal
3047 "ip-profile": {
3048 "ip-version": "ipv4",
3049 "subnet-address": "10.9.8.0/24",
3050 "dhcp-params": {
3051 "count": 100,
3052 "start-address": "10.9.8.100",
3053 },
3054 },
3055 "internal-connection-point": [
3056 {
3057 "id-ref": "eth2",
3058 "ip-address": "10.9.8.2",
3059 },
3060 {
3061 "id-ref": "eth3",
3062 "ip-address": "10.9.8.3",
3063 },
3064 ],
3065 },
3066 ],
3067 "vdu": [
3068 {
3069 "id": "VM1",
3070 "interface": [
3071 # {
3072 # "name": "iface11",
3073 # "floating-ip-required": True,
3074 # },
3075 {"name": "iface13", "mac-address": "52:33:44:55:66:13"},
3076 ],
3077 },
3078 {
3079 "id": "VM2",
3080 "interface": [
3081 {
3082 "name": "iface21",
3083 "ip-address": "10.31.31.22",
3084 "mac-address": "52:33:44:55:66:21",
3085 },
3086 ],
3087 },
3088 ],
3089 },
3090 ]
3091 }
3092
3093 super().run(
3094 engine,
3095 test_osm,
3096 manual_check,
3097 test_params={"ns-config": instantiation_params},
3098 )
3099
3100
3101 class TestDeployHackfest4(TestDeploy):
3102 description = "Load and deploy Hackfest 4 example."
3103
3104 def __init__(self):
3105 super().__init__()
3106 self.test_name = "HACKFEST4-"
3107 self.vnfd_filenames = ("hackfest_4_vnfd.tar.gz",)
3108 self.nsd_filename = "hackfest_4_nsd.tar.gz"
3109 self.uses_configuration = True
3110 self.commands = {
3111 "1": [
3112 "ls -lrt",
3113 ],
3114 "2": [
3115 "ls -lrt",
3116 ],
3117 }
3118 self.users = {"1": "ubuntu", "2": "ubuntu"}
3119 self.passwords = {"1": "osm4u", "2": "osm4u"}
3120 # Modify VNFD to add scaling
3121 # self.descriptor_edit = {
3122 # "vnfd0": {
3123 # 'vnf-configuration': {
3124 # 'config-primitive': [{
3125 # 'name': 'touch',
3126 # 'parameter': [{
3127 # 'name': 'filename',
3128 # 'data-type': 'STRING',
3129 # 'default-value': '/home/ubuntu/touched'
3130 # }]
3131 # }]
3132 # },
3133 # 'scaling-group-descriptor': [{
3134 # 'name': 'scale_dataVM',
3135 # 'scaling-policy': [{
3136 # 'threshold-time': 0,
3137 # 'name': 'auto_cpu_util_above_threshold',
3138 # 'scaling-type': 'automatic',
3139 # 'scaling-criteria': [{
3140 # 'name': 'cpu_util_above_threshold',
3141 # 'vnf-monitoring-param-ref': 'all_aaa_cpu_util',
3142 # 'scale-out-relational-operation': 'GE',
3143 # 'scale-in-threshold': 15,
3144 # 'scale-out-threshold': 60,
3145 # 'scale-in-relational-operation': 'LE'
3146 # }],
3147 # 'cooldown-time': 60
3148 # }],
3149 # 'max-instance-count': 10,
3150 # 'scaling-config-action': [
3151 # {'vnf-config-primitive-name-ref': 'touch',
3152 # 'trigger': 'post-scale-out'},
3153 # {'vnf-config-primitive-name-ref': 'touch',
3154 # 'trigger': 'pre-scale-in'}
3155 # ],
3156 # 'vdu': [{
3157 # 'vdu-id-ref': 'dataVM',
3158 # 'count': 1
3159 # }]
3160 # }]
3161 # }
3162 # }
3163
3164
3165 class TestDeployHackfest3Charmed(TestDeploy):
3166 description = "Load and deploy Hackfest 3charmed_ns example"
3167
3168 def __init__(self):
3169 super().__init__()
3170 self.test_name = "HACKFEST3-"
3171 self.vnfd_filenames = ("hackfest_3charmed_vnfd.tar.gz",)
3172 self.nsd_filename = "hackfest_3charmed_nsd.tar.gz"
3173 self.uses_configuration = True
3174 self.commands = {
3175 "1": ["ls -lrt /home/ubuntu/first-touch"],
3176 "2": ["ls -lrt /home/ubuntu/first-touch"],
3177 }
3178 self.users = {"1": "ubuntu", "2": "ubuntu"}
3179 self.passwords = {"1": "osm4u", "2": "osm4u"}
3180 self.descriptor_edit = {
3181 "vnfd0": yaml.safe_load(
3182 """
3183 vnf-configuration:
3184 terminate-config-primitive:
3185 - seq: '1'
3186 name: touch
3187 parameter:
3188 - name: filename
3189 value: '/home/ubuntu/last-touch1'
3190 - seq: '3'
3191 name: touch
3192 parameter:
3193 - name: filename
3194 value: '/home/ubuntu/last-touch3'
3195 - seq: '2'
3196 name: touch
3197 parameter:
3198 - name: filename
3199 value: '/home/ubuntu/last-touch2'
3200 """
3201 )
3202 }
3203
3204 def additional_operations(self, engine, test_osm, manual_check):
3205 if not test_osm:
3206 return
3207 # 1 perform action
3208 vnfr_index_selected = "2"
3209 payload = '{member_vnf_index: "2", primitive: touch, primitive_params: { filename: /home/ubuntu/OSMTESTNBI }}'
3210 engine.test(
3211 "Exec service primitive over NS",
3212 "POST",
3213 "/nslcm/v1/ns_instances/{}/action".format(self.ns_id),
3214 headers_yaml,
3215 payload,
3216 (201, 202),
3217 r_headers_yaml_location_nslcmop,
3218 "yaml",
3219 )
3220 nslcmop2_action = engine.last_id
3221 # Wait until status is Ok
3222 engine.wait_operation_ready("ns", nslcmop2_action, timeout_deploy)
3223 vnfr_ip = self.get_vnfr_ip(engine, vnfr_index_selected)
3224 if manual_check:
3225 input(
3226 "NS service primitive has been executed."
3227 "Check that file /home/ubuntu/OSMTESTNBI is present at {}".format(
3228 vnfr_ip
3229 )
3230 )
3231 if test_osm:
3232 commands = {
3233 "1": [""],
3234 "2": [
3235 "ls -lrt /home/ubuntu/OSMTESTNBI",
3236 ],
3237 }
3238 self.test_ns(engine, test_osm, commands=commands)
3239
3240 # # 2 perform scale out
3241 # payload = '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_OUT, scaleByStepData: ' \
3242 # '{scaling-group-descriptor: scale_dataVM, member-vnf-index: "1"}}}'
3243 # engine.test("Execute scale action over NS", "POST",
3244 # "/nslcm/v1/ns_instances/{}/scale".format(self.ns_id), headers_yaml, payload,
3245 # (201, 202), r_headers_yaml_location_nslcmop, "yaml")
3246 # nslcmop2_scale_out = engine.last_id
3247 # engine.wait_operation_ready("ns", nslcmop2_scale_out, timeout_deploy)
3248 # if manual_check:
3249 # input('NS scale out done. Check that file /home/ubuntu/touched is present and new VM is created')
3250 # # TODO check automatic
3251 #
3252 # # 2 perform scale in
3253 # payload = '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_IN, scaleByStepData: ' \
3254 # '{scaling-group-descriptor: scale_dataVM, member-vnf-index: "1"}}}'
3255 # engine.test("Execute scale action over NS", "POST",
3256 # "/nslcm/v1/ns_instances/{}/scale".format(self.ns_id), headers_yaml, payload,
3257 # (201, 202), r_headers_yaml_location_nslcmop, "yaml")
3258 # nslcmop2_scale_in = engine.last_id
3259 # engine.wait_operation_ready("ns", nslcmop2_scale_in, timeout_deploy)
3260 # if manual_check:
3261 # input('NS scale in done. Check that file /home/ubuntu/touched is updated and new VM is deleted')
3262 # # TODO check automatic
3263
3264
3265 class TestDeployHackfest3Charmed2(TestDeployHackfest3Charmed):
3266 description = (
3267 "Load and deploy Hackfest 3charmed_ns example modified version of descriptors to have dots in "
3268 "ids and member-vnf-index."
3269 )
3270
3271 def __init__(self):
3272 super().__init__()
3273 self.test_name = "HACKFEST3v2-"
3274 self.qforce = "?FORCE=True"
3275 self.descriptor_edit = {
3276 "vnfd0": {
3277 "vdu": {
3278 "$[0]": {
3279 "interface": {
3280 "$[0]": {"external-connection-point-ref": "pdu-mgmt"}
3281 }
3282 },
3283 "$[1]": None,
3284 },
3285 "vnf-configuration": None,
3286 "connection-point": {
3287 "$[0]": {
3288 "id": "pdu-mgmt",
3289 "name": "pdu-mgmt",
3290 "short-name": "pdu-mgmt",
3291 },
3292 "$[1]": None,
3293 },
3294 "mgmt-interface": {"cp": "pdu-mgmt"},
3295 "description": "A vnf single vdu to be used as PDU",
3296 "id": "vdu-as-pdu",
3297 "internal-vld": {
3298 "$[0]": {
3299 "id": "pdu_internal",
3300 "name": "pdu_internal",
3301 "internal-connection-point": {"$[1]": None},
3302 "short-name": "pdu_internal",
3303 "type": "ELAN",
3304 }
3305 },
3306 },
3307 # Modify NSD accordingly
3308 "nsd": {
3309 "constituent-vnfd": {
3310 "$[0]": {"vnfd-id-ref": "vdu-as-pdu"},
3311 "$[1]": None,
3312 },
3313 "description": "A nsd to deploy the vnf to act as as PDU",
3314 "id": "nsd-as-pdu",
3315 "name": "nsd-as-pdu",
3316 "short-name": "nsd-as-pdu",
3317 "vld": {
3318 "$[0]": {
3319 "id": "mgmt_pdu",
3320 "name": "mgmt_pdu",
3321 "short-name": "mgmt_pdu",
3322 "vnfd-connection-point-ref": {
3323 "$[0]": {
3324 "vnfd-connection-point-ref": "pdu-mgmt",
3325 "vnfd-id-ref": "vdu-as-pdu",
3326 },
3327 "$[1]": None,
3328 },
3329 "type": "ELAN",
3330 },
3331 "$[1]": None,
3332 },
3333 },
3334 }
3335
3336
3337 class TestDeployHackfest3Charmed3(TestDeployHackfest3Charmed):
3338 description = "Load and deploy Hackfest 3charmed_ns example modified version to test scaling and NS parameters"
3339
3340 def __init__(self):
3341 super().__init__()
3342 self.test_name = "HACKFEST3v3-"
3343 self.commands = {
3344 "1": ["ls -lrt /home/ubuntu/first-touch-1"],
3345 "2": ["ls -lrt /home/ubuntu/first-touch-2"],
3346 }
3347 self.descriptor_edit = {
3348 "vnfd0": yaml.load(
3349 """
3350 scaling-group-descriptor:
3351 - name: "scale_dataVM"
3352 max-instance-count: 10
3353 scaling-policy:
3354 - name: "auto_cpu_util_above_threshold"
3355 scaling-type: "automatic"
3356 threshold-time: 0
3357 cooldown-time: 60
3358 scaling-criteria:
3359 - name: "cpu_util_above_threshold"
3360 scale-in-threshold: 15
3361 scale-in-relational-operation: "LE"
3362 scale-out-threshold: 60
3363 scale-out-relational-operation: "GE"
3364 vnf-monitoring-param-ref: "monitor1"
3365 vdu:
3366 - vdu-id-ref: dataVM
3367 count: 1
3368 scaling-config-action:
3369 - trigger: post-scale-out
3370 vnf-config-primitive-name-ref: touch
3371 - trigger: pre-scale-in
3372 vnf-config-primitive-name-ref: touch
3373 vdu:
3374 "$id: dataVM":
3375 monitoring-param:
3376 - id: "dataVM_cpu_util"
3377 nfvi-metric: "cpu_utilization"
3378
3379 monitoring-param:
3380 - id: "monitor1"
3381 name: "monitor1"
3382 aggregation-type: AVERAGE
3383 vdu-monitoring-param:
3384 vdu-ref: "dataVM"
3385 vdu-monitoring-param-ref: "dataVM_cpu_util"
3386 vnf-configuration:
3387 initial-config-primitive:
3388 "$[1]":
3389 parameter:
3390 "$[0]":
3391 value: "<touch_filename>" # default-value: /home/ubuntu/first-touch
3392 config-primitive:
3393 "$[0]":
3394 parameter:
3395 "$[0]":
3396 default-value: "<touch_filename2>"
3397 """,
3398 Loader=yaml.Loader,
3399 )
3400 }
3401 self.ns_params = {
3402 "additionalParamsForVnf": [
3403 {
3404 "member-vnf-index": "1",
3405 "additionalParams": {
3406 "touch_filename": "/home/ubuntu/first-touch-1",
3407 "touch_filename2": "/home/ubuntu/second-touch-1",
3408 },
3409 },
3410 {
3411 "member-vnf-index": "2",
3412 "additionalParams": {
3413 "touch_filename": "/home/ubuntu/first-touch-2",
3414 "touch_filename2": "/home/ubuntu/second-touch-2",
3415 },
3416 },
3417 ]
3418 }
3419
3420 def additional_operations(self, engine, test_osm, manual_check):
3421 super().additional_operations(engine, test_osm, manual_check)
3422 if not test_osm:
3423 return
3424
3425 # 2 perform scale out
3426 payload = (
3427 "{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_OUT, scaleByStepData: "
3428 '{scaling-group-descriptor: scale_dataVM, member-vnf-index: "1"}}}'
3429 )
3430 engine.test(
3431 "Execute scale action over NS",
3432 "POST",
3433 "/nslcm/v1/ns_instances/{}/scale".format(self.ns_id),
3434 headers_yaml,
3435 payload,
3436 (201, 202),
3437 r_headers_yaml_location_nslcmop,
3438 "yaml",
3439 )
3440 nslcmop2_scale_out = engine.last_id
3441 engine.wait_operation_ready("ns", nslcmop2_scale_out, timeout_deploy)
3442 if manual_check:
3443 input(
3444 "NS scale out done. Check that file /home/ubuntu/second-touch-1 is present and new VM is created"
3445 )
3446 if test_osm:
3447 commands = {
3448 "1": [
3449 "ls -lrt /home/ubuntu/second-touch-1",
3450 ]
3451 }
3452 self.test_ns(engine, test_osm, commands=commands)
3453 # TODO check automatic connection to scaled VM
3454
3455 # 2 perform scale in
3456 payload = (
3457 "{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_IN, scaleByStepData: "
3458 '{scaling-group-descriptor: scale_dataVM, member-vnf-index: "1"}}}'
3459 )
3460 engine.test(
3461 "Execute scale action over NS",
3462 "POST",
3463 "/nslcm/v1/ns_instances/{}/scale".format(self.ns_id),
3464 headers_yaml,
3465 payload,
3466 (201, 202),
3467 r_headers_yaml_location_nslcmop,
3468 "yaml",
3469 )
3470 nslcmop2_scale_in = engine.last_id
3471 engine.wait_operation_ready("ns", nslcmop2_scale_in, timeout_deploy)
3472 if manual_check:
3473 input(
3474 "NS scale in done. Check that file /home/ubuntu/second-touch-1 is updated and new VM is deleted"
3475 )
3476 # TODO check automatic
3477
3478
3479 class TestDeploySimpleCharm(TestDeploy):
3480 description = "Deploy hackfest-4 hackfest_simplecharm example"
3481
3482 def __init__(self):
3483 super().__init__()
3484 self.test_name = "HACKFEST-SIMPLE"
3485 self.descriptor_url = (
3486 "https://osm-download.etsi.org/ftp/osm-4.0-four/4th-hackfest/packages/"
3487 )
3488 self.vnfd_filenames = ("hackfest_simplecharm_vnf.tar.gz",)
3489 self.nsd_filename = "hackfest_simplecharm_ns.tar.gz"
3490 self.uses_configuration = True
3491 self.commands = {
3492 "1": [""],
3493 "2": [
3494 "ls -lrt /home/ubuntu/first-touch",
3495 ],
3496 }
3497 self.users = {"1": "ubuntu", "2": "ubuntu"}
3498 self.passwords = {"1": "osm4u", "2": "osm4u"}
3499
3500
3501 class TestDeploySimpleCharm2(TestDeploySimpleCharm):
3502 description = (
3503 "Deploy hackfest-4 hackfest_simplecharm example changing naming to contain dots on ids and "
3504 "vnf-member-index"
3505 )
3506
3507 def __init__(self):
3508 super().__init__()
3509 self.test_name = "HACKFEST-SIMPLE2-"
3510 self.qforce = "?FORCE=True"
3511 self.descriptor_edit = {
3512 "vnfd0": {"id": "hackfest.simplecharm.vnf"},
3513 "nsd": {
3514 "id": "hackfest.simplecharm.ns",
3515 "constituent-vnfd": {
3516 "$[0]": {
3517 "vnfd-id-ref": "hackfest.simplecharm.vnf",
3518 "member-vnf-index": "$1",
3519 },
3520 "$[1]": {
3521 "vnfd-id-ref": "hackfest.simplecharm.vnf",
3522 "member-vnf-index": "$2",
3523 },
3524 },
3525 "vld": {
3526 "$[0]": {
3527 "vnfd-connection-point-ref": {
3528 "$[0]": {
3529 "member-vnf-index-ref": "$1",
3530 "vnfd-id-ref": "hackfest.simplecharm.vnf",
3531 },
3532 "$[1]": {
3533 "member-vnf-index-ref": "$2",
3534 "vnfd-id-ref": "hackfest.simplecharm.vnf",
3535 },
3536 },
3537 },
3538 "$[1]": {
3539 "vnfd-connection-point-ref": {
3540 "$[0]": {
3541 "member-vnf-index-ref": "$1",
3542 "vnfd-id-ref": "hackfest.simplecharm.vnf",
3543 },
3544 "$[1]": {
3545 "member-vnf-index-ref": "$2",
3546 "vnfd-id-ref": "hackfest.simplecharm.vnf",
3547 },
3548 },
3549 },
3550 },
3551 },
3552 }
3553
3554
3555 class TestDeploySingleVdu(TestDeployHackfest3Charmed):
3556 description = (
3557 "Generate a single VDU base on editing Hackfest3Charmed descriptors and deploy"
3558 )
3559
3560 def __init__(self):
3561 super().__init__()
3562 self.test_name = "SingleVDU"
3563 self.qforce = "?FORCE=True"
3564 self.descriptor_edit = {
3565 # Modify VNFD to remove one VDU
3566 "vnfd0": {
3567 "vdu": {
3568 "$[0]": {
3569 "interface": {
3570 "$[0]": {"external-connection-point-ref": "pdu-mgmt"}
3571 }
3572 },
3573 "$[1]": None,
3574 },
3575 "vnf-configuration": None,
3576 "connection-point": {
3577 "$[0]": {
3578 "id": "pdu-mgmt",
3579 "name": "pdu-mgmt",
3580 "short-name": "pdu-mgmt",
3581 },
3582 "$[1]": None,
3583 },
3584 "mgmt-interface": {"cp": "pdu-mgmt"},
3585 "description": "A vnf single vdu to be used as PDU",
3586 "id": "vdu-as-pdu",
3587 "internal-vld": {
3588 "$[0]": {
3589 "id": "pdu_internal",
3590 "name": "pdu_internal",
3591 "internal-connection-point": {"$[1]": None},
3592 "short-name": "pdu_internal",
3593 "type": "ELAN",
3594 }
3595 },
3596 },
3597 # Modify NSD accordingly
3598 "nsd": {
3599 "constituent-vnfd": {
3600 "$[0]": {"vnfd-id-ref": "vdu-as-pdu"},
3601 "$[1]": None,
3602 },
3603 "description": "A nsd to deploy the vnf to act as as PDU",
3604 "id": "nsd-as-pdu",
3605 "name": "nsd-as-pdu",
3606 "short-name": "nsd-as-pdu",
3607 "vld": {
3608 "$[0]": {
3609 "id": "mgmt_pdu",
3610 "name": "mgmt_pdu",
3611 "short-name": "mgmt_pdu",
3612 "vnfd-connection-point-ref": {
3613 "$[0]": {
3614 "vnfd-connection-point-ref": "pdu-mgmt",
3615 "vnfd-id-ref": "vdu-as-pdu",
3616 },
3617 "$[1]": None,
3618 },
3619 "type": "ELAN",
3620 },
3621 "$[1]": None,
3622 },
3623 },
3624 }
3625
3626
3627 class TestDeployHnfd(TestDeployHackfest3Charmed):
3628 description = (
3629 "Generate a HNFD base on editing Hackfest3Charmed descriptors and deploy"
3630 )
3631
3632 def __init__(self):
3633 super().__init__()
3634 self.test_name = "HNFD"
3635 self.pduDeploy = TestDeploySingleVdu()
3636 self.pdu_interface_0 = {}
3637 self.pdu_interface_1 = {}
3638
3639 self.pdu_id = None
3640 # self.vnf_to_pdu = """
3641 # vdu:
3642 # "$[0]":
3643 # pdu-type: PDU-TYPE-1
3644 # interface:
3645 # "$[0]":
3646 # name: mgmt-iface
3647 # "$[1]":
3648 # name: pdu-iface-internal
3649 # id: hfn1
3650 # description: HFND, one PDU + One VDU
3651 # name: hfn1
3652 # short-name: hfn1
3653 #
3654 # """
3655
3656 self.pdu_descriptor = {
3657 "name": "my-PDU",
3658 "type": "PDU-TYPE-1",
3659 "vim_accounts": "to-override",
3660 "interfaces": [
3661 {
3662 "name": "mgmt-iface",
3663 "mgmt": True,
3664 "type": "overlay",
3665 "ip-address": "to override",
3666 "mac-address": "mac_address",
3667 "vim-network-name": "mgmt",
3668 },
3669 {
3670 "name": "pdu-iface-internal",
3671 "mgmt": False,
3672 "type": "overlay",
3673 "ip-address": "to override",
3674 "mac-address": "mac_address",
3675 "vim-network-name": "pdu_internal", # OSMNBITEST-PDU-pdu_internal
3676 },
3677 ],
3678 }
3679 self.vnfd_filenames = (
3680 "hackfest_3charmed_vnfd.tar.gz",
3681 "hackfest_3charmed_vnfd.tar.gz",
3682 )
3683
3684 self.descriptor_edit = {
3685 "vnfd0": {
3686 "id": "hfnd1",
3687 "name": "hfn1",
3688 "short-name": "hfn1",
3689 "vdu": {
3690 "$[0]": {
3691 "pdu-type": "PDU-TYPE-1",
3692 "interface": {
3693 "$[0]": {"name": "mgmt-iface"},
3694 "$[1]": {"name": "pdu-iface-internal"},
3695 },
3696 }
3697 },
3698 },
3699 "nsd": {
3700 "constituent-vnfd": {"$[1]": {"vnfd-id-ref": "hfnd1"}},
3701 "vld": {
3702 "$[0]": {
3703 "vnfd-connection-point-ref": {"$[1]": {"vnfd-id-ref": "hfnd1"}}
3704 },
3705 "$[1]": {
3706 "vnfd-connection-point-ref": {"$[1]": {"vnfd-id-ref": "hfnd1"}}
3707 },
3708 },
3709 },
3710 }
3711
3712 def create_descriptors(self, engine):
3713 super().create_descriptors(engine)
3714
3715 # Create PDU
3716 self.pdu_descriptor["interfaces"][0].update(self.pdu_interface_0)
3717 self.pdu_descriptor["interfaces"][1].update(self.pdu_interface_1)
3718 self.pdu_descriptor["vim_accounts"] = [self.vim_id]
3719 # TODO get vim-network-name from vnfr.vld.name
3720 self.pdu_descriptor["interfaces"][1]["vim-network-name"] = "{}-{}-{}".format(
3721 os.environ.get("OSMNBITEST_NS_NAME", "OSMNBITEST"),
3722 "PDU",
3723 self.pdu_descriptor["interfaces"][1]["vim-network-name"],
3724 )
3725 engine.test(
3726 "Onboard PDU descriptor",
3727 "POST",
3728 "/pdu/v1/pdu_descriptors",
3729 {
3730 "Location": "/pdu/v1/pdu_descriptors/",
3731 "Content-Type": "application/yaml",
3732 },
3733 self.pdu_descriptor,
3734 201,
3735 r_header_yaml,
3736 "yaml",
3737 )
3738 self.pdu_id = engine.last_id
3739
3740 def run(self, engine, test_osm, manual_check, test_params=None):
3741 engine.get_autorization()
3742 engine.set_test_name(self.test_name)
3743 nsname = os.environ.get("OSMNBITEST_NS_NAME", "OSMNBITEST")
3744
3745 # create real VIM if not exist
3746 self.vim_id = engine.get_create_vim(test_osm)
3747 # instantiate PDU
3748 self.pduDeploy.create_descriptors(engine)
3749 self.pduDeploy.instantiate(
3750 engine,
3751 {
3752 "nsDescription": "to be used as PDU",
3753 "nsName": nsname + "-PDU",
3754 "nsdId": self.pduDeploy.nsd_id,
3755 "vimAccountId": self.vim_id,
3756 },
3757 )
3758 if manual_check:
3759 input(
3760 "VNF to be used as PDU has been deployed. Perform manual check and press enter to resume"
3761 )
3762 if test_osm:
3763 self.pduDeploy.test_ns(engine, test_osm)
3764
3765 if test_osm:
3766 r = engine.test(
3767 "Get VNFR to obtain IP_ADDRESS",
3768 "GET",
3769 "/nslcm/v1/vnfrs?nsr-id-ref={}".format(self.pduDeploy.ns_id),
3770 headers_json,
3771 None,
3772 200,
3773 r_header_json,
3774 "json",
3775 )
3776 if not r:
3777 return
3778 vnfr_data = r.json()
3779 # print(vnfr_data)
3780
3781 self.pdu_interface_0["ip-address"] = vnfr_data[0]["vdur"][0]["interfaces"][
3782 0
3783 ].get("ip-address")
3784 self.pdu_interface_1["ip-address"] = vnfr_data[0]["vdur"][0]["interfaces"][
3785 1
3786 ].get("ip-address")
3787 self.pdu_interface_0["mac-address"] = vnfr_data[0]["vdur"][0]["interfaces"][
3788 0
3789 ].get("mac-address")
3790 self.pdu_interface_1["mac-address"] = vnfr_data[0]["vdur"][0]["interfaces"][
3791 1
3792 ].get("mac-address")
3793 if not self.pdu_interface_0["ip-address"]:
3794 raise TestException("Vnfr has not managment ip address")
3795 else:
3796 self.pdu_interface_0["ip-address"] = "192.168.10.10"
3797 self.pdu_interface_1["ip-address"] = "192.168.11.10"
3798 self.pdu_interface_0["mac-address"] = "52:33:44:55:66:13"
3799 self.pdu_interface_1["mac-address"] = "52:33:44:55:66:14"
3800
3801 self.create_descriptors(engine)
3802
3803 ns_data = {
3804 "nsDescription": "default description",
3805 "nsName": nsname,
3806 "nsdId": self.nsd_id,
3807 "vimAccountId": self.vim_id,
3808 }
3809 if test_params and test_params.get("ns-config"):
3810 if isinstance(test_params["ns-config"], str):
3811 ns_data.update(yaml.load(test_params["ns-config"]), Loader=yaml.Loader)
3812 else:
3813 ns_data.update(test_params["ns-config"])
3814
3815 self.instantiate(engine, ns_data)
3816 if manual_check:
3817 input(
3818 "NS has been deployed. Perform manual check and press enter to resume"
3819 )
3820 if test_osm:
3821 self.test_ns(engine, test_osm)
3822 self.additional_operations(engine, test_osm, manual_check)
3823 self.terminate(engine)
3824 self.pduDeploy.terminate(engine)
3825 self.delete_descriptors(engine)
3826 self.pduDeploy.delete_descriptors(engine)
3827
3828 def delete_descriptors(self, engine):
3829 super().delete_descriptors(engine)
3830 # delete pdu
3831 engine.test(
3832 "Delete PDU SOL005",
3833 "DELETE",
3834 "/pdu/v1/pdu_descriptors/{}".format(self.pdu_id),
3835 headers_yaml,
3836 None,
3837 204,
3838 None,
3839 0,
3840 )
3841
3842
3843 class TestDescriptors:
3844 description = "Test VNFD, NSD, PDU descriptors CRUD and dependencies"
3845 vnfd_empty = """vnfd:vnfd-catalog:
3846 vnfd:
3847 - name: prova
3848 short-name: prova
3849 id: prova
3850 """
3851 vnfd_prova = """vnfd:vnfd-catalog:
3852 vnfd:
3853 - connection-point:
3854 - name: cp_0h8m
3855 type: VPORT
3856 id: prova
3857 name: prova
3858 short-name: prova
3859 vdu:
3860 - id: vdu_z4bm
3861 image: ubuntu
3862 interface:
3863 - external-connection-point-ref: cp_0h8m
3864 name: eth0
3865 virtual-interface:
3866 type: VIRTIO
3867 name: vdu_z4bm
3868 version: '1.0'
3869 """
3870
3871 def __init__(self):
3872 self.vnfd_filename = "hackfest_3charmed_vnfd.tar.gz"
3873 self.nsd_filename = "hackfest_3charmed_nsd.tar.gz"
3874 self.descriptor_url = (
3875 "https://osm-download.etsi.org/ftp/osm-3.0-three/2nd-hackfest/packages/"
3876 )
3877 self.vnfd_id = None
3878 self.nsd_id = None
3879
3880 def run(self, engine, test_osm, manual_check, test_params=None):
3881 engine.set_test_name("Descriptors")
3882 engine.get_autorization()
3883 temp_dir = os.path.dirname(os.path.abspath(__file__)) + "/temp/"
3884 if not os.path.exists(temp_dir):
3885 os.makedirs(temp_dir)
3886
3887 # download files
3888 for filename in (self.vnfd_filename, self.nsd_filename):
3889 filename_path = temp_dir + filename
3890 if not os.path.exists(filename_path):
3891 with open(filename_path, "wb") as file:
3892 response = requests.get(self.descriptor_url + filename)
3893 if response.status_code >= 300:
3894 raise TestException(
3895 "Error downloading descriptor from '{}': {}".format(
3896 self.descriptor_url + filename, response.status_code
3897 )
3898 )
3899 file.write(response.content)
3900
3901 vnfd_filename_path = temp_dir + self.vnfd_filename
3902 nsd_filename_path = temp_dir + self.nsd_filename
3903
3904 engine.test(
3905 "Onboard empty VNFD in one step",
3906 "POST",
3907 "/vnfpkgm/v1/vnf_packages_content",
3908 headers_yaml,
3909 self.vnfd_empty,
3910 201,
3911 r_headers_yaml_location_vnfd,
3912 "yaml",
3913 )
3914 self.vnfd_id = engine.last_id
3915
3916 # test bug 605
3917 engine.test(
3918 "Upload invalid VNFD ",
3919 "PUT",
3920 "/vnfpkgm/v1/vnf_packages/{}/package_content".format(self.vnfd_id),
3921 headers_yaml,
3922 self.vnfd_prova,
3923 422,
3924 r_header_yaml,
3925 "yaml",
3926 )
3927
3928 engine.test(
3929 "Upload VNFD {}".format(self.vnfd_filename),
3930 "PUT",
3931 "/vnfpkgm/v1/vnf_packages/{}/package_content".format(self.vnfd_id),
3932 headers_zip_yaml,
3933 "@b" + vnfd_filename_path,
3934 204,
3935 None,
3936 0,
3937 )
3938
3939 queries = [
3940 "mgmt-interface.cp=mgmt",
3941 "vdu.0.interface.0.external-connection-point-ref=mgmt",
3942 "vdu.0.interface.1.internal-connection-point-ref=internal",
3943 "internal-vld.0.internal-connection-point.0.id-ref=internal",
3944 # Detection of duplicated VLD names in VNF Descriptors
3945 # URL: internal-vld=[
3946 # {id: internal1, name: internal, type:ELAN,
3947 # internal-connection-point: [{id-ref: mgmtVM-internal}, {id-ref: dataVM-internal}]},
3948 # {id: internal2, name: internal, type:ELAN,
3949 # internal-connection-point: [{id-ref: mgmtVM-internal}, {id-ref: dataVM-internal}]}
3950 # ]
3951 "internal-vld=%5B%7Bid%3A%20internal1%2C%20name%3A%20internal%2C%20type%3A%20ELAN%2C%20"
3952 "internal-connection-point%3A%20%5B%7Bid-ref%3A%20mgmtVM-internal%7D%2C%20%7Bid-ref%3A%20"
3953 "dataVM-internal%7D%5D%7D%2C%20%7Bid%3A%20internal2%2C%20name%3A%20internal%2C%20type%3A%20"
3954 "ELAN%2C%20internal-connection-point%3A%20%5B%7Bid-ref%3A%20mgmtVM-internal%7D%2C%20%7B"
3955 "id-ref%3A%20dataVM-internal%7D%5D%7D%5D",
3956 ]
3957 for query in queries:
3958 engine.test(
3959 "Upload invalid VNFD ",
3960 "PUT",
3961 "/vnfpkgm/v1/vnf_packages/{}/package_content?{}".format(
3962 self.vnfd_id, query
3963 ),
3964 headers_zip_yaml,
3965 "@b" + vnfd_filename_path,
3966 422,
3967 r_header_yaml,
3968 "yaml",
3969 )
3970
3971 # test bug 605
3972 engine.test(
3973 "Upload invalid VNFD ",
3974 "PUT",
3975 "/vnfpkgm/v1/vnf_packages/{}/package_content".format(self.vnfd_id),
3976 headers_yaml,
3977 self.vnfd_prova,
3978 422,
3979 r_header_yaml,
3980 "yaml",
3981 )
3982
3983 # get vnfd descriptor
3984 engine.test(
3985 "Get VNFD descriptor",
3986 "GET",
3987 "/vnfpkgm/v1/vnf_packages/{}".format(self.vnfd_id),
3988 headers_yaml,
3989 None,
3990 200,
3991 r_header_yaml,
3992 "yaml",
3993 )
3994
3995 # get vnfd file descriptor
3996 engine.test(
3997 "Get VNFD file descriptor",
3998 "GET",
3999 "/vnfpkgm/v1/vnf_packages/{}/vnfd".format(self.vnfd_id),
4000 headers_text,
4001 None,
4002 200,
4003 r_header_text,
4004 "text",
4005 temp_dir + "vnfd-yaml",
4006 )
4007 # TODO compare files: diff vnfd-yaml hackfest_3charmed_vnfd/hackfest_3charmed_vnfd.yaml
4008
4009 # get vnfd zip file package
4010 engine.test(
4011 "Get VNFD zip package",
4012 "GET",
4013 "/vnfpkgm/v1/vnf_packages/{}/package_content".format(self.vnfd_id),
4014 headers_zip,
4015 None,
4016 200,
4017 r_header_zip,
4018 "zip",
4019 temp_dir + "vnfd-zip",
4020 )
4021 # TODO compare files: diff vnfd-zip hackfest_3charmed_vnfd.tar.gz
4022
4023 # get vnfd artifact
4024 engine.test(
4025 "Get VNFD artifact package",
4026 "GET",
4027 "/vnfpkgm/v1/vnf_packages/{}/artifacts/icons/osm.png".format(self.vnfd_id),
4028 headers_zip,
4029 None,
4030 200,
4031 r_header_octect,
4032 "octet-string",
4033 temp_dir + "vnfd-icon",
4034 )
4035 # TODO compare files: diff vnfd-icon hackfest_3charmed_vnfd/icons/osm.png
4036
4037 # nsd CREATE AND UPLOAD in one step:
4038 engine.test(
4039 "Onboard NSD in one step",
4040 "POST",
4041 "/nsd/v1/ns_descriptors_content",
4042 headers_zip_yaml,
4043 "@b" + nsd_filename_path,
4044 201,
4045 r_headers_yaml_location_nsd,
4046 "yaml",
4047 )
4048 self.nsd_id = engine.last_id
4049
4050 queries = ["vld.0.vnfd-connection-point-ref.0.vnfd-id-ref=hf"]
4051 for query in queries:
4052 engine.test(
4053 "Upload invalid NSD ",
4054 "PUT",
4055 "/nsd/v1/ns_descriptors/{}/nsd_content?{}".format(self.nsd_id, query),
4056 headers_zip_yaml,
4057 "@b" + nsd_filename_path,
4058 422,
4059 r_header_yaml,
4060 "yaml",
4061 )
4062
4063 # get nsd descriptor
4064 engine.test(
4065 "Get NSD descriptor",
4066 "GET",
4067 "/nsd/v1/ns_descriptors/{}".format(self.nsd_id),
4068 headers_yaml,
4069 None,
4070 200,
4071 r_header_yaml,
4072 "yaml",
4073 )
4074
4075 # get nsd file descriptor
4076 engine.test(
4077 "Get NSD file descriptor",
4078 "GET",
4079 "/nsd/v1/ns_descriptors/{}/nsd".format(self.nsd_id),
4080 headers_text,
4081 None,
4082 200,
4083 r_header_text,
4084 "text",
4085 temp_dir + "nsd-yaml",
4086 )
4087 # TODO compare files: diff nsd-yaml hackfest_3charmed_nsd/hackfest_3charmed_nsd.yaml
4088
4089 # get nsd zip file package
4090 engine.test(
4091 "Get NSD zip package",
4092 "GET",
4093 "/nsd/v1/ns_descriptors/{}/nsd_content".format(self.nsd_id),
4094 headers_zip,
4095 None,
4096 200,
4097 r_header_zip,
4098 "zip",
4099 temp_dir + "nsd-zip",
4100 )
4101 # TODO compare files: diff nsd-zip hackfest_3charmed_nsd.tar.gz
4102
4103 # get nsd artifact
4104 engine.test(
4105 "Get NSD artifact package",
4106 "GET",
4107 "/nsd/v1/ns_descriptors/{}/artifacts/icons/osm.png".format(self.nsd_id),
4108 headers_zip,
4109 None,
4110 200,
4111 r_header_octect,
4112 "octet-string",
4113 temp_dir + "nsd-icon",
4114 )
4115 # TODO compare files: diff nsd-icon hackfest_3charmed_nsd/icons/osm.png
4116
4117 # vnfd DELETE
4118 test_rest.test(
4119 "Delete VNFD conflict",
4120 "DELETE",
4121 "/vnfpkgm/v1/vnf_packages/{}".format(self.vnfd_id),
4122 headers_yaml,
4123 None,
4124 409,
4125 None,
4126 None,
4127 )
4128
4129 test_rest.test(
4130 "Delete VNFD force",
4131 "DELETE",
4132 "/vnfpkgm/v1/vnf_packages/{}?FORCE=TRUE".format(self.vnfd_id),
4133 headers_yaml,
4134 None,
4135 204,
4136 None,
4137 0,
4138 )
4139
4140 # nsd DELETE
4141 test_rest.test(
4142 "Delete NSD",
4143 "DELETE",
4144 "/nsd/v1/ns_descriptors/{}".format(self.nsd_id),
4145 headers_yaml,
4146 None,
4147 204,
4148 None,
4149 0,
4150 )
4151
4152
4153 class TestNetSliceTemplates:
4154 description = "Upload a NST to OSM"
4155
4156 def __init__(self):
4157 self.vnfd_filename = "@./slice_shared/vnfd/slice_shared_vnfd.yaml"
4158 self.vnfd_filename_middle = "@./slice_shared/vnfd/slice_shared_middle_vnfd.yaml"
4159 self.nsd_filename = "@./slice_shared/nsd/slice_shared_nsd.yaml"
4160 self.nsd_filename_middle = "@./slice_shared/nsd/slice_shared_middle_nsd.yaml"
4161 self.nst_filenames = "@./slice_shared/slice_shared_nstd.yaml"
4162
4163 def run(self, engine, test_osm, manual_check, test_params=None):
4164 # nst CREATE
4165 engine.set_test_name("NST step ")
4166 engine.get_autorization()
4167 temp_dir = os.path.dirname(os.path.abspath(__file__)) + "/temp/"
4168 if not os.path.exists(temp_dir):
4169 os.makedirs(temp_dir)
4170
4171 # Onboard VNFDs
4172 engine.test(
4173 "Onboard edge VNFD",
4174 "POST",
4175 "/vnfpkgm/v1/vnf_packages_content",
4176 headers_yaml,
4177 self.vnfd_filename,
4178 201,
4179 r_headers_yaml_location_vnfd,
4180 "yaml",
4181 )
4182 self.vnfd_edge_id = engine.last_id
4183
4184 engine.test(
4185 "Onboard middle VNFD",
4186 "POST",
4187 "/vnfpkgm/v1/vnf_packages_content",
4188 headers_yaml,
4189 self.vnfd_filename_middle,
4190 201,
4191 r_headers_yaml_location_vnfd,
4192 "yaml",
4193 )
4194 self.vnfd_middle_id = engine.last_id
4195
4196 # Onboard NSDs
4197 engine.test(
4198 "Onboard NSD edge",
4199 "POST",
4200 "/nsd/v1/ns_descriptors_content",
4201 headers_yaml,
4202 self.nsd_filename,
4203 201,
4204 r_headers_yaml_location_nsd,
4205 "yaml",
4206 )
4207 self.nsd_edge_id = engine.last_id
4208
4209 engine.test(
4210 "Onboard NSD middle",
4211 "POST",
4212 "/nsd/v1/ns_descriptors_content",
4213 headers_yaml,
4214 self.nsd_filename_middle,
4215 201,
4216 r_headers_yaml_location_nsd,
4217 "yaml",
4218 )
4219 self.nsd_middle_id = engine.last_id
4220
4221 # Onboard NST
4222 engine.test(
4223 "Onboard NST",
4224 "POST",
4225 "/nst/v1/netslice_templates_content",
4226 headers_yaml,
4227 self.nst_filenames,
4228 201,
4229 r_headers_yaml_location_nst,
4230 "yaml",
4231 )
4232 nst_id = engine.last_id
4233
4234 # nstd SHOW OSM format
4235 engine.test(
4236 "Show NSTD OSM format",
4237 "GET",
4238 "/nst/v1/netslice_templates/{}".format(nst_id),
4239 headers_json,
4240 None,
4241 200,
4242 r_header_json,
4243 "json",
4244 )
4245
4246 # nstd DELETE
4247 engine.test(
4248 "Delete NSTD",
4249 "DELETE",
4250 "/nst/v1/netslice_templates/{}".format(nst_id),
4251 headers_json,
4252 None,
4253 204,
4254 None,
4255 0,
4256 )
4257
4258 # NSDs DELETE
4259 test_rest.test(
4260 "Delete NSD middle",
4261 "DELETE",
4262 "/nsd/v1/ns_descriptors/{}".format(self.nsd_middle_id),
4263 headers_json,
4264 None,
4265 204,
4266 None,
4267 0,
4268 )
4269
4270 test_rest.test(
4271 "Delete NSD edge",
4272 "DELETE",
4273 "/nsd/v1/ns_descriptors/{}".format(self.nsd_edge_id),
4274 headers_json,
4275 None,
4276 204,
4277 None,
4278 0,
4279 )
4280
4281 # VNFDs DELETE
4282 test_rest.test(
4283 "Delete VNFD edge",
4284 "DELETE",
4285 "/vnfpkgm/v1/vnf_packages/{}".format(self.vnfd_edge_id),
4286 headers_yaml,
4287 None,
4288 204,
4289 None,
4290 0,
4291 )
4292
4293 test_rest.test(
4294 "Delete VNFD middle",
4295 "DELETE",
4296 "/vnfpkgm/v1/vnf_packages/{}".format(self.vnfd_middle_id),
4297 headers_yaml,
4298 None,
4299 204,
4300 None,
4301 0,
4302 )
4303
4304
4305 class TestNetSliceInstances:
4306 """
4307 Test procedure:
4308 1. Populate databases with VNFD, NSD, NST with the following scenario
4309 +-----------------management-----------------+
4310 | | |
4311 +--+---+ +----+----+ +---+--+
4312 | | | | | |
4313 | edge +---data1----+ middle +---data2-----+ edge |
4314 | | | | | |
4315 +------+ +---------+ +------+
4316 shared-nss
4317 2. Create NSI-1
4318 3. Instantiate NSI-1
4319 4. Create NSI-2
4320 5. Instantiate NSI-2
4321 Manual check - Are 2 slices instantiated correctly?
4322 NSI-1 3 nss (2 nss-edges + 1 nss-middle)
4323 NSI-2 2 nss (2 nss-edge sharing nss-middle)
4324 6. Terminate NSI-1
4325 7. Delete NSI-1
4326 Manual check - Is slice NSI-1 deleted correctly?
4327 NSI-2 with 2 nss-edge + 1 nss-middle (The one from NSI-1)
4328 8. Create NSI-3
4329 9. Instantiate NSI-3
4330 Manual check - Is slice NSI-3 instantiated correctly?
4331 NSI-3 reuse nss-middle. NSI-3 only create 2 nss-edge
4332 10. Delete NSI-2
4333 11. Terminate NSI-2
4334 12. Delete NSI-3
4335 13. Terminate NSI-3
4336 Manual check - All cleaned correctly?
4337 NSI-2 and NSI-3 were terminated and deleted
4338 14. Cleanup database
4339 """
4340
4341 description = "Upload a NST to OSM"
4342
4343 def __init__(self):
4344 self.vim_id = None
4345 self.vnfd_filename = "@./slice_shared/vnfd/slice_shared_vnfd.yaml"
4346 self.vnfd_filename_middle = "@./slice_shared/vnfd/slice_shared_middle_vnfd.yaml"
4347 self.nsd_filename = "@./slice_shared/nsd/slice_shared_nsd.yaml"
4348 self.nsd_filename_middle = "@./slice_shared/nsd/slice_shared_middle_nsd.yaml"
4349 self.nst_filenames = "@./slice_shared/slice_shared_nstd.yaml"
4350
4351 def create_slice(self, engine, nsi_data, name):
4352 ns_data_text = yaml.safe_dump(nsi_data, default_flow_style=True, width=256)
4353 r = engine.test(
4354 name,
4355 "POST",
4356 "/nsilcm/v1/netslice_instances",
4357 headers_yaml,
4358 ns_data_text,
4359 (201, 202),
4360 {
4361 "Location": "nsilcm/v1/netslice_instances/",
4362 "Content-Type": "application/yaml",
4363 },
4364 "yaml",
4365 )
4366 return r
4367
4368 def instantiate_slice(self, engine, nsi_data, nsi_id, name):
4369 ns_data_text = yaml.safe_dump(nsi_data, default_flow_style=True, width=256)
4370 engine.test(
4371 name,
4372 "POST",
4373 "/nsilcm/v1/netslice_instances/{}/instantiate".format(nsi_id),
4374 headers_yaml,
4375 ns_data_text,
4376 (201, 202),
4377 r_headers_yaml_location_nsilcmop,
4378 "yaml",
4379 )
4380
4381 def terminate_slice(self, engine, nsi_id, name):
4382 engine.test(
4383 name,
4384 "POST",
4385 "/nsilcm/v1/netslice_instances/{}/terminate".format(nsi_id),
4386 headers_yaml,
4387 None,
4388 (201, 202),
4389 r_headers_yaml_location_nsilcmop,
4390 "yaml",
4391 )
4392
4393 def delete_slice(self, engine, nsi_id, name):
4394 engine.test(
4395 name,
4396 "DELETE",
4397 "/nsilcm/v1/netslice_instances/{}".format(nsi_id),
4398 headers_yaml,
4399 None,
4400 204,
4401 None,
4402 0,
4403 )
4404
4405 def run(self, engine, test_osm, manual_check, test_params=None):
4406 # nst CREATE
4407 engine.set_test_name("NSI")
4408 engine.get_autorization()
4409
4410 # Onboard VNFDs
4411 engine.test(
4412 "Onboard edge VNFD",
4413 "POST",
4414 "/vnfpkgm/v1/vnf_packages_content",
4415 headers_yaml,
4416 self.vnfd_filename,
4417 201,
4418 r_headers_yaml_location_vnfd,
4419 "yaml",
4420 )
4421 self.vnfd_edge_id = engine.last_id
4422
4423 engine.test(
4424 "Onboard middle VNFD",
4425 "POST",
4426 "/vnfpkgm/v1/vnf_packages_content",
4427 headers_yaml,
4428 self.vnfd_filename_middle,
4429 201,
4430 r_headers_yaml_location_vnfd,
4431 "yaml",
4432 )
4433 self.vnfd_middle_id = engine.last_id
4434
4435 # Onboard NSDs
4436 engine.test(
4437 "Onboard NSD edge",
4438 "POST",
4439 "/nsd/v1/ns_descriptors_content",
4440 headers_yaml,
4441 self.nsd_filename,
4442 201,
4443 r_headers_yaml_location_nsd,
4444 "yaml",
4445 )
4446 self.nsd_edge_id = engine.last_id
4447
4448 engine.test(
4449 "Onboard NSD middle",
4450 "POST",
4451 "/nsd/v1/ns_descriptors_content",
4452 headers_yaml,
4453 self.nsd_filename_middle,
4454 201,
4455 r_headers_yaml_location_nsd,
4456 "yaml",
4457 )
4458 self.nsd_middle_id = engine.last_id
4459
4460 # Onboard NST
4461 engine.test(
4462 "Onboard NST",
4463 "POST",
4464 "/nst/v1/netslice_templates_content",
4465 headers_yaml,
4466 self.nst_filenames,
4467 201,
4468 r_headers_yaml_location_nst,
4469 "yaml",
4470 )
4471 nst_id = engine.last_id
4472
4473 self.vim_id = engine.get_create_vim(test_osm)
4474
4475 # CREATE NSI-1
4476 ns_data = {
4477 "nsiName": "Deploy-NSI-1",
4478 "vimAccountId": self.vim_id,
4479 "nstId": nst_id,
4480 "nsiDescription": "default",
4481 }
4482 r = self.create_slice(engine, ns_data, "Create NSI-1 step 1")
4483 if not r:
4484 return
4485 self.nsi_id1 = engine.last_id
4486
4487 # INSTANTIATE NSI-1
4488 self.instantiate_slice(
4489 engine, ns_data, self.nsi_id1, "Instantiate NSI-1 step 2"
4490 )
4491 nsilcmop_id1 = engine.last_id
4492
4493 # Waiting for NSI-1
4494 if test_osm:
4495 engine.wait_operation_ready("nsi", nsilcmop_id1, timeout_deploy)
4496
4497 # CREATE NSI-2
4498 ns_data = {
4499 "nsiName": "Deploy-NSI-2",
4500 "vimAccountId": self.vim_id,
4501 "nstId": nst_id,
4502 "nsiDescription": "default",
4503 }
4504 r = self.create_slice(engine, ns_data, "Create NSI-2 step 1")
4505 if not r:
4506 return
4507 self.nsi_id2 = engine.last_id
4508
4509 # INSTANTIATE NSI-2
4510 self.instantiate_slice(
4511 engine, ns_data, self.nsi_id2, "Instantiate NSI-2 step 2"
4512 )
4513 nsilcmop_id2 = engine.last_id
4514
4515 # Waiting for NSI-2
4516 if test_osm:
4517 engine.wait_operation_ready("nsi", nsilcmop_id2, timeout_deploy)
4518
4519 if manual_check:
4520 input(
4521 "NSI-1 AND NSI-2 has been deployed. Perform manual check and press enter to resume"
4522 )
4523
4524 # TERMINATE NSI-1
4525 if test_osm:
4526 self.terminate_slice(engine, self.nsi_id1, "Terminate NSI-1")
4527 nsilcmop1_id = engine.last_id
4528
4529 # Wait terminate NSI-1
4530 engine.wait_operation_ready("nsi", nsilcmop1_id, timeout_deploy)
4531
4532 # DELETE NSI-1
4533 self.delete_slice(engine, self.nsi_id1, "Delete NS")
4534
4535 if manual_check:
4536 input(
4537 "NSI-1 has been deleted. Perform manual check and press enter to resume"
4538 )
4539
4540 # CREATE NSI-3
4541 ns_data = {
4542 "nsiName": "Deploy-NSI-3",
4543 "vimAccountId": self.vim_id,
4544 "nstId": nst_id,
4545 "nsiDescription": "default",
4546 }
4547 r = self.create_slice(engine, ns_data, "Create NSI-3 step 1")
4548
4549 if not r:
4550 return
4551 self.nsi_id3 = engine.last_id
4552
4553 # INSTANTIATE NSI-3
4554 self.instantiate_slice(
4555 engine, ns_data, self.nsi_id3, "Instantiate NSI-3 step 2"
4556 )
4557 nsilcmop_id3 = engine.last_id
4558
4559 # Wait Instantiate NSI-3
4560 if test_osm:
4561 engine.wait_operation_ready("nsi", nsilcmop_id3, timeout_deploy)
4562
4563 if manual_check:
4564 input(
4565 "NSI-3 has been deployed. Perform manual check and press enter to resume"
4566 )
4567
4568 # TERMINATE NSI-2
4569 if test_osm:
4570 self.terminate_slice(engine, self.nsi_id2, "Terminate NSI-2")
4571 nsilcmop2_id = engine.last_id
4572
4573 # Wait terminate NSI-2
4574 engine.wait_operation_ready("nsi", nsilcmop2_id, timeout_deploy)
4575
4576 # DELETE NSI-2
4577 self.delete_slice(engine, self.nsi_id2, "DELETE NSI-2")
4578
4579 # TERMINATE NSI-3
4580 if test_osm:
4581 self.terminate_slice(engine, self.nsi_id3, "Terminate NSI-3")
4582 nsilcmop3_id = engine.last_id
4583
4584 # Wait terminate NSI-3
4585 engine.wait_operation_ready("nsi", nsilcmop3_id, timeout_deploy)
4586
4587 # DELETE NSI-3
4588 self.delete_slice(engine, self.nsi_id3, "DELETE NSI-3")
4589
4590 if manual_check:
4591 input(
4592 "NSI-2 and NSI-3 has been deleted. Perform manual check and press enter to resume"
4593 )
4594
4595 # nstd DELETE
4596 engine.test(
4597 "Delete NSTD",
4598 "DELETE",
4599 "/nst/v1/netslice_templates/{}".format(nst_id),
4600 headers_json,
4601 None,
4602 204,
4603 None,
4604 0,
4605 )
4606
4607 # NSDs DELETE
4608 test_rest.test(
4609 "Delete NSD middle",
4610 "DELETE",
4611 "/nsd/v1/ns_descriptors/{}".format(self.nsd_middle_id),
4612 headers_json,
4613 None,
4614 204,
4615 None,
4616 0,
4617 )
4618
4619 test_rest.test(
4620 "Delete NSD edge",
4621 "DELETE",
4622 "/nsd/v1/ns_descriptors/{}".format(self.nsd_edge_id),
4623 headers_json,
4624 None,
4625 204,
4626 None,
4627 0,
4628 )
4629
4630 # VNFDs DELETE
4631 test_rest.test(
4632 "Delete VNFD edge",
4633 "DELETE",
4634 "/vnfpkgm/v1/vnf_packages/{}".format(self.vnfd_edge_id),
4635 headers_yaml,
4636 None,
4637 204,
4638 None,
4639 0,
4640 )
4641
4642 test_rest.test(
4643 "Delete VNFD middle",
4644 "DELETE",
4645 "/vnfpkgm/v1/vnf_packages/{}".format(self.vnfd_middle_id),
4646 headers_yaml,
4647 None,
4648 204,
4649 None,
4650 0,
4651 )
4652
4653
4654 class TestAuthentication:
4655 description = "Test Authentication"
4656
4657 @staticmethod
4658 def run(engine, test_osm, manual_check, test_params=None):
4659 engine.set_test_name("Authentication")
4660 # backend = test_params.get("backend") if test_params else None # UNUSED
4661
4662 admin_project_id = test_project_id = None
4663 project_admin_role_id = project_user_role_id = None
4664 test_user_id = empty_user_id = None
4665 default_role_id = empty_role_id = token_role_id = None
4666
4667 engine.get_autorization()
4668
4669 # GET
4670 engine.test(
4671 "Get tokens",
4672 "GET",
4673 "/admin/v1/tokens",
4674 headers_json,
4675 {},
4676 (200),
4677 {"Content-Type": "application/json"},
4678 "json",
4679 )
4680 engine.test(
4681 "Get projects",
4682 "GET",
4683 "/admin/v1/projects",
4684 headers_json,
4685 {},
4686 (200),
4687 {"Content-Type": "application/json"},
4688 "json",
4689 )
4690 engine.test(
4691 "Get users",
4692 "GET",
4693 "/admin/v1/users",
4694 headers_json,
4695 {},
4696 (200),
4697 {"Content-Type": "application/json"},
4698 "json",
4699 )
4700 engine.test(
4701 "Get roles",
4702 "GET",
4703 "/admin/v1/roles",
4704 headers_json,
4705 {},
4706 (200),
4707 {"Content-Type": "application/json"},
4708 "json",
4709 )
4710 res = engine.test(
4711 "Get admin project",
4712 "GET",
4713 "/admin/v1/projects?name=admin",
4714 headers_json,
4715 {},
4716 (200),
4717 {"Content-Type": "application/json"},
4718 "json",
4719 )
4720 admin_project_id = res.json()[0]["_id"] if res else None
4721 res = engine.test(
4722 "Get project admin role",
4723 "GET",
4724 "/admin/v1/roles?name=project_admin",
4725 headers_json,
4726 {},
4727 (200),
4728 {"Content-Type": "application/json"},
4729 "json",
4730 )
4731 project_admin_role_id = res.json()[0]["_id"] if res else None
4732 res = engine.test(
4733 "Get project user role",
4734 "GET",
4735 "/admin/v1/roles?name=project_user",
4736 headers_json,
4737 {},
4738 (200),
4739 {"Content-Type": "application/json"},
4740 "json",
4741 )
4742 project_user_role_id = res.json()[0]["_id"] if res else None
4743
4744 # POST
4745 res = engine.test(
4746 "Create test project",
4747 "POST",
4748 "/admin/v1/projects",
4749 headers_json,
4750 {"name": "test"},
4751 (201),
4752 {"Location": "/admin/v1/projects/", "Content-Type": "application/json"},
4753 "json",
4754 )
4755 test_project_id = engine.last_id if res else None
4756 res = engine.test(
4757 "Create role without permissions",
4758 "POST",
4759 "/admin/v1/roles",
4760 headers_json,
4761 {"name": "empty"},
4762 (201),
4763 {"Content-Type": "application/json"},
4764 "json",
4765 )
4766 empty_role_id = engine.last_id if res else None
4767 res = engine.test(
4768 "Create role with default permissions",
4769 "POST",
4770 "/admin/v1/roles",
4771 headers_json,
4772 {"name": "default", "permissions": {"default": True}},
4773 (201),
4774 {"Location": "/admin/v1/roles/", "Content-Type": "application/json"},
4775 "json",
4776 )
4777 default_role_id = engine.last_id if res else None
4778 res = engine.test(
4779 "Create role with token permissions",
4780 "POST",
4781 "/admin/v1/roles",
4782 headers_json,
4783 {
4784 "name": "tokens",
4785 "permissions": {"tokens": True},
4786 }, # is default required ?
4787 (201),
4788 {"Location": "/admin/v1/roles/", "Content-Type": "application/json"},
4789 "json",
4790 )
4791 token_role_id = engine.last_id if res else None
4792 pr = "project-role mappings"
4793 res = engine.test(
4794 "Create user without " + pr,
4795 "POST",
4796 "/admin/v1/users",
4797 headers_json,
4798 {"username": "empty", "password": "empty"},
4799 201,
4800 {"Content-Type": "application/json"},
4801 "json",
4802 )
4803 empty_user_id = engine.last_id if res else None
4804 if (
4805 admin_project_id
4806 and test_project_id
4807 and project_admin_role_id
4808 and project_user_role_id
4809 ):
4810 data = {"username": "test", "password": "test"}
4811 data["project_role_mappings"] = [
4812 {"project": test_project_id, "role": project_admin_role_id},
4813 {"project": admin_project_id, "role": project_user_role_id},
4814 ]
4815 res = engine.test(
4816 "Create user with " + pr,
4817 "POST",
4818 "/admin/v1/users",
4819 headers_json,
4820 data,
4821 (201),
4822 {"Content-Type": "application/json"},
4823 "json",
4824 )
4825 test_user_id = engine.last_id if res else None
4826
4827 # PUT
4828 if test_user_id:
4829 engine.test(
4830 "Modify test user's password",
4831 "PUT",
4832 "/admin/v1/users/" + test_user_id,
4833 headers_json,
4834 {"password": "password"},
4835 (204),
4836 {},
4837 0,
4838 )
4839 if (
4840 empty_user_id
4841 and admin_project_id
4842 and test_project_id
4843 and project_admin_role_id
4844 and project_user_role_id
4845 ):
4846 data = {
4847 "project_role_mappings": [
4848 {"project": test_project_id, "role": project_admin_role_id},
4849 {"project": admin_project_id, "role": project_user_role_id},
4850 ]
4851 }
4852 engine.test(
4853 "Modify empty user's " + pr,
4854 "PUT",
4855 "/admin/v1/users/" + empty_user_id,
4856 headers_json,
4857 data,
4858 (204),
4859 {},
4860 0,
4861 )
4862
4863 # DELETE
4864 if empty_user_id:
4865 engine.test(
4866 "Delete empty user",
4867 "DELETE",
4868 "/admin/v1/users/" + empty_user_id,
4869 headers_json,
4870 {},
4871 (204),
4872 {},
4873 0,
4874 )
4875 if test_user_id:
4876 engine.test(
4877 "Delete test user",
4878 "DELETE",
4879 "/admin/v1/users/" + test_user_id,
4880 headers_json,
4881 {},
4882 (204),
4883 {},
4884 0,
4885 )
4886 if empty_role_id:
4887 engine.test(
4888 "Delete empty role",
4889 "DELETE",
4890 "/admin/v1/roles/" + empty_role_id,
4891 headers_json,
4892 {},
4893 (204),
4894 {},
4895 0,
4896 )
4897 if default_role_id:
4898 engine.test(
4899 "Delete default role",
4900 "DELETE",
4901 "/admin/v1/roles/" + default_role_id,
4902 headers_json,
4903 {},
4904 (204),
4905 {},
4906 0,
4907 )
4908 if token_role_id:
4909 engine.test(
4910 "Delete token role",
4911 "DELETE",
4912 "/admin/v1/roles/" + token_role_id,
4913 headers_json,
4914 {},
4915 (204),
4916 {},
4917 0,
4918 )
4919 if test_project_id:
4920 engine.test(
4921 "Delete test project",
4922 "DELETE",
4923 "/admin/v1/projects/" + test_project_id,
4924 headers_json,
4925 {},
4926 (204),
4927 {},
4928 0,
4929 )
4930
4931 # END Tests
4932
4933 engine.remove_authorization() # To finish
4934
4935
4936 class TestNbiQuotas:
4937 description = "Test NBI Quotas"
4938
4939 @staticmethod
4940 def run(engine, test_osm, manual_check, test_params=None):
4941 engine.set_test_name("NBI-Quotas_")
4942 # backend = test_params.get("backend") if test_params else None # UNUSED
4943
4944 test_username = "test-nbi-quotas"
4945 test_password = "test-nbi-quotas"
4946 test_project = "test-nbi-quotas"
4947
4948 test_vim = "test-nbi-quotas"
4949 test_wim = "test-nbi-quotas"
4950 test_sdn = "test-nbi-quotas"
4951
4952 test_user_id = None
4953 test_project_id = None
4954
4955 test_vim_ids = []
4956 test_wim_ids = []
4957 test_sdn_ids = []
4958 test_vnfd_ids = []
4959 test_nsd_ids = []
4960 test_nst_ids = []
4961 test_pdu_ids = []
4962 test_nsr_ids = []
4963 test_nsi_ids = []
4964
4965 # Save admin access data
4966 admin_username = engine.user
4967 admin_password = engine.password
4968 admin_project = engine.project
4969
4970 # Get admin access
4971 engine.get_autorization()
4972 admin_token = engine.last_id
4973
4974 # Check that test project,user do not exist
4975 res1 = engine.test(
4976 "Check that test project doesn't exist",
4977 "GET",
4978 "/admin/v1/projects/" + test_project,
4979 headers_json,
4980 {},
4981 (404),
4982 {},
4983 True,
4984 )
4985 res2 = engine.test(
4986 "Check that test user doesn't exist",
4987 "GET",
4988 "/admin/v1/users/" + test_username,
4989 headers_json,
4990 {},
4991 (404),
4992 {},
4993 True,
4994 )
4995 if None in [res1, res2]:
4996 engine.remove_authorization()
4997 logger.error("Test project and/or user already exist")
4998 return
4999
5000 # Create test project&user
5001 res = engine.test(
5002 "Create test project",
5003 "POST",
5004 "/admin/v1/projects",
5005 headers_json,
5006 {
5007 "name": test_username,
5008 "quotas": {
5009 "vnfds": 2,
5010 "nsds": 2,
5011 "nsts": 1,
5012 "pdus": 1,
5013 "nsrs": 2,
5014 "nsis": 1,
5015 "vim_accounts": 1,
5016 "wim_accounts": 1,
5017 "sdns": 1,
5018 },
5019 },
5020 (201),
5021 r_header_json,
5022 "json",
5023 )
5024 test_project_id = engine.last_id if res else None
5025 res = engine.test(
5026 "Create test user",
5027 "POST",
5028 "/admin/v1/users",
5029 headers_json,
5030 {
5031 "username": test_username,
5032 "password": test_password,
5033 "project_role_mappings": [
5034 {"project": test_project, "role": "project_admin"}
5035 ],
5036 },
5037 (201),
5038 r_header_json,
5039 "json",
5040 )
5041 test_user_id = engine.last_id if res else None
5042
5043 if test_project_id and test_user_id:
5044
5045 # Get user access
5046 engine.token = None
5047 engine.user = test_username
5048 engine.password = test_password
5049 engine.project = test_project
5050 engine.get_autorization()
5051 user_token = engine.last_id
5052
5053 # Create test VIM
5054 res = engine.test(
5055 "Create test VIM",
5056 "POST",
5057 "/admin/v1/vim_accounts",
5058 headers_json,
5059 {
5060 "name": test_vim,
5061 "vim_type": "openvim",
5062 "vim_user": test_username,
5063 "vim_password": test_password,
5064 "vim_tenant_name": test_project,
5065 "vim_url": "https://0.0.0.0:0/v0.0",
5066 },
5067 (202),
5068 r_header_json,
5069 "json",
5070 )
5071 test_vim_ids += [engine.last_id if res else None]
5072
5073 res = engine.test(
5074 "Try to create second test VIM",
5075 "POST",
5076 "/admin/v1/vim_accounts",
5077 headers_json,
5078 {
5079 "name": test_vim + "_2",
5080 "vim_type": "openvim",
5081 "vim_user": test_username,
5082 "vim_password": test_password,
5083 "vim_tenant_name": test_project,
5084 "vim_url": "https://0.0.0.0:0/v0.0",
5085 },
5086 (422),
5087 r_header_json,
5088 "json",
5089 )
5090 test_vim_ids += [engine.last_id if res is None else None]
5091
5092 res = engine.test(
5093 "Try to create second test VIM with FORCE",
5094 "POST",
5095 "/admin/v1/vim_accounts?FORCE",
5096 headers_json,
5097 {
5098 "name": test_vim + "_3",
5099 "vim_type": "openvim",
5100 "vim_user": test_username,
5101 "vim_password": test_password,
5102 "vim_tenant_name": test_project,
5103 "vim_url": "https://0.0.0.0:0/v0.0",
5104 },
5105 (202),
5106 r_header_json,
5107 "json",
5108 )
5109 test_vim_ids += [engine.last_id if res else None]
5110
5111 if test_vim_ids[0]:
5112
5113 # Download descriptor files (if required)
5114 test_dir = "/tmp/" + test_username + "/"
5115 test_url = "https://osm-download.etsi.org/ftp/osm-6.0-six/7th-hackfest/packages/"
5116 vnfd_filenames = [
5117 "slice_hackfest_vnfd.tar.gz",
5118 "slice_hackfest_middle_vnfd.tar.gz",
5119 ]
5120 nsd_filenames = [
5121 "slice_hackfest_nsd.tar.gz",
5122 "slice_hackfest_middle_nsd.tar.gz",
5123 ]
5124 nst_filenames = ["slice_hackfest_nstd.yaml"]
5125 pdu_filenames = ["PDU_router.yaml"]
5126 desc_filenames = (
5127 vnfd_filenames + nsd_filenames + nst_filenames + pdu_filenames
5128 )
5129 if not os.path.exists(test_dir):
5130 os.makedirs(test_dir)
5131 for filename in desc_filenames:
5132 if not os.path.exists(test_dir + filename):
5133 res = requests.get(test_url + filename)
5134 if res.status_code < 300:
5135 with open(test_dir + filename, "wb") as file:
5136 file.write(res.content)
5137
5138 if all([os.path.exists(test_dir + p) for p in desc_filenames]):
5139
5140 # Test VNFD Quotas
5141 res = engine.test(
5142 "Create test VNFD #1",
5143 "POST",
5144 "/vnfpkgm/v1/vnf_packages_content",
5145 headers_zip_json,
5146 "@b" + test_dir + vnfd_filenames[0],
5147 (201),
5148 r_header_json,
5149 "json",
5150 )
5151 test_vnfd_ids += [engine.last_id if res else None]
5152 res = engine.test(
5153 "Create test VNFD #2",
5154 "POST",
5155 "/vnfpkgm/v1/vnf_packages_content",
5156 headers_zip_json,
5157 "@b" + test_dir + vnfd_filenames[1],
5158 (201),
5159 r_header_json,
5160 "json",
5161 )
5162 test_vnfd_ids += [engine.last_id if res else None]
5163 res = engine.test(
5164 "Try to create extra test VNFD",
5165 "POST",
5166 "/vnfpkgm/v1/vnf_packages_content",
5167 headers_zip_json,
5168 "@b" + test_dir + vnfd_filenames[0],
5169 (422),
5170 r_header_json,
5171 "json",
5172 )
5173 test_vnfd_ids += [engine.last_id if res is None else None]
5174 res = engine.test(
5175 "Try to create extra test VNFD with FORCE",
5176 "POST",
5177 "/vnfpkgm/v1/vnf_packages_content?FORCE",
5178 headers_zip_json,
5179 "@b" + test_dir + vnfd_filenames[0],
5180 (201),
5181 r_header_json,
5182 "json",
5183 )
5184 test_vnfd_ids += [engine.last_id if res else None]
5185
5186 # Remove extra VNFDs to prevent further errors
5187 for i in [2, 3]:
5188 if test_vnfd_ids[i]:
5189 res = engine.test(
5190 "Delete test VNFD #" + str(i),
5191 "DELETE",
5192 "/vnfpkgm/v1/vnf_packages_content/"
5193 + test_vnfd_ids[i]
5194 + "?FORCE",
5195 headers_json,
5196 {},
5197 (204),
5198 {},
5199 0,
5200 )
5201 if res:
5202 test_vnfd_ids[i] = None
5203
5204 if test_vnfd_ids[0] and test_vnfd_ids[1]:
5205
5206 # Test NSD Quotas
5207 res = engine.test(
5208 "Create test NSD #1",
5209 "POST",
5210 "/nsd/v1/ns_descriptors_content",
5211 headers_zip_json,
5212 "@b" + test_dir + nsd_filenames[0],
5213 (201),
5214 r_header_json,
5215 "json",
5216 )
5217 test_nsd_ids += [engine.last_id if res else None]
5218 res = engine.test(
5219 "Create test NSD #2",
5220 "POST",
5221 "/nsd/v1/ns_descriptors_content",
5222 headers_zip_json,
5223 "@b" + test_dir + nsd_filenames[1],
5224 (201),
5225 r_header_json,
5226 "json",
5227 )
5228 test_nsd_ids += [engine.last_id if res else None]
5229 res = engine.test(
5230 "Try to create extra test NSD",
5231 "POST",
5232 "/nsd/v1/ns_descriptors_content",
5233 headers_zip_json,
5234 "@b" + test_dir + nsd_filenames[0],
5235 (422),
5236 r_header_json,
5237 "json",
5238 )
5239 test_nsd_ids += [engine.last_id if res is None else None]
5240 res = engine.test(
5241 "Try to create extra test NSD with FORCE",
5242 "POST",
5243 "/nsd/v1/ns_descriptors_content?FORCE",
5244 headers_zip_json,
5245 "@b" + test_dir + nsd_filenames[0],
5246 (201),
5247 r_header_json,
5248 "json",
5249 )
5250 test_nsd_ids += [engine.last_id if res else None]
5251
5252 # Remove extra NSDs to prevent further errors
5253 for i in [2, 3]:
5254 if test_nsd_ids[i]:
5255 res = engine.test(
5256 "Delete test NSD #" + str(i),
5257 "DELETE",
5258 "/nsd/v1/ns_descriptors_content/"
5259 + test_nsd_ids[i]
5260 + "?FORCE",
5261 headers_json,
5262 {},
5263 (204),
5264 {},
5265 0,
5266 )
5267 if res:
5268 test_nsd_ids[i] = None
5269
5270 if test_nsd_ids[0] and test_nsd_ids[1]:
5271
5272 # Test NSR Quotas
5273 res = engine.test(
5274 "Create test NSR #1",
5275 "POST",
5276 "/nslcm/v1/ns_instances_content",
5277 headers_json,
5278 {
5279 "nsName": test_username + "_1",
5280 "nsdId": test_nsd_ids[0],
5281 "vimAccountId": test_vim_ids[0],
5282 },
5283 (201),
5284 r_header_json,
5285 "json",
5286 )
5287 test_nsr_ids += [engine.last_id if res else None]
5288 res = engine.test(
5289 "Create test NSR #2",
5290 "POST",
5291 "/nslcm/v1/ns_instances_content",
5292 headers_json,
5293 {
5294 "nsName": test_username + "_2",
5295 "nsdId": test_nsd_ids[1],
5296 "vimAccountId": test_vim_ids[0],
5297 },
5298 (201),
5299 r_header_json,
5300 "json",
5301 )
5302 test_nsr_ids += [engine.last_id if res else None]
5303 res = engine.test(
5304 "Try to create extra test NSR",
5305 "POST",
5306 "/nslcm/v1/ns_instances_content",
5307 headers_json,
5308 {
5309 "nsName": test_username + "_3",
5310 "nsdId": test_nsd_ids[0],
5311 "vimAccountId": test_vim_ids[0],
5312 },
5313 (422),
5314 r_header_json,
5315 "json",
5316 )
5317 test_nsr_ids += [engine.last_id if res is None else None]
5318 res = engine.test(
5319 "Try to create test NSR with FORCE",
5320 "POST",
5321 "/nslcm/v1/ns_instances_content?FORCE",
5322 headers_json,
5323 {
5324 "nsName": test_username + "_4",
5325 "nsdId": test_nsd_ids[0],
5326 "vimAccountId": test_vim_ids[0],
5327 },
5328 (201),
5329 r_header_json,
5330 "json",
5331 )
5332 test_nsr_ids += [engine.last_id if res else None]
5333
5334 # Test NST Quotas
5335 res = engine.test(
5336 "Create test NST",
5337 "POST",
5338 "/nst/v1/netslice_templates_content",
5339 headers_txt_json,
5340 "@b" + test_dir + nst_filenames[0],
5341 (201),
5342 r_header_json,
5343 "json",
5344 )
5345 test_nst_ids += [engine.last_id if res else None]
5346 res = engine.test(
5347 "Try to create extra test NST",
5348 "POST",
5349 "/nst/v1/netslice_templates_content",
5350 headers_txt_json,
5351 "@b" + test_dir + nst_filenames[0],
5352 (422),
5353 r_header_json,
5354 "json",
5355 )
5356 test_nst_ids += [engine.last_id if res is None else None]
5357 res = engine.test(
5358 "Try to create extra test NST with FORCE",
5359 "POST",
5360 "/nst/v1/netslice_templates_content?FORCE",
5361 headers_txt_json,
5362 "@b" + test_dir + nst_filenames[0],
5363 (201),
5364 r_header_json,
5365 "json",
5366 )
5367 test_nst_ids += [engine.last_id if res else None]
5368
5369 if test_nst_ids[0]:
5370 # Remove NSR Quota
5371 engine.set_header(
5372 {"Authorization": "Bearer {}".format(admin_token)}
5373 )
5374 res = engine.test(
5375 "Remove NSR Quota",
5376 "PUT",
5377 "/admin/v1/projects/" + test_project_id,
5378 headers_json,
5379 {"quotas": {"nsrs": None}},
5380 (204),
5381 {},
5382 0,
5383 )
5384 engine.set_header(
5385 {"Authorization": "Bearer {}".format(user_token)}
5386 )
5387 if res:
5388 # Test NSI Quotas
5389 res = engine.test(
5390 "Create test NSI",
5391 "POST",
5392 "/nsilcm/v1/netslice_instances_content",
5393 headers_json,
5394 {
5395 "nsiName": test_username,
5396 "nstId": test_nst_ids[0],
5397 "vimAccountId": test_vim_ids[0],
5398 },
5399 (201),
5400 r_header_json,
5401 "json",
5402 )
5403 test_nsi_ids += [engine.last_id if res else None]
5404 res = engine.test(
5405 "Try to create extra test NSI",
5406 "POST",
5407 "/nsilcm/v1/netslice_instances_content",
5408 headers_json,
5409 {
5410 "nsiName": test_username,
5411 "nstId": test_nst_ids[0],
5412 "vimAccountId": test_vim_ids[0],
5413 },
5414 (400),
5415 r_header_json,
5416 "json",
5417 )
5418 test_nsi_ids += [
5419 engine.last_id if res is None else None
5420 ]
5421 res = engine.test(
5422 "Try to create extra test NSI with FORCE",
5423 "POST",
5424 "/nsilcm/v1/netslice_instances_content?FORCE",
5425 headers_json,
5426 {
5427 "nsiName": test_username,
5428 "nstId": test_nst_ids[0],
5429 "vimAccountId": test_vim_ids[0],
5430 },
5431 (201),
5432 r_header_json,
5433 "json",
5434 )
5435 test_nsi_ids += [engine.last_id if res else None]
5436
5437 # Test PDU Quotas
5438 with open(test_dir + pdu_filenames[0], "rb") as file:
5439 pdu_text = re.sub(
5440 r"ip-address: *\[[^\]]*\]",
5441 "ip-address: '0.0.0.0'",
5442 file.read().decode("utf-8"),
5443 )
5444 with open(test_dir + pdu_filenames[0], "wb") as file:
5445 file.write(pdu_text.encode("utf-8"))
5446 res = engine.test(
5447 "Create test PDU",
5448 "POST",
5449 "/pdu/v1/pdu_descriptors",
5450 headers_yaml,
5451 "@b" + test_dir + pdu_filenames[0],
5452 (201),
5453 r_header_yaml,
5454 "yaml",
5455 )
5456 test_pdu_ids += [engine.last_id if res else None]
5457 res = engine.test(
5458 "Try to create extra test PDU",
5459 "POST",
5460 "/pdu/v1/pdu_descriptors",
5461 headers_yaml,
5462 "@b" + test_dir + pdu_filenames[0],
5463 (422),
5464 r_header_yaml,
5465 "yaml",
5466 )
5467 test_pdu_ids += [engine.last_id if res is None else None]
5468 res = engine.test(
5469 "Try to create extra test PDU with FORCE",
5470 "POST",
5471 "/pdu/v1/pdu_descriptors?FORCE",
5472 headers_yaml,
5473 "@b" + test_dir + pdu_filenames[0],
5474 (201),
5475 r_header_yaml,
5476 "yaml",
5477 )
5478 test_pdu_ids += [engine.last_id if res else None]
5479
5480 # Cleanup
5481 for i, id in enumerate(test_nsi_ids):
5482 if id:
5483 engine.test(
5484 "Delete test NSI #" + str(i),
5485 "DELETE",
5486 "/nsilcm/v1/netslice_instances_content/"
5487 + id
5488 + "?FORCE",
5489 headers_json,
5490 {},
5491 (204),
5492 {},
5493 0,
5494 )
5495 for i, id in enumerate(test_nsr_ids):
5496 if id:
5497 engine.test(
5498 "Delete test NSR #" + str(i),
5499 "DELETE",
5500 "/nslcm/v1/ns_instances_content/" + id + "?FORCE",
5501 headers_json,
5502 {},
5503 (204),
5504 {},
5505 0,
5506 )
5507 for i, id in enumerate(test_nst_ids):
5508 if id:
5509 engine.test(
5510 "Delete test NST #" + str(i),
5511 "DELETE",
5512 "/nst/v1/netslice_templates_content/" + id + "?FORCE",
5513 headers_json,
5514 {},
5515 (204),
5516 {},
5517 0,
5518 )
5519 for i, id in enumerate(test_nsd_ids):
5520 if id:
5521 engine.test(
5522 "Delete test NSD #" + str(i),
5523 "DELETE",
5524 "/nsd/v1/ns_descriptors_content/" + id + "?FORCE",
5525 headers_json,
5526 {},
5527 (204),
5528 {},
5529 0,
5530 )
5531 for i, id in enumerate(test_vnfd_ids):
5532 if id:
5533 engine.test(
5534 "Delete test VNFD #" + str(i),
5535 "DELETE",
5536 "/vnfpkgm/v1/vnf_packages_content/" + id + "?FORCE",
5537 headers_json,
5538 {},
5539 (204),
5540 {},
5541 0,
5542 )
5543 for i, id in enumerate(test_pdu_ids):
5544 if id:
5545 engine.test(
5546 "Delete test PDU #" + str(i),
5547 "DELETE",
5548 "/pdu/v1/pdu_descriptors/" + id + "?FORCE",
5549 headers_json,
5550 {},
5551 (204),
5552 {},
5553 0,
5554 )
5555
5556 # END Test NBI Quotas
5557
5558 # Test WIM Quotas
5559 res = engine.test(
5560 "Create test WIM",
5561 "POST",
5562 "/admin/v1/wim_accounts",
5563 headers_json,
5564 {
5565 "name": test_wim,
5566 "wim_type": "onos",
5567 "wim_url": "https://0.0.0.0:0/v0.0",
5568 },
5569 (202),
5570 r_header_json,
5571 "json",
5572 )
5573 test_wim_ids += [engine.last_id if res else None]
5574 res = engine.test(
5575 "Try to create second test WIM",
5576 "POST",
5577 "/admin/v1/wim_accounts",
5578 headers_json,
5579 {
5580 "name": test_wim + "_2",
5581 "wim_type": "onos",
5582 "wim_url": "https://0.0.0.0:0/v0.0",
5583 },
5584 (422),
5585 r_header_json,
5586 "json",
5587 )
5588 test_wim_ids += [engine.last_id if res is None else None]
5589 res = engine.test(
5590 "Try to create second test WIM with FORCE",
5591 "POST",
5592 "/admin/v1/wim_accounts?FORCE",
5593 headers_json,
5594 {
5595 "name": test_wim + "_3",
5596 "wim_type": "onos",
5597 "wim_url": "https://0.0.0.0:0/v0.0",
5598 },
5599 (202),
5600 r_header_json,
5601 "json",
5602 )
5603 test_wim_ids += [engine.last_id if res else None]
5604
5605 # Test SDN Quotas
5606 res = engine.test(
5607 "Create test SDN",
5608 "POST",
5609 "/admin/v1/sdns",
5610 headers_json,
5611 {
5612 "name": test_sdn,
5613 "type": "onos",
5614 "ip": "0.0.0.0",
5615 "port": 9999,
5616 "dpid": "00:00:00:00:00:00:00:00",
5617 },
5618 (202),
5619 r_header_json,
5620 "json",
5621 )
5622 test_sdn_ids += [engine.last_id if res else None]
5623 res = engine.test(
5624 "Try to create second test SDN",
5625 "POST",
5626 "/admin/v1/sdns",
5627 headers_json,
5628 {
5629 "name": test_sdn + "_2",
5630 "type": "onos",
5631 "ip": "0.0.0.0",
5632 "port": 9999,
5633 "dpid": "00:00:00:00:00:00:00:00",
5634 },
5635 (422),
5636 r_header_json,
5637 "json",
5638 )
5639 test_sdn_ids += [engine.last_id if res is None else None]
5640 res = engine.test(
5641 "Try to create second test SDN with FORCE",
5642 "POST",
5643 "/admin/v1/sdns?FORCE",
5644 headers_json,
5645 {
5646 "name": test_sdn + "_3",
5647 "type": "onos",
5648 "ip": "0.0.0.0",
5649 "port": 9999,
5650 "dpid": "00:00:00:00:00:00:00:00",
5651 },
5652 (202),
5653 r_header_json,
5654 "json",
5655 )
5656 test_sdn_ids += [engine.last_id if res else None]
5657
5658 # Cleanup
5659 for i, id in enumerate(test_vim_ids):
5660 if id:
5661 engine.test(
5662 "Delete test VIM #" + str(i),
5663 "DELETE",
5664 "/admin/v1/vim_accounts/" + id + "?FORCE",
5665 headers_json,
5666 {},
5667 (202),
5668 {},
5669 0,
5670 )
5671 for i, id in enumerate(test_wim_ids):
5672 if id:
5673 engine.test(
5674 "Delete test WIM #" + str(i),
5675 "DELETE",
5676 "/admin/v1/wim_accounts/" + id + "?FORCE",
5677 headers_json,
5678 {},
5679 (202),
5680 {},
5681 0,
5682 )
5683 for i, id in enumerate(test_sdn_ids):
5684 if id:
5685 engine.test(
5686 "Delete test SDN #" + str(i),
5687 "DELETE",
5688 "/admin/v1/sdns/" + id + "?FORCE",
5689 headers_json,
5690 {},
5691 (202),
5692 {},
5693 0,
5694 )
5695
5696 # Release user access
5697 engine.remove_authorization()
5698
5699 # Cleanup
5700 engine.user = admin_username
5701 engine.password = admin_password
5702 engine.project = admin_project
5703 engine.get_autorization()
5704 if test_user_id:
5705 engine.test(
5706 "Delete test user",
5707 "DELETE",
5708 "/admin/v1/users/" + test_user_id + "?FORCE",
5709 headers_json,
5710 {},
5711 (204),
5712 {},
5713 0,
5714 )
5715 if test_project_id:
5716 engine.test(
5717 "Delete test project",
5718 "DELETE",
5719 "/admin/v1/projects/" + test_project_id + "?FORCE",
5720 headers_json,
5721 {},
5722 (204),
5723 {},
5724 0,
5725 )
5726 engine.remove_authorization()
5727
5728 # END class TestNbiQuotas
5729
5730
5731 if __name__ == "__main__":
5732 global logger
5733 test = ""
5734
5735 # Disable warnings from self-signed certificates.
5736 requests.packages.urllib3.disable_warnings()
5737 try:
5738 logging.basicConfig(format="%(levelname)s %(message)s", level=logging.ERROR)
5739 logger = logging.getLogger("NBI")
5740 # load parameters and configuration
5741 opts, args = getopt.getopt(
5742 sys.argv[1:],
5743 "hvu:p:",
5744 [
5745 "url=",
5746 "user=",
5747 "password=",
5748 "help",
5749 "version",
5750 "verbose",
5751 "no-verbose",
5752 "project=",
5753 "insecure",
5754 "timeout",
5755 "timeout-deploy",
5756 "timeout-configure",
5757 "test=",
5758 "list",
5759 "test-osm",
5760 "manual-check",
5761 "params=",
5762 "fail-fast",
5763 ],
5764 )
5765 url = "https://localhost:9999/osm"
5766 user = password = project = "admin"
5767 test_osm = False
5768 manual_check = False
5769 verbose = 0
5770 verify = True
5771 fail_fast = False
5772 test_classes = {
5773 "NonAuthorized": TestNonAuthorized,
5774 "FakeVIM": TestFakeVim,
5775 "Users-Projects": TestUsersProjects,
5776 "Projects-Descriptors": TestProjectsDescriptors,
5777 "VIM-SDN": TestVIMSDN,
5778 "Deploy-Custom": TestDeploy,
5779 "Deploy-Hackfest-Cirros": TestDeployHackfestCirros,
5780 "Deploy-Hackfest-Cirros-Scaling": TestDeployHackfestCirrosScaling,
5781 "Deploy-Hackfest-3Charmed": TestDeployHackfest3Charmed,
5782 "Deploy-Hackfest-3Charmed2": TestDeployHackfest3Charmed2,
5783 "Deploy-Hackfest-3Charmed3": TestDeployHackfest3Charmed3,
5784 "Deploy-Hackfest-4": TestDeployHackfest4,
5785 "Deploy-CirrosMacIp": TestDeployIpMac,
5786 "Descriptors": TestDescriptors,
5787 "Deploy-Hackfest1": TestDeployHackfest1,
5788 # "Deploy-MultiVIM": TestDeployMultiVIM,
5789 "Deploy-SingleVdu": TestDeploySingleVdu,
5790 "Deploy-Hnfd": TestDeployHnfd,
5791 "Upload-Slice-Template": TestNetSliceTemplates,
5792 "Deploy-Slice-Instance": TestNetSliceInstances,
5793 "Deploy-SimpleCharm": TestDeploySimpleCharm,
5794 "Deploy-SimpleCharm2": TestDeploySimpleCharm2,
5795 "Authentication": TestAuthentication,
5796 "NBI-Quotas": TestNbiQuotas,
5797 }
5798 test_to_do = []
5799 test_params = {}
5800
5801 for o, a in opts:
5802 # print("parameter:", o, a)
5803 if o == "--version":
5804 print("test version " + __version__ + " " + version_date)
5805 exit()
5806 elif o == "--list":
5807 for test, test_class in sorted(test_classes.items()):
5808 print("{:32} {}".format(test + ":", test_class.description))
5809 exit()
5810 elif o in ("-v", "--verbose"):
5811 verbose += 1
5812 elif o == "no-verbose":
5813 verbose = -1
5814 elif o in ("-h", "--help"):
5815 usage()
5816 sys.exit()
5817 elif o == "--test-osm":
5818 test_osm = True
5819 elif o == "--manual-check":
5820 manual_check = True
5821 elif o == "--url":
5822 url = a
5823 elif o in ("-u", "--user"):
5824 user = a
5825 elif o in ("-p", "--password"):
5826 password = a
5827 elif o == "--project":
5828 project = a
5829 elif o == "--fail-fast":
5830 fail_fast = True
5831 elif o == "--test":
5832 for _test in a.split(","):
5833 if _test not in test_classes:
5834 print(
5835 "Invalid test name '{}'. Use option '--list' to show available tests".format(
5836 _test
5837 ),
5838 file=sys.stderr,
5839 )
5840 exit(1)
5841 test_to_do.append(_test)
5842 elif o == "--params":
5843 param_key, _, param_value = a.partition("=")
5844 text_index = len(test_to_do)
5845 if text_index not in test_params:
5846 test_params[text_index] = {}
5847 test_params[text_index][param_key] = param_value
5848 elif o == "--insecure":
5849 verify = False
5850 elif o == "--timeout":
5851 timeout = int(a)
5852 elif o == "--timeout-deploy":
5853 timeout_deploy = int(a)
5854 elif o == "--timeout-configure":
5855 timeout_configure = int(a)
5856 else:
5857 assert False, "Unhandled option"
5858 if verbose == 0:
5859 logger.setLevel(logging.WARNING)
5860 elif verbose > 1:
5861 logger.setLevel(logging.DEBUG)
5862 else:
5863 logger.setLevel(logging.ERROR)
5864
5865 test_rest = TestRest(url, user=user, password=password, project=project)
5866 # print("tests to do:", test_to_do)
5867 if test_to_do:
5868 text_index = 0
5869 for test in test_to_do:
5870 if fail_fast and test_rest.failed_tests:
5871 break
5872 text_index += 1
5873 test_class = test_classes[test]
5874 test_class().run(
5875 test_rest, test_osm, manual_check, test_params.get(text_index)
5876 )
5877 else:
5878 for test, test_class in sorted(test_classes.items()):
5879 if fail_fast and test_rest.failed_tests:
5880 break
5881 test_class().run(test_rest, test_osm, manual_check, test_params.get(0))
5882 test_rest.print_results()
5883 exit(1 if test_rest.failed_tests else 0)
5884
5885 except TestException as e:
5886 logger.error(test + "Test {} Exception: {}".format(test, str(e)))
5887 exit(1)
5888 except getopt.GetoptError as e:
5889 logger.error(e)
5890 print(e, file=sys.stderr)
5891 exit(1)
5892 except Exception as e:
5893 logger.critical(test + " Exception: " + str(e), exc_info=True)