projects / osm / NBI.git / blob
? search:


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