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