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