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