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


f33935478b79f58c9de1bf145449feca26c16b75
[osm/NBI.git] / osm_nbi / tests / run_test.py
1 #! /usr/bin/python3
2 # -*- coding: utf-8 -*-
3
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
7 #
8 # http://www.apache.org/licenses/LICENSE-2.0
9 #
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
13 # implied.
14 # See the License for the specific language governing permissions and
15 # limitations under the License.
16
17 import getopt
18 import sys
19 import requests
20 import json
21 import logging
22 import yaml
23
24 # import json
25 # import tarfile
26 from time import sleep
27 from random import randint
28 import os
29 from sys import stderr
30 from uuid import uuid4
31 import re
32
33 __author__ = "Alfonso Tierno, alfonso.tiernosepulveda@telefonica.com"
34 __date__ = "$2018-03-01$"
35 __version__ = "0.3"
36 version_date = "Oct 2018"
37
38
39 def usage():
40 print("Usage: ", sys.argv[0], "[options]")
41 print(
42 " Performs system tests over running NBI. It can be used for real OSM test using option '--test-osm'"
43 )
44 print(
45 " If this is the case env variables 'OSMNBITEST_VIM_NAME' must be supplied to create a VIM if not exist "
46 "where deployment is done"
47 )
48 print("OPTIONS")
49 print(" -h|--help: shows this help")
50 print(" --insecure: Allows non trusted https NBI server")
51 print(" --list: list available tests")
52 print(
53 " --manual-check: Deployment tests stop after deployed to allow manual inspection. Only make sense with "
54 "'--test-osm'"
55 )
56 print(" -p|--password PASSWORD: NBI access password. 'admin' by default")
57 print(" ---project PROJECT: NBI access project. 'admin' by default")
58 print(
59 " --test TEST[,...]: Execute only a test or a comma separated list of tests"
60 )
61 print(
62 " --params key=val: params to the previous test. key can be vnfd-files, nsd-file, ns-name, ns-config"
63 )
64 print(
65 " --test-osm: If missing this test is intended for NBI only, no other OSM components are expected. Use "
66 "this flag to test the system. LCM and RO components are expected to be up and running"
67 )
68 print(
69 " --timeout TIMEOUT: General NBI timeout, by default {}s".format(timeout)
70 )
71 print(
72 " --timeout-deploy TIMEOUT: Timeout used for getting NS deployed, by default {}s".format(
73 timeout_deploy
74 )
75 )
76 print(
77 " --timeout-configure TIMEOUT: Timeout used for getting NS deployed and configured,"
78 " by default {}s".format(timeout_configure)
79 )
80 print(" -u|--user USERNAME: NBI access username. 'admin' by default")
81 print(
82 " --url URL: complete NBI server URL. 'https//localhost:9999/osm' by default"
83 )
84 print(" -v|--verbose print debug information, can be used several times")
85 print(" --no-verbose remove verbosity")
86 print(" --version: prints current version")
87 print("ENV variables used for real deployment tests with option osm-test.")
88 print(" export OSMNBITEST_VIM_NAME=vim-name")
89 print(" export OSMNBITEST_VIM_URL=vim-url")
90 print(" export OSMNBITEST_VIM_TYPE=vim-type")
91 print(" export OSMNBITEST_VIM_TENANT=vim-tenant")
92 print(" export OSMNBITEST_VIM_USER=vim-user")
93 print(" export OSMNBITEST_VIM_PASSWORD=vim-password")
94 print(' export OSMNBITEST_VIM_CONFIG="vim-config"')
95 print(' export OSMNBITEST_NS_NAME="vim-config"')
96 return
97
98
99 r_header_json = {"Content-type": "application/json"}
100 headers_json = {"Content-type": "application/json", "Accept": "application/json"}
101 r_header_yaml = {"Content-type": "application/yaml"}
102 headers_yaml = {"Content-type": "application/yaml", "Accept": "application/yaml"}
103 r_header_text = {"Content-type": "text/plain"}
104 r_header_octect = {"Content-type": "application/octet-stream"}
105 headers_text = {"Accept": "text/plain,application/yaml"}
106 r_header_zip = {"Content-type": "application/zip"}
107 headers_zip = {"Accept": "application/zip,application/yaml"}
108 headers_zip_yaml = {"Accept": "application/yaml", "Content-type": "application/zip"}
109 headers_zip_json = {"Accept": "application/json", "Content-type": "application/zip"}
110 headers_txt_json = {"Accept": "application/json", "Content-type": "text/plain"}
111 r_headers_yaml_location_vnfd = {
112 "Location": "/vnfpkgm/v1/vnf_packages_content/",
113 "Content-Type": "application/yaml",
114 }
115 r_headers_yaml_location_nsd = {
116 "Location": "/nsd/v1/ns_descriptors_content/",
117 "Content-Type": "application/yaml",
118 }
119 r_headers_yaml_location_nst = {
120 "Location": "/nst/v1/netslice_templates_content",
121 "Content-Type": "application/yaml",
122 }
123 r_headers_yaml_location_nslcmop = {
124 "Location": "nslcm/v1/ns_lcm_op_occs/",
125 "Content-Type": "application/yaml",
126 }
127 r_headers_yaml_location_nsilcmop = {
128 "Location": "/osm/nsilcm/v1/nsi_lcm_op_occs/",
129 "Content-Type": "application/yaml",
130 }
131
132 # test ones authorized
133 test_authorized_list = (
134 (
135 "AU1",
136 "Invalid vnfd id",
137 "GET",
138 "/vnfpkgm/v1/vnf_packages/non-existing-id",
139 headers_json,
140 None,
141 404,
142 r_header_json,
143 "json",
144 ),
145 (
146 "AU2",
147 "Invalid nsd id",
148 "GET",
149 "/nsd/v1/ns_descriptors/non-existing-id",
150 headers_yaml,
151 None,
152 404,
153 r_header_yaml,
154 "yaml",
155 ),
156 (
157 "AU3",
158 "Invalid nsd id",
159 "DELETE",
160 "/nsd/v1/ns_descriptors_content/non-existing-id",
161 headers_yaml,
162 None,
163 404,
164 r_header_yaml,
165 "yaml",
166 ),
167 )
168 timeout = 120 # general timeout
169 timeout_deploy = 60 * 10 # timeout for NS deploying without charms
170 timeout_configure = 60 * 20 # timeout for NS deploying and configuring
171
172
173 class TestException(Exception):
174 pass
175
176
177 class TestRest:
178 def __init__(
179 self,
180 url_base,
181 header_base=None,
182 verify=False,
183 user="admin",
184 password="admin",
185 project="admin",
186 ):
187 self.url_base = url_base
188 if header_base is None:
189 self.header_base = {}
190 else:
191 self.header_base = header_base.copy()
192 self.s = requests.session()
193 self.s.headers = self.header_base
194 self.verify = verify
195 self.token = False
196 self.user = user
197 self.password = password
198 self.project = project
199 self.vim_id = None
200 # contains ID of tests obtained from Location response header. "" key contains last obtained id
201 self.last_id = ""
202 self.test_name = None
203 self.step = 0 # number of subtest under test
204 self.passed_tests = 0
205 self.failed_tests = 0
206
207 def set_test_name(self, test_name):
208 self.test_name = test_name
209 self.step = 0
210 self.last_id = ""
211
212 def set_header(self, header):
213 self.s.headers.update(header)
214
215 def set_tet_name(self, test_name):
216 self.test_name = test_name
217
218 def unset_header(self, key):
219 if key in self.s.headers:
220 del self.s.headers[key]
221
222 def test(
223 self,
224 description,
225 method,
226 url,
227 headers,
228 payload,
229 expected_codes,
230 expected_headers,
231 expected_payload,
232 store_file=None,
233 pooling=False,
234 ):
235 """
236 Performs an http request and check http code response. Exit if different than allowed. It get the returned id
237 that can be used by following test in the URL with {name} where name is the name of the test
238 :param description: description of the test
239 :param method: HTTP method: GET,PUT,POST,DELETE,...
240 :param url: complete URL or relative URL
241 :param headers: request headers to add to the base headers
242 :param payload: Can be a dict, transformed to json, a text or a file if starts with '@'
243 :param expected_codes: expected response codes, can be int, int tuple or int range
244 :param expected_headers: expected response headers, dict with key values
245 :param expected_payload: expected payload, 0 if empty, 'yaml', 'json', 'text', 'zip', 'octet-stream'
246 :param store_file: filename to store content
247 :param pooling: if True do not count neither log this test. Because a pooling is done with many equal requests
248 :return: requests response
249 """
250 r = None
251 try:
252 if not self.s:
253 self.s = requests.session()
254 # URL
255 if not url:
256 url = self.url_base
257 elif not url.startswith("http"):
258 url = self.url_base + url
259
260 # replace url <> with the last ID
261 url = url.replace("<>", self.last_id)
262 if payload:
263 if isinstance(payload, str):
264 if payload.startswith("@"):
265 mode = "r"
266 file_name = payload[1:]
267 if payload.startswith("@b"):
268 mode = "rb"
269 file_name = payload[2:]
270 with open(file_name, mode) as f:
271 payload = f.read()
272 elif isinstance(payload, dict):
273 payload = json.dumps(payload)
274
275 if not pooling:
276 test_description = "Test {}{} {} {} {}".format(
277 self.test_name, self.step, description, method, url
278 )
279 logger.warning(test_description)
280 self.step += 1
281 stream = False
282 if expected_payload in ("zip", "octet-string") or store_file:
283 stream = True
284 __retry = 0
285 while True:
286 try:
287 r = getattr(self.s, method.lower())(
288 url,
289 data=payload,
290 headers=headers,
291 verify=self.verify,
292 stream=stream,
293 )
294 break
295 except requests.exceptions.ConnectionError as e:
296 if __retry == 2:
297 raise
298 logger.error("Exception {}. Retrying".format(e))
299 __retry += 1
300
301 if expected_payload in ("zip", "octet-string") or store_file:
302 logger.debug("RX {}".format(r.status_code))
303 else:
304 logger.debug("RX {}: {}".format(r.status_code, r.text))
305
306 # check response
307 if expected_codes:
308 if isinstance(expected_codes, int):
309 expected_codes = (expected_codes,)
310 if r.status_code not in expected_codes:
311 raise TestException(
312 "Got status {}. Expected {}. {}".format(
313 r.status_code, expected_codes, r.text
314 )
315 )
316
317 if expected_headers:
318 for header_key, header_val in expected_headers.items():
319 if header_key.lower() not in r.headers:
320 raise TestException("Header {} not present".format(header_key))
321 if header_val and header_val.lower() not in r.headers[header_key]:
322 raise TestException(
323 "Header {} does not contain {} but {}".format(
324 header_key, header_val, r.headers[header_key]
325 )
326 )
327
328 if expected_payload is not None:
329 if expected_payload == 0 and len(r.content) > 0:
330 raise TestException("Expected empty payload")
331 elif expected_payload == "json":
332 try:
333 r.json()
334 except Exception as e:
335 raise TestException(
336 "Expected json response payload, but got Exception {}".format(
337 e
338 )
339 )
340 elif expected_payload == "yaml":
341 try:
342 yaml.safe_load(r.text)
343 except Exception as e:
344 raise TestException(
345 "Expected yaml response payload, but got Exception {}".format(
346 e
347 )
348 )
349 elif expected_payload in ("zip", "octet-string"):
350 if len(r.content) == 0:
351 raise TestException(
352 "Expected some response payload, but got empty"
353 )
354 # try:
355 # tar = tarfile.open(None, 'r:gz', fileobj=r.raw)
356 # for tarinfo in tar:
357 # tarname = tarinfo.name
358 # print(tarname)
359 # except Exception as e:
360 # raise TestException("Expected zip response payload, but got Exception {}".format(e))
361 elif expected_payload == "text":
362 if len(r.content) == 0:
363 raise TestException(
364 "Expected some response payload, but got empty"
365 )
366 # r.text
367 if store_file:
368 with open(store_file, "wb") as fd:
369 for chunk in r.iter_content(chunk_size=128):
370 fd.write(chunk)
371
372 location = r.headers.get("Location")
373 if location:
374 _id = location[location.rfind("/") + 1 :]
375 if _id:
376 self.last_id = str(_id)
377 if not pooling:
378 self.passed_tests += 1
379 return r
380 except TestException as e:
381 self.failed_tests += 1
382 r_status_code = None
383 r_text = None
384 if r:
385 r_status_code = r.status_code
386 r_text = r.text
387 logger.error("{} \nRX code{}: {}".format(e, r_status_code, r_text))
388 return None
389 # exit(1)
390 except IOError as e:
391 if store_file:
392 logger.error("Cannot open file {}: {}".format(store_file, e))
393 else:
394 logger.error("Exception: {}".format(e), exc_info=True)
395 self.failed_tests += 1
396 return None
397 # exit(1)
398 except requests.exceptions.RequestException as e:
399 logger.error("Exception: {}".format(e))
400
401 def get_autorization(self): # user=None, password=None, project=None):
402 if (
403 self.token
404 ): # and self.user == user and self.password == password and self.project == project:
405 return
406 # self.user = user
407 # self.password = password
408 # self.project = project
409 r = self.test(
410 "Obtain token",
411 "POST",
412 "/admin/v1/tokens",
413 headers_json,
414 {
415 "username": self.user,
416 "password": self.password,
417 "project_id": self.project,
418 },
419 (200, 201),
420 r_header_json,
421 "json",
422 )
423 if not r:
424 return
425 response = r.json()
426 self.token = response["id"]
427 self.set_header({"Authorization": "Bearer {}".format(self.token)})
428
429 def remove_authorization(self):
430 if self.token:
431 self.test(
432 "Delete token",
433 "DELETE",
434 "/admin/v1/tokens/{}".format(self.token),
435 headers_json,
436 None,
437 (200, 201, 204),
438 None,
439 None,
440 )
441 self.token = None
442 self.unset_header("Authorization")
443
444 def get_create_vim(self, test_osm):
445 if self.vim_id:
446 return self.vim_id
447 self.get_autorization()
448 if test_osm:
449 vim_name = os.environ.get("OSMNBITEST_VIM_NAME")
450 if not vim_name:
451 raise TestException(
452 "Needed to define OSMNBITEST_VIM_XXX variables to create a real VIM for deployment"
453 )
454 else:
455 vim_name = "fakeVim"
456 # Get VIM
457 r = self.test(
458 "Get VIM ID",
459 "GET",
460 "/admin/v1/vim_accounts?name={}".format(vim_name),
461 headers_json,
462 None,
463 200,
464 r_header_json,
465 "json",
466 )
467 if not r:
468 return
469 vims = r.json()
470 if vims:
471 return vims[0]["_id"]
472 # Add VIM
473 if test_osm:
474 # check needed environ parameters:
475 if not os.environ.get("OSMNBITEST_VIM_URL") or not os.environ.get(
476 "OSMNBITEST_VIM_TENANT"
477 ):
478 raise TestException(
479 "Env OSMNBITEST_VIM_URL and OSMNBITEST_VIM_TENANT are needed for create a real VIM"
480 " to deploy on whit the --test-osm option"
481 )
482 vim_data = "{{schema_version: '1.0', name: '{}', vim_type: {}, vim_url: '{}', vim_tenant_name: '{}', " "vim_user: {}, vim_password: {}".format(
483 vim_name,
484 os.environ.get("OSMNBITEST_VIM_TYPE", "openstack"),
485 os.environ.get("OSMNBITEST_VIM_URL"),
486 os.environ.get("OSMNBITEST_VIM_TENANT"),
487 os.environ.get("OSMNBITEST_VIM_USER"),
488 os.environ.get("OSMNBITEST_VIM_PASSWORD"),
489 )
490 if os.environ.get("OSMNBITEST_VIM_CONFIG"):
491 vim_data += " ,config: {}".format(
492 os.environ.get("OSMNBITEST_VIM_CONFIG")
493 )
494 vim_data += "}"
495 else:
496 vim_data = (
497 "{schema_version: '1.0', name: fakeVim, vim_type: openstack, vim_url: 'http://10.11.12.13/fake'"
498 ", vim_tenant_name: 'vimtenant', vim_user: vimuser, vim_password: vimpassword}"
499 )
500 self.test(
501 "Create VIM",
502 "POST",
503 "/admin/v1/vim_accounts",
504 headers_yaml,
505 vim_data,
506 (201, 202),
507 {"Location": "/admin/v1/vim_accounts/", "Content-Type": "application/yaml"},
508 "yaml",
509 )
510 return self.last_id
511
512 def print_results(self):
513 print("\n\n\n--------------------------------------------")
514 print(
515 "TEST RESULTS: Total: {}, Passed: {}, Failed: {}".format(
516 self.passed_tests + self.failed_tests,
517 self.passed_tests,
518 self.failed_tests,
519 )
520 )
521 print("--------------------------------------------")
522
523 def wait_until_delete(self, url_op, timeout_delete):
524 """
525 Make a pooling until topic is not present, because of deleted
526 :param url_op:
527 :param timeout_delete:
528 :return:
529 """
530 description = "Wait to topic being deleted"
531 test_description = "Test {}{} {} {} {}".format(
532 self.test_name, self.step, description, "GET", url_op
533 )
534 logger.warning(test_description)
535 self.step += 1
536
537 wait = timeout_delete
538 while wait >= 0:
539 r = self.test(
540 description,
541 "GET",
542 url_op,
543 headers_yaml,
544 None,
545 (200, 404),
546 None,
547 r_header_yaml,
548 "yaml",
549 pooling=True,
550 )
551 if not r:
552 return
553 if r.status_code == 404:
554 self.passed_tests += 1
555 break
556 elif r.status_code == 200:
557 wait -= 5
558 sleep(5)
559 else:
560 raise TestException(
561 "Topic is not deleted after {} seconds".format(timeout_delete)
562 )
563 self.failed_tests += 1
564
565 def wait_operation_ready(self, ns_nsi, opp_id, timeout, expected_fail=False):
566 """
567 Wait until nslcmop or nsilcmop finished
568 :param ns_nsi: "ns" o "nsi"
569 :param opp_id: Id o fthe operation
570 :param timeout:
571 :param expected_fail:
572 :return: None. Updates passed/failed_tests
573 """
574 if ns_nsi == "ns":
575 url_op = "/nslcm/v1/ns_lcm_op_occs/{}".format(opp_id)
576 else:
577 url_op = "/nsilcm/v1/nsi_lcm_op_occs/{}".format(opp_id)
578 description = "Wait to {} lcm operation complete".format(ns_nsi)
579 test_description = "Test {}{} {} {} {}".format(
580 self.test_name, self.step, description, "GET", url_op
581 )
582 logger.warning(test_description)
583 self.step += 1
584 wait = timeout
585 while wait >= 0:
586 r = self.test(
587 description,
588 "GET",
589 url_op,
590 headers_json,
591 None,
592 200,
593 r_header_json,
594 "json",
595 pooling=True,
596 )
597 if not r:
598 return
599 nslcmop = r.json()
600 if "COMPLETED" in nslcmop["operationState"]:
601 if expected_fail:
602 logger.error(
603 "NS terminate has success, expecting failing: {}".format(
604 nslcmop["detailed-status"]
605 )
606 )
607 self.failed_tests += 1
608 else:
609 self.passed_tests += 1
610 break
611 elif "FAILED" in nslcmop["operationState"]:
612 if not expected_fail:
613 logger.error(
614 "NS terminate has failed: {}".format(nslcmop["detailed-status"])
615 )
616 self.failed_tests += 1
617 else:
618 self.passed_tests += 1
619 break
620
621 print(".", end="", file=stderr)
622 wait -= 10
623 sleep(10)
624 else:
625 self.failed_tests += 1
626 logger.error(
627 "NS instantiate is not terminate after {} seconds".format(timeout)
628 )
629 return
630 print("", file=stderr)
631
632
633 class TestNonAuthorized:
634 description = "Test invalid URLs. methods and no authorization"
635
636 @staticmethod
637 def run(engine, test_osm, manual_check, test_params=None):
638 engine.set_test_name("NonAuth")
639 engine.remove_authorization()
640 test_not_authorized_list = (
641 (
642 "Invalid token",
643 "GET",
644 "/admin/v1/users",
645 headers_json,
646 None,
647 401,
648 r_header_json,
649 "json",
650 ),
651 (
652 "Invalid URL",
653 "POST",
654 "/admin/v1/nonexist",
655 headers_yaml,
656 None,
657 405,
658 r_header_yaml,
659 "yaml",
660 ),
661 (
662 "Invalid version",
663 "DELETE",
664 "/admin/v2/users",
665 headers_yaml,
666 None,
667 405,
668 r_header_yaml,
669 "yaml",
670 ),
671 )
672 for t in test_not_authorized_list:
673 engine.test(*t)
674
675
676 class TestUsersProjects:
677 description = "test project and user creation"
678
679 @staticmethod
680 def run(engine, test_osm, manual_check, test_params=None):
681 engine.set_test_name("UserProject")
682 # backend = test_params.get("backend") if test_params else None # UNUSED
683
684 # Initialisation
685 p1 = p2 = p3 = None
686 padmin = pbad = None
687 u1 = u2 = u3 = u4 = None
688
689 engine.get_autorization()
690
691 res = engine.test(
692 "Create project non admin 1",
693 "POST",
694 "/admin/v1/projects",
695 headers_json,
696 {"name": "P1"},
697 (201, 204),
698 {"Location": "/admin/v1/projects/", "Content-Type": "application/json"},
699 "json",
700 )
701 p1 = engine.last_id if res else None
702
703 res = engine.test(
704 "Create project admin",
705 "POST",
706 "/admin/v1/projects",
707 headers_json,
708 {"name": "Padmin", "admin": True},
709 (201, 204),
710 {"Location": "/admin/v1/projects/", "Content-Type": "application/json"},
711 "json",
712 )
713 padmin = engine.last_id if res else None
714
715 res = engine.test(
716 "Create project bad format",
717 "POST",
718 "/admin/v1/projects",
719 headers_json,
720 {"name": 1},
721 (400, 422),
722 r_header_json,
723 "json",
724 )
725 pbad = engine.last_id if res else None
726
727 res = engine.test(
728 "Get project admin role",
729 "GET",
730 "/admin/v1/roles?name=project_admin",
731 headers_json,
732 {},
733 (200),
734 {"Content-Type": "application/json"},
735 "json",
736 )
737 rpa = res.json()[0]["_id"] if res else None
738 res = engine.test(
739 "Get project user role",
740 "GET",
741 "/admin/v1/roles?name=project_user",
742 headers_json,
743 {},
744 (200),
745 {"Content-Type": "application/json"},
746 "json",
747 )
748 rpu = res.json()[0]["_id"] if res else None
749 res = engine.test(
750 "Get system admin role",
751 "GET",
752 "/admin/v1/roles?name=system_admin",
753 headers_json,
754 {},
755 (200),
756 {"Content-Type": "application/json"},
757 "json",
758 )
759 rsa = res.json()[0]["_id"] if res else None
760
761 data = {"username": "U1", "password": "pw1"}
762 p2 = uuid4().hex
763 data["project_role_mappings"] = [
764 {"project": p1, "role": rpa},
765 {"project": p2, "role": rpa},
766 {"project": padmin, "role": rpu},
767 ]
768 rc = 201
769 xhd = {"Location": "/admin/v1/users/", "Content-Type": "application/json"}
770 res = engine.test(
771 "Create user with bad project and force",
772 "POST",
773 "/admin/v1/users?FORCE=True",
774 headers_json,
775 data,
776 rc,
777 xhd,
778 "json",
779 )
780 if res:
781 u1 = engine.last_id
782 else:
783 # User is created sometimes even though an exception is raised
784 res = engine.test(
785 "Get user U1",
786 "GET",
787 "/admin/v1/users?username=U1",
788 headers_json,
789 {},
790 (200),
791 {"Content-Type": "application/json"},
792 "json",
793 )
794 u1 = res.json()[0]["_id"] if res else None
795
796 data = {"username": "U2", "password": "pw2"}
797 data["project_role_mappings"] = [
798 {"project": p1, "role": rpa},
799 {"project": padmin, "role": rsa},
800 ]
801 res = engine.test(
802 "Create user 2",
803 "POST",
804 "/admin/v1/users",
805 headers_json,
806 data,
807 201,
808 {"Location": "/admin/v1/users/", "Content-Type": "application/json"},
809 "json",
810 )
811 u2 = engine.last_id if res else None
812
813 if u1:
814 ftt = "project_role_mappings"
815 xpr = [{"project": p1, "role": rpa}, {"project": padmin, "role": rpu}]
816 data = {ftt: xpr}
817 engine.test(
818 "Edit user U1, delete P2 project",
819 "PATCH",
820 "/admin/v1/users/" + u1,
821 headers_json,
822 data,
823 204,
824 None,
825 None,
826 )
827 res = engine.test(
828 "Check user U1, contains the right projects",
829 "GET",
830 "/admin/v1/users/" + u1,
831 headers_json,
832 None,
833 200,
834 None,
835 json,
836 )
837 if res:
838 rj = res.json()
839 xpr[0]["project_name"] = "P1"
840 xpr[0]["role_name"] = "project_admin"
841 xpr[1]["project_name"] = "Padmin"
842 xpr[1]["role_name"] = "project_user"
843 ok = True
844 for pr in rj[ftt]:
845 if pr not in xpr:
846 ok = False
847 for pr in xpr:
848 if pr not in rj[ftt]:
849 ok = False
850 if not ok:
851 logger.error(
852 "User {} '{}' are different than expected '{}'. Edition was not done properly".format(
853 ftt, rj[ftt], xpr
854 )
855 )
856 engine.failed_tests += 1
857
858 p2 = None # To prevent deletion attempts
859
860 # Add a test of 'default project' for Keystone?
861
862 if u2:
863 engine.test(
864 "Edit user U2, change password",
865 "PUT",
866 "/admin/v1/users/" + u2,
867 headers_json,
868 {"password": "pw2_new"},
869 204,
870 None,
871 None,
872 )
873
874 if p1:
875 engine.test(
876 "Change to project P1 non existing",
877 "POST",
878 "/admin/v1/tokens/",
879 headers_json,
880 {"project_id": p1},
881 401,
882 r_header_json,
883 "json",
884 )
885
886 if u2 and p1:
887 res = engine.test(
888 "Change to user U2 project P1",
889 "POST",
890 "/admin/v1/tokens",
891 headers_json,
892 {"username": "U2", "password": "pw2_new", "project_id": "P1"},
893 (200, 201),
894 r_header_json,
895 "json",
896 )
897 if res:
898 rj = res.json()
899 engine.set_header({"Authorization": "Bearer {}".format(rj["id"])})
900
901 engine.test(
902 "Edit user projects non admin",
903 "PUT",
904 "/admin/v1/users/U1",
905 headers_json,
906 {"remove_project_role_mappings": [{"project": "P1", "role": None}]},
907 401,
908 r_header_json,
909 "json",
910 )
911
912 res = engine.test(
913 "Add new project non admin",
914 "POST",
915 "/admin/v1/projects",
916 headers_json,
917 {"name": "P2"},
918 401,
919 r_header_json,
920 "json",
921 )
922 if res is None or res.status_code == 201:
923 # The project has been created even though it shouldn't
924 res = engine.test(
925 "Get project P2",
926 "GET",
927 "/admin/v1/projects/P2",
928 headers_json,
929 None,
930 200,
931 r_header_json,
932 "json",
933 )
934 p2 = res.json()["_id"] if res else None
935
936 if p1:
937 data = {"username": "U3", "password": "pw3"}
938 data["project_role_mappings"] = [{"project": p1, "role": rpu}]
939 res = engine.test(
940 "Add new user non admin",
941 "POST",
942 "/admin/v1/users",
943 headers_json,
944 data,
945 401,
946 r_header_json,
947 "json",
948 )
949 if res is None or res.status_code == 201:
950 # The user has been created even though it shouldn't
951 res = engine.test(
952 "Get user U3",
953 "GET",
954 "/admin/v1/users/U3",
955 headers_json,
956 None,
957 200,
958 r_header_json,
959 "json",
960 )
961 u3 = res.json()["_id"] if res else None
962 else:
963 u3 = None
964
965 if padmin:
966 res = engine.test(
967 "Change to user U2 project Padmin",
968 "POST",
969 "/admin/v1/tokens",
970 headers_json,
971 {
972 "project_id": "Padmin"
973 }, # Caused a Keystone authentication error
974 # {"username": "U2", "password": "pw2_new", "project_id": "Padmin"},
975 (200, 201),
976 r_header_json,
977 "json",
978 )
979 if res:
980 rj = res.json()
981 engine.set_header(
982 {"Authorization": "Bearer {}".format(rj["id"])}
983 )
984
985 res = engine.test(
986 "Add new project admin",
987 "POST",
988 "/admin/v1/projects",
989 headers_json,
990 {"name": "P3"},
991 (201, 204),
992 {
993 "Location": "/admin/v1/projects/",
994 "Content-Type": "application/json",
995 },
996 "json",
997 )
998 p3 = engine.last_id if res else None
999
1000 if p1:
1001 data = {"username": "U4", "password": "pw4"}
1002 data["project_role_mappings"] = [
1003 {"project": p1, "role": rpa}
1004 ]
1005 res = engine.test(
1006 "Add new user admin",
1007 "POST",
1008 "/admin/v1/users",
1009 headers_json,
1010 data,
1011 (201, 204),
1012 {
1013 "Location": "/admin/v1/users/",
1014 "Content-Type": "application/json",
1015 },
1016 "json",
1017 )
1018 u4 = engine.last_id if res else None
1019 else:
1020 u4 = None
1021
1022 if u4 and p3:
1023 data = {
1024 "project_role_mappings": [{"project": p3, "role": rpa}]
1025 }
1026 engine.test(
1027 "Edit user projects admin",
1028 "PUT",
1029 "/admin/v1/users/U4",
1030 headers_json,
1031 data,
1032 204,
1033 None,
1034 None,
1035 )
1036 # Project is deleted even though it shouldn't - PROVISIONAL?
1037 res = engine.test(
1038 "Delete project P3 conflict",
1039 "DELETE",
1040 "/admin/v1/projects/" + p3,
1041 headers_json,
1042 None,
1043 409,
1044 None,
1045 None,
1046 )
1047 if res and res.status_code in (200, 204):
1048 p3 = None
1049 if p3:
1050 res = engine.test(
1051 "Delete project P3 forcing",
1052 "DELETE",
1053 "/admin/v1/projects/" + p3 + "?FORCE=True",
1054 headers_json,
1055 None,
1056 204,
1057 None,
1058 None,
1059 )
1060 if res and res.status_code in (200, 204):
1061 p3 = None
1062
1063 if u2:
1064 res = engine.test(
1065 "Delete user U2. Conflict deleting own user",
1066 "DELETE",
1067 "/admin/v1/users/" + u2,
1068 headers_json,
1069 None,
1070 409,
1071 r_header_json,
1072 "json",
1073 )
1074 if res is None or res.status_code in (200, 204):
1075 u2 = None
1076 if u4:
1077 res = engine.test(
1078 "Delete user U4",
1079 "DELETE",
1080 "/admin/v1/users/" + u4,
1081 headers_json,
1082 None,
1083 204,
1084 None,
1085 None,
1086 )
1087 if res and res.status_code in (200, 204):
1088 u4 = None
1089 if p3:
1090 res = engine.test(
1091 "Delete project P3",
1092 "DELETE",
1093 "/admin/v1/projects/" + p3,
1094 headers_json,
1095 None,
1096 204,
1097 None,
1098 None,
1099 )
1100 if res and res.status_code in (200, 204):
1101 p3 = None
1102
1103 if u3:
1104 res = engine.test(
1105 "Delete user U3",
1106 "DELETE",
1107 "/admin/v1/users/" + u3,
1108 headers_json,
1109 None,
1110 204,
1111 None,
1112 None,
1113 )
1114 if res:
1115 u3 = None
1116
1117 # change to admin
1118 engine.remove_authorization() # To force get authorization
1119 engine.get_autorization()
1120 if u1:
1121 engine.test(
1122 "Delete user U1",
1123 "DELETE",
1124 "/admin/v1/users/" + u1,
1125 headers_json,
1126 None,
1127 204,
1128 None,
1129 None,
1130 )
1131 if u2:
1132 engine.test(
1133 "Delete user U2",
1134 "DELETE",
1135 "/admin/v1/users/" + u2,
1136 headers_json,
1137 None,
1138 204,
1139 None,
1140 None,
1141 )
1142 if u3:
1143 engine.test(
1144 "Delete user U3",
1145 "DELETE",
1146 "/admin/v1/users/" + u3,
1147 headers_json,
1148 None,
1149 204,
1150 None,
1151 None,
1152 )
1153 if u4:
1154 engine.test(
1155 "Delete user U4",
1156 "DELETE",
1157 "/admin/v1/users/" + u4,
1158 headers_json,
1159 None,
1160 204,
1161 None,
1162 None,
1163 )
1164 if p1:
1165 engine.test(
1166 "Delete project P1",
1167 "DELETE",
1168 "/admin/v1/projects/" + p1,
1169 headers_json,
1170 None,
1171 204,
1172 None,
1173 None,
1174 )
1175 if p2:
1176 engine.test(
1177 "Delete project P2",
1178 "DELETE",
1179 "/admin/v1/projects/" + p2,
1180 headers_json,
1181 None,
1182 204,
1183 None,
1184 None,
1185 )
1186 if p3:
1187 engine.test(
1188 "Delete project P3",
1189 "DELETE",
1190 "/admin/v1/projects/" + p3,
1191 headers_json,
1192 None,
1193 204,
1194 None,
1195 None,
1196 )
1197 if padmin:
1198 engine.test(
1199 "Delete project Padmin",
1200 "DELETE",
1201 "/admin/v1/projects/" + padmin,
1202 headers_json,
1203 None,
1204 204,
1205 None,
1206 None,
1207 )
1208 if pbad:
1209 engine.test(
1210 "Delete bad project",
1211 "DELETE",
1212 "/admin/v1/projects/" + pbad,
1213 headers_json,
1214 None,
1215 204,
1216 None,
1217 None,
1218 )
1219
1220 # BEGIN New Tests - Addressing Projects/Users by Name/ID
1221 pid1 = pid2 = None
1222 uid1 = uid2 = None
1223 res = engine.test(
1224 "Create new project P1",
1225 "POST",
1226 "/admin/v1/projects",
1227 headers_json,
1228 {"name": "P1"},
1229 201,
1230 {"Location": "/admin/v1/projects/", "Content-Type": "application/json"},
1231 "json",
1232 )
1233 if res:
1234 pid1 = res.json()["id"]
1235 # print("# pid =", pid1)
1236 res = engine.test(
1237 "Create new project P2",
1238 "POST",
1239 "/admin/v1/projects",
1240 headers_json,
1241 {"name": "P2"},
1242 201,
1243 {"Location": "/admin/v1/projects/", "Content-Type": "application/json"},
1244 "json",
1245 )
1246 if res:
1247 pid2 = res.json()["id"]
1248 # print("# pid =", pid2)
1249 data = {"username": "U1", "password": "pw1"}
1250 data["project_role_mappings"] = [{"project": pid1, "role": rpu}]
1251 res = engine.test(
1252 "Create new user U1",
1253 "POST",
1254 "/admin/v1/users",
1255 headers_json,
1256 data,
1257 201,
1258 {"Location": "/admin/v1/users/", "Content-Type": "application/json"},
1259 "json",
1260 )
1261 if res:
1262 uid1 = res.json()["id"]
1263 # print("# uid =", uid1)
1264 data = {"username": "U2", "password": "pw2"}
1265 data["project_role_mappings"] = [{"project": pid2, "role": rpu}]
1266 res = engine.test(
1267 "Create new user U2",
1268 "POST",
1269 "/admin/v1/users",
1270 headers_json,
1271 data,
1272 201,
1273 {"Location": "/admin/v1/users/", "Content-Type": "application/json"},
1274 "json",
1275 )
1276 if res:
1277 uid2 = res.json()["id"]
1278 # print("# uid =", uid2)
1279 if pid1:
1280 engine.test(
1281 "Get Project P1 by Name",
1282 "GET",
1283 "/admin/v1/projects/P1",
1284 headers_json,
1285 None,
1286 200,
1287 None,
1288 "json",
1289 )
1290 engine.test(
1291 "Get Project P1 by ID",
1292 "GET",
1293 "/admin/v1/projects/" + pid1,
1294 headers_json,
1295 None,
1296 200,
1297 None,
1298 "json",
1299 )
1300 if uid1:
1301 engine.test(
1302 "Get User U1 by Name",
1303 "GET",
1304 "/admin/v1/users/U1",
1305 headers_json,
1306 None,
1307 200,
1308 None,
1309 "json",
1310 )
1311 engine.test(
1312 "Get User U1 by ID",
1313 "GET",
1314 "/admin/v1/users/" + uid1,
1315 headers_json,
1316 None,
1317 200,
1318 None,
1319 "json",
1320 )
1321 if pid1:
1322 res = engine.test(
1323 "Rename Project P1 by Name",
1324 "PUT",
1325 "/admin/v1/projects/P1",
1326 headers_json,
1327 {"name": "P3"},
1328 204,
1329 None,
1330 None,
1331 )
1332 if res:
1333 engine.test(
1334 "Get Project P1 by new Name",
1335 "GET",
1336 "/admin/v1/projects/P3",
1337 headers_json,
1338 None,
1339 200,
1340 None,
1341 "json",
1342 )
1343 if pid2:
1344 res = engine.test(
1345 "Rename Project P2 by ID",
1346 "PUT",
1347 "/admin/v1/projects/" + pid2,
1348 headers_json,
1349 {"name": "P4"},
1350 204,
1351 None,
1352 None,
1353 )
1354 if res:
1355 engine.test(
1356 "Get Project P2 by new Name",
1357 "GET",
1358 "/admin/v1/projects/P4",
1359 headers_json,
1360 None,
1361 200,
1362 None,
1363 "json",
1364 )
1365
1366 if uid1:
1367 res = engine.test(
1368 "Rename User U1 by Name",
1369 "PUT",
1370 "/admin/v1/users/U1",
1371 headers_json,
1372 {"username": "U3"},
1373 204,
1374 None,
1375 None,
1376 )
1377 if res:
1378 engine.test(
1379 "Get User U1 by new Name",
1380 "GET",
1381 "/admin/v1/users/U3",
1382 headers_json,
1383 None,
1384 200,
1385 None,
1386 "json",
1387 )
1388
1389 if uid2:
1390 res = engine.test(
1391 "Rename User U2 by ID",
1392 "PUT",
1393 "/admin/v1/users/" + uid2,
1394 headers_json,
1395 {"username": "U4"},
1396 204,
1397 None,
1398 None,
1399 )
1400 if res:
1401 engine.test(
1402 "Get User U2 by new Name",
1403 "GET",
1404 "/admin/v1/users/U4",
1405 headers_json,
1406 None,
1407 200,
1408 None,
1409 "json",
1410 )
1411 if uid1:
1412 res = engine.test(
1413 "Delete User U1 by Name",
1414 "DELETE",
1415 "/admin/v1/users/U3",
1416 headers_json,
1417 None,
1418 204,
1419 None,
1420 None,
1421 )
1422 if res:
1423 uid1 = None
1424
1425 if uid2:
1426 res = engine.test(
1427 "Delete User U2 by ID",
1428 "DELETE",
1429 "/admin/v1/users/" + uid2,
1430 headers_json,
1431 None,
1432 204,
1433 None,
1434 None,
1435 )
1436 if res:
1437 uid2 = None
1438
1439 if pid1:
1440 res = engine.test(
1441 "Delete Project P1 by Name",
1442 "DELETE",
1443 "/admin/v1/projects/P3",
1444 headers_json,
1445 None,
1446 204,
1447 None,
1448 None,
1449 )
1450 if res:
1451 pid1 = None
1452
1453 if pid2:
1454 res = engine.test(
1455 "Delete Project P2 by ID",
1456 "DELETE",
1457 "/admin/v1/projects/" + pid2,
1458 headers_json,
1459 None,
1460 204,
1461 None,
1462 None,
1463 )
1464 if res:
1465 pid2 = None
1466
1467 # END New Tests - Addressing Projects/Users by Name
1468
1469 # CLEANUP
1470 if pid1:
1471 engine.test(
1472 "Delete Project P1",
1473 "DELETE",
1474 "/admin/v1/projects/" + pid1,
1475 headers_json,
1476 None,
1477 204,
1478 None,
1479 None,
1480 )
1481 if pid2:
1482 engine.test(
1483 "Delete Project P2",
1484 "DELETE",
1485 "/admin/v1/projects/" + pid2,
1486 headers_json,
1487 None,
1488 204,
1489 None,
1490 None,
1491 )
1492 if uid1:
1493 engine.test(
1494 "Delete User U1",
1495 "DELETE",
1496 "/admin/v1/users/" + uid1,
1497 headers_json,
1498 None,
1499 204,
1500 None,
1501 None,
1502 )
1503 if uid2:
1504 engine.test(
1505 "Delete User U2",
1506 "DELETE",
1507 "/admin/v1/users/" + uid2,
1508 headers_json,
1509 None,
1510 204,
1511 None,
1512 None,
1513 )
1514
1515 engine.remove_authorization() # To finish
1516
1517
1518 class TestProjectsDescriptors:
1519 description = "test descriptors visibility among projects"
1520
1521 @staticmethod
1522 def run(engine, test_osm, manual_check, test_params=None):
1523 vnfd_ids = []
1524 engine.set_test_name("ProjectDescriptors")
1525 engine.get_autorization()
1526
1527 project_admin_id = None
1528 res = engine.test(
1529 "Get my project Padmin",
1530 "GET",
1531 "/admin/v1/projects/{}".format(engine.project),
1532 headers_json,
1533 None,
1534 200,
1535 r_header_json,
1536 "json",
1537 )
1538 if res:
1539 response = res.json()
1540 project_admin_id = response["_id"]
1541 engine.test(
1542 "Create project Padmin",
1543 "POST",
1544 "/admin/v1/projects",
1545 headers_json,
1546 {"name": "Padmin", "admin": True},
1547 (201, 204),
1548 {"Location": "/admin/v1/projects/", "Content-Type": "application/json"},
1549 "json",
1550 )
1551 engine.test(
1552 "Create project P2",
1553 "POST",
1554 "/admin/v1/projects",
1555 headers_json,
1556 {"name": "P2"},
1557 (201, 204),
1558 {"Location": "/admin/v1/projects/", "Content-Type": "application/json"},
1559 "json",
1560 )
1561 engine.test(
1562 "Create project P3",
1563 "POST",
1564 "/admin/v1/projects",
1565 headers_json,
1566 {"name": "P3"},
1567 (201, 204),
1568 {"Location": "/admin/v1/projects/", "Content-Type": "application/json"},
1569 "json",
1570 )
1571
1572 engine.test(
1573 "Create user U1",
1574 "POST",
1575 "/admin/v1/users",
1576 headers_json,
1577 {
1578 "username": "U1",
1579 "password": "pw1",
1580 "project_role_mappings": [
1581 {"project": "Padmin", "role": "system_admin"},
1582 {"project": "P2", "role": "project_admin"},
1583 {"project": "P3", "role": "project_admin"},
1584 ],
1585 },
1586 201,
1587 {"Location": "/admin/v1/users/", "Content-Type": "application/json"},
1588 "json",
1589 )
1590
1591 engine.test(
1592 "Onboard VNFD id1",
1593 "POST",
1594 "/vnfpkgm/v1/vnf_packages_content?id=id1",
1595 headers_yaml,
1596 TestDescriptors.vnfd_empty,
1597 201,
1598 r_headers_yaml_location_vnfd,
1599 "yaml",
1600 )
1601 vnfd_ids.append(engine.last_id)
1602 engine.test(
1603 "Onboard VNFD id2 PUBLIC",
1604 "POST",
1605 "/vnfpkgm/v1/vnf_packages_content?id=id2&PUBLIC=TRUE",
1606 headers_yaml,
1607 TestDescriptors.vnfd_empty,
1608 201,
1609 r_headers_yaml_location_vnfd,
1610 "yaml",
1611 )
1612 vnfd_ids.append(engine.last_id)
1613 engine.test(
1614 "Onboard VNFD id3",
1615 "POST",
1616 "/vnfpkgm/v1/vnf_packages_content?id=id3&PUBLIC=FALSE",
1617 headers_yaml,
1618 TestDescriptors.vnfd_empty,
1619 201,
1620 r_headers_yaml_location_vnfd,
1621 "yaml",
1622 )
1623 vnfd_ids.append(engine.last_id)
1624
1625 res = engine.test(
1626 "Get VNFD descriptors",
1627 "GET",
1628 "/vnfpkgm/v1/vnf_packages?id=id1,id2,id3",
1629 headers_json,
1630 None,
1631 200,
1632 r_header_json,
1633 "json",
1634 )
1635 response = res.json()
1636 if len(response) != 3:
1637 logger.error(
1638 "Only 3 vnfds should be present for project admin. {} listed".format(
1639 len(response)
1640 )
1641 )
1642 engine.failed_tests += 1
1643
1644 # Change to other project Padmin
1645 res = engine.test(
1646 "Change to user U1 project Padmin",
1647 "POST",
1648 "/admin/v1/tokens",
1649 headers_json,
1650 {"username": "U1", "password": "pw1", "project_id": "Padmin"},
1651 (200, 201),
1652 r_header_json,
1653 "json",
1654 )
1655 if res:
1656 response = res.json()
1657 engine.set_header({"Authorization": "Bearer {}".format(response["id"])})
1658
1659 # list vnfds
1660 res = engine.test(
1661 "List VNFD descriptors for Padmin",
1662 "GET",
1663 "/vnfpkgm/v1/vnf_packages",
1664 headers_json,
1665 None,
1666 200,
1667 r_header_json,
1668 "json",
1669 )
1670 response = res.json()
1671 if len(response) != 0:
1672 logger.error(
1673 "Only 0 vnfds should be present for project Padmin. {} listed".format(
1674 len(response)
1675 )
1676 )
1677 engine.failed_tests += 1
1678
1679 # list Public vnfds
1680 res = engine.test(
1681 "List VNFD public descriptors",
1682 "GET",
1683 "/vnfpkgm/v1/vnf_packages?PUBLIC=True",
1684 headers_json,
1685 None,
1686 200,
1687 r_header_json,
1688 "json",
1689 )
1690 response = res.json()
1691 if len(response) != 1:
1692 logger.error(
1693 "Only 1 vnfds should be present for project Padmin. {} listed".format(
1694 len(response)
1695 )
1696 )
1697 engine.failed_tests += 1
1698
1699 # list vnfds belonging to project "admin"
1700 res = engine.test(
1701 "List VNFD of admin project",
1702 "GET",
1703 "/vnfpkgm/v1/vnf_packages?ADMIN={}".format(project_admin_id),
1704 headers_json,
1705 None,
1706 200,
1707 r_header_json,
1708 "json",
1709 )
1710 if res:
1711 response = res.json()
1712 if len(response) != 3:
1713 logger.error(
1714 "Only 3 vnfds should be present for project Padmin. {} listed".format(
1715 len(response)
1716 )
1717 )
1718 engine.failed_tests += 1
1719
1720 # Get Public vnfds
1721 engine.test(
1722 "Get VNFD public descriptors",
1723 "GET",
1724 "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_ids[1]),
1725 headers_json,
1726 None,
1727 200,
1728 r_header_json,
1729 "json",
1730 )
1731 # Edit not owned vnfd
1732 engine.test(
1733 "Edit VNFD ",
1734 "PATCH",
1735 "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_ids[0]),
1736 headers_yaml,
1737 "{name: pepe}",
1738 404,
1739 r_header_yaml,
1740 "yaml",
1741 )
1742
1743 # Add to my catalog
1744 engine.test(
1745 "Add VNFD id2 to my catalog",
1746 "PATCH",
1747 "/vnfpkgm/v1/vnf_packages/{}?SET_PROJECT".format(vnfd_ids[1]),
1748 headers_json,
1749 None,
1750 204,
1751 None,
1752 0,
1753 )
1754
1755 # Add a new vnfd
1756 engine.test(
1757 "Onboard VNFD id4",
1758 "POST",
1759 "/vnfpkgm/v1/vnf_packages_content?id=id4",
1760 headers_yaml,
1761 TestDescriptors.vnfd_empty,
1762 201,
1763 r_headers_yaml_location_vnfd,
1764 "yaml",
1765 )
1766 vnfd_ids.append(engine.last_id)
1767
1768 # list vnfds
1769 res = engine.test(
1770 "List VNFD public descriptors",
1771 "GET",
1772 "/vnfpkgm/v1/vnf_packages",
1773 headers_json,
1774 None,
1775 200,
1776 r_header_json,
1777 "json",
1778 )
1779 response = res.json()
1780 if len(response) != 2:
1781 logger.error(
1782 "Only 2 vnfds should be present for project Padmin. {} listed".format(
1783 len(response)
1784 )
1785 )
1786 engine.failed_tests += 1
1787
1788 if manual_check:
1789 input(
1790 "VNFDs have been omboarded. Perform manual check and press enter to resume"
1791 )
1792
1793 test_rest.test(
1794 "Delete VNFD id2",
1795 "DELETE",
1796 "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_ids[1]),
1797 headers_yaml,
1798 None,
1799 204,
1800 None,
1801 0,
1802 )
1803
1804 # change to admin project
1805 engine.remove_authorization() # To force get authorization
1806 engine.get_autorization()
1807 test_rest.test(
1808 "Delete VNFD id1",
1809 "DELETE",
1810 "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_ids[0]),
1811 headers_yaml,
1812 None,
1813 204,
1814 None,
1815 0,
1816 )
1817 test_rest.test(
1818 "Delete VNFD id2",
1819 "DELETE",
1820 "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_ids[1]),
1821 headers_yaml,
1822 None,
1823 204,
1824 None,
1825 0,
1826 )
1827 test_rest.test(
1828 "Delete VNFD id3",
1829 "DELETE",
1830 "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_ids[2]),
1831 headers_yaml,
1832 None,
1833 204,
1834 None,
1835 0,
1836 )
1837 test_rest.test(
1838 "Delete VNFD id4",
1839 "DELETE",
1840 "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_ids[3]),
1841 headers_yaml,
1842 None,
1843 404,
1844 r_header_yaml,
1845 "yaml",
1846 )
1847 test_rest.test(
1848 "Delete VNFD id4",
1849 "DELETE",
1850 "/vnfpkgm/v1/vnf_packages/{}?ADMIN".format(vnfd_ids[3]),
1851 headers_yaml,
1852 None,
1853 204,
1854 None,
1855 0,
1856 )
1857 # Get Public vnfds
1858 engine.test(
1859 "Get VNFD deleted id1",
1860 "GET",
1861 "/vnfpkgm/v1/vnf_packages/{}?ADMIN".format(vnfd_ids[0]),
1862 headers_json,
1863 None,
1864 404,
1865 r_header_json,
1866 "json",
1867 )
1868 engine.test(
1869 "Get VNFD deleted id2",
1870 "GET",
1871 "/vnfpkgm/v1/vnf_packages/{}?ADMIN".format(vnfd_ids[1]),
1872 headers_json,
1873 None,
1874 404,
1875 r_header_json,
1876 "json",
1877 )
1878 engine.test(
1879 "Get VNFD deleted id3",
1880 "GET",
1881 "/vnfpkgm/v1/vnf_packages/{}?ADMIN".format(vnfd_ids[2]),
1882 headers_json,
1883 None,
1884 404,
1885 r_header_json,
1886 "json",
1887 )
1888 engine.test(
1889 "Get VNFD deleted id4",
1890 "GET",
1891 "/vnfpkgm/v1/vnf_packages/{}?ADMIN".format(vnfd_ids[3]),
1892 headers_json,
1893 None,
1894 404,
1895 r_header_json,
1896 "json",
1897 )
1898
1899 engine.test(
1900 "Delete user U1",
1901 "DELETE",
1902 "/admin/v1/users/U1",
1903 headers_json,
1904 None,
1905 204,
1906 None,
1907 None,
1908 )
1909 engine.test(
1910 "Delete project Padmin",
1911 "DELETE",
1912 "/admin/v1/projects/Padmin",
1913 headers_json,
1914 None,
1915 204,
1916 None,
1917 None,
1918 )
1919 engine.test(
1920 "Delete project P2",
1921 "DELETE",
1922 "/admin/v1/projects/P2",
1923 headers_json,
1924 None,
1925 204,
1926 None,
1927 None,
1928 )
1929 engine.test(
1930 "Delete project P3",
1931 "DELETE",
1932 "/admin/v1/projects/P3",
1933 headers_json,
1934 None,
1935 204,
1936 None,
1937 None,
1938 )
1939
1940
1941 class TestFakeVim:
1942 description = "Creates/edit/delete fake VIMs and SDN controllers"
1943
1944 def __init__(self):
1945 self.vim = {
1946 "schema_version": "1.0",
1947 "schema_type": "No idea",
1948 "name": "myVim",
1949 "description": "Descriptor name",
1950 "vim_type": "openstack",
1951 "vim_url": "http://localhost:/vim",
1952 "vim_tenant_name": "vimTenant",
1953 "vim_user": "user",
1954 "vim_password": "password",
1955 "config": {"config_param": 1},
1956 }
1957 self.sdn = {
1958 "name": "sdn-name",
1959 "description": "sdn-description",
1960 "dpid": "50:50:52:54:00:94:21:21",
1961 "ip": "192.168.15.17",
1962 "port": 8080,
1963 "type": "opendaylight",
1964 "version": "3.5.6",
1965 "user": "user",
1966 "password": "passwd",
1967 }
1968 self.port_mapping = [
1969 {
1970 "compute_node": "compute node 1",
1971 "ports": [
1972 {
1973 "pci": "0000:81:00.0",
1974 "switch_port": "port-2/1",
1975 "switch_mac": "52:54:00:94:21:21",
1976 },
1977 {
1978 "pci": "0000:81:00.1",
1979 "switch_port": "port-2/2",
1980 "switch_mac": "52:54:00:94:21:22",
1981 },
1982 ],
1983 },
1984 {
1985 "compute_node": "compute node 2",
1986 "ports": [
1987 {
1988 "pci": "0000:81:00.0",
1989 "switch_port": "port-2/3",
1990 "switch_mac": "52:54:00:94:21:23",
1991 },
1992 {
1993 "pci": "0000:81:00.1",
1994 "switch_port": "port-2/4",
1995 "switch_mac": "52:54:00:94:21:24",
1996 },
1997 ],
1998 },
1999 ]
2000
2001 def run(self, engine, test_osm, manual_check, test_params=None):
2002
2003 vim_bad = self.vim.copy()
2004 vim_bad.pop("name")
2005
2006 engine.set_test_name("FakeVim")
2007 engine.get_autorization()
2008 engine.test(
2009 "Create VIM",
2010 "POST",
2011 "/admin/v1/vim_accounts",
2012 headers_json,
2013 self.vim,
2014 (201, 202),
2015 {"Location": "/admin/v1/vim_accounts/", "Content-Type": "application/json"},
2016 "json",
2017 )
2018 vim_id = engine.last_id
2019 engine.test(
2020 "Create VIM without name, bad schema",
2021 "POST",
2022 "/admin/v1/vim_accounts",
2023 headers_json,
2024 vim_bad,
2025 422,
2026 None,
2027 headers_json,
2028 )
2029 engine.test(
2030 "Create VIM name repeated",
2031 "POST",
2032 "/admin/v1/vim_accounts",
2033 headers_json,
2034 self.vim,
2035 409,
2036 None,
2037 headers_json,
2038 )
2039 engine.test(
2040 "Show VIMs",
2041 "GET",
2042 "/admin/v1/vim_accounts",
2043 headers_yaml,
2044 None,
2045 200,
2046 r_header_yaml,
2047 "yaml",
2048 )
2049 engine.test(
2050 "Show VIM",
2051 "GET",
2052 "/admin/v1/vim_accounts/{}".format(vim_id),
2053 headers_yaml,
2054 None,
2055 200,
2056 r_header_yaml,
2057 "yaml",
2058 )
2059 if not test_osm:
2060 # delete with FORCE
2061 engine.test(
2062 "Delete VIM",
2063 "DELETE",
2064 "/admin/v1/vim_accounts/{}?FORCE=True".format(vim_id),
2065 headers_yaml,
2066 None,
2067 202,
2068 None,
2069 0,
2070 )
2071 engine.test(
2072 "Check VIM is deleted",
2073 "GET",
2074 "/admin/v1/vim_accounts/{}".format(vim_id),
2075 headers_yaml,
2076 None,
2077 404,
2078 r_header_yaml,
2079 "yaml",
2080 )
2081 else:
2082 # delete and wait until is really deleted
2083 engine.test(
2084 "Delete VIM",
2085 "DELETE",
2086 "/admin/v1/vim_accounts/{}".format(vim_id),
2087 headers_yaml,
2088 None,
2089 202,
2090 None,
2091 0,
2092 )
2093 engine.wait_until_delete(
2094 "/admin/v1/vim_accounts/{}".format(vim_id), timeout
2095 )
2096
2097
2098 class TestVIMSDN(TestFakeVim):
2099 description = "Creates VIM with SDN editing SDN controllers and port_mapping"
2100
2101 def __init__(self):
2102 TestFakeVim.__init__(self)
2103 self.wim = {
2104 "schema_version": "1.0",
2105 "schema_type": "No idea",
2106 "name": "myWim",
2107 "description": "Descriptor name",
2108 "wim_type": "odl",
2109 "wim_url": "http://localhost:/wim",
2110 "user": "user",
2111 "password": "password",
2112 "config": {"config_param": 1},
2113 }
2114
2115 def run(self, engine, test_osm, manual_check, test_params=None):
2116 engine.set_test_name("VimSdn")
2117 engine.get_autorization()
2118 # Added SDN
2119 engine.test(
2120 "Create SDN",
2121 "POST",
2122 "/admin/v1/sdns",
2123 headers_json,
2124 self.sdn,
2125 (201, 202),
2126 {"Location": "/admin/v1/sdns/", "Content-Type": "application/json"},
2127 "json",
2128 )
2129 sdnc_id = engine.last_id
2130 # sleep(5)
2131 # Edit SDN
2132 engine.test(
2133 "Edit SDN",
2134 "PATCH",
2135 "/admin/v1/sdns/{}".format(sdnc_id),
2136 headers_json,
2137 {"name": "new_sdn_name"},
2138 (202, 204),
2139 None,
2140 None,
2141 )
2142 # sleep(5)
2143 # VIM with SDN
2144 self.vim["config"]["sdn-controller"] = sdnc_id
2145 self.vim["config"]["sdn-port-mapping"] = self.port_mapping
2146 engine.test(
2147 "Create VIM",
2148 "POST",
2149 "/admin/v1/vim_accounts",
2150 headers_json,
2151 self.vim,
2152 (200, 202, 201),
2153 {"Location": "/admin/v1/vim_accounts/", "Content-Type": "application/json"},
2154 "json",
2155 ),
2156
2157 vim_id = engine.last_id
2158 self.port_mapping[0]["compute_node"] = "compute node XX"
2159 engine.test(
2160 "Edit VIM change port-mapping",
2161 "PUT",
2162 "/admin/v1/vim_accounts/{}".format(vim_id),
2163 headers_json,
2164 {"config": {"sdn-port-mapping": self.port_mapping}},
2165 (202, 204),
2166 None,
2167 None,
2168 )
2169 engine.test(
2170 "Edit VIM remove port-mapping",
2171 "PUT",
2172 "/admin/v1/vim_accounts/{}".format(vim_id),
2173 headers_json,
2174 {"config": {"sdn-port-mapping": None}},
2175 (202, 204),
2176 None,
2177 None,
2178 )
2179
2180 engine.test(
2181 "Create WIM",
2182 "POST",
2183 "/admin/v1/wim_accounts",
2184 headers_json,
2185 self.wim,
2186 (200, 202, 201),
2187 {"Location": "/admin/v1/wim_accounts/", "Content-Type": "application/json"},
2188 "json",
2189 ),
2190 wim_id = engine.last_id
2191
2192 if not test_osm:
2193 # delete with FORCE
2194 engine.test(
2195 "Delete VIM remove port-mapping",
2196 "DELETE",
2197 "/admin/v1/vim_accounts/{}?FORCE=True".format(vim_id),
2198 headers_json,
2199 None,
2200 202,
2201 None,
2202 0,
2203 )
2204 engine.test(
2205 "Delete SDNC",
2206 "DELETE",
2207 "/admin/v1/sdns/{}?FORCE=True".format(sdnc_id),
2208 headers_json,
2209 None,
2210 202,
2211 None,
2212 0,
2213 )
2214
2215 engine.test(
2216 "Delete WIM",
2217 "DELETE",
2218 "/admin/v1/wim_accounts/{}?FORCE=True".format(wim_id),
2219 headers_json,
2220 None,
2221 202,
2222 None,
2223 0,
2224 )
2225 engine.test(
2226 "Check VIM is deleted",
2227 "GET",
2228 "/admin/v1/vim_accounts/{}".format(vim_id),
2229 headers_yaml,
2230 None,
2231 404,
2232 r_header_yaml,
2233 "yaml",
2234 )
2235 engine.test(
2236 "Check SDN is deleted",
2237 "GET",
2238 "/admin/v1/sdns/{}".format(sdnc_id),
2239 headers_yaml,
2240 None,
2241 404,
2242 r_header_yaml,
2243 "yaml",
2244 )
2245 engine.test(
2246 "Check WIM is deleted",
2247 "GET",
2248 "/admin/v1/wim_accounts/{}".format(wim_id),
2249 headers_yaml,
2250 None,
2251 404,
2252 r_header_yaml,
2253 "yaml",
2254 )
2255 else:
2256 if manual_check:
2257 input(
2258 "VIM, SDN, WIM has been deployed. Perform manual check and press enter to resume"
2259 )
2260 # delete and wait until is really deleted
2261 engine.test(
2262 "Delete VIM remove port-mapping",
2263 "DELETE",
2264 "/admin/v1/vim_accounts/{}".format(vim_id),
2265 headers_json,
2266 None,
2267 (202, 201, 204),
2268 None,
2269 0,
2270 )
2271 engine.test(
2272 "Delete SDN",
2273 "DELETE",
2274 "/admin/v1/sdns/{}".format(sdnc_id),
2275 headers_json,
2276 None,
2277 (202, 201, 204),
2278 None,
2279 0,
2280 )
2281 engine.test(
2282 "Delete VIM",
2283 "DELETE",
2284 "/admin/v1/wim_accounts/{}".format(wim_id),
2285 headers_json,
2286 None,
2287 (202, 201, 204),
2288 None,
2289 0,
2290 )
2291 engine.wait_until_delete(
2292 "/admin/v1/vim_accounts/{}".format(vim_id), timeout
2293 )
2294 engine.wait_until_delete("/admin/v1/sdns/{}".format(sdnc_id), timeout)
2295 engine.wait_until_delete(
2296 "/admin/v1/wim_accounts/{}".format(wim_id), timeout
2297 )
2298
2299
2300 class TestDeploy:
2301 description = "Base class for downloading descriptors from ETSI, onboard and deploy in real VIM"
2302
2303 def __init__(self):
2304 self.test_name = "DEPLOY"
2305 self.nsd_id = None
2306 self.vim_id = None
2307 self.ns_id = None
2308 self.vnfds_id = []
2309 self.descriptor_url = (
2310 "https://osm-download.etsi.org/ftp/osm-3.0-three/2nd-hackfest/packages/"
2311 )
2312 self.vnfd_filenames = ("cirros_vnf.tar.gz",)
2313 self.nsd_filename = "cirros_2vnf_ns.tar.gz"
2314 self.descriptor_edit = None
2315 self.uses_configuration = False
2316 self.users = {}
2317 self.passwords = {}
2318 self.commands = {}
2319 self.keys = {}
2320 self.timeout = 120
2321 self.qforce = ""
2322 self.ns_params = None
2323 self.vnfr_ip_list = {}
2324
2325 def create_descriptors(self, engine):
2326 temp_dir = os.path.dirname(os.path.abspath(__file__)) + "/temp/"
2327 if not os.path.exists(temp_dir):
2328 os.makedirs(temp_dir)
2329 for vnfd_index, vnfd_filename in enumerate(self.vnfd_filenames):
2330 if "/" in vnfd_filename:
2331 vnfd_filename_path = vnfd_filename
2332 if not os.path.exists(vnfd_filename_path):
2333 raise TestException(
2334 "File '{}' does not exist".format(vnfd_filename_path)
2335 )
2336 else:
2337 vnfd_filename_path = temp_dir + vnfd_filename
2338 if not os.path.exists(vnfd_filename_path):
2339 with open(vnfd_filename_path, "wb") as file:
2340 response = requests.get(self.descriptor_url + vnfd_filename)
2341 if response.status_code >= 300:
2342 raise TestException(
2343 "Error downloading descriptor from '{}': {}".format(
2344 self.descriptor_url + vnfd_filename,
2345 response.status_code,
2346 )
2347 )
2348 file.write(response.content)
2349 if vnfd_filename_path.endswith(".yaml"):
2350 headers = headers_yaml
2351 else:
2352 headers = headers_zip_yaml
2353 if randint(0, 1) == 0:
2354 # vnfd CREATE AND UPLOAD in one step:
2355 engine.test(
2356 "Onboard VNFD in one step",
2357 "POST",
2358 "/vnfpkgm/v1/vnf_packages_content" + self.qforce,
2359 headers,
2360 "@b" + vnfd_filename_path,
2361 201,
2362 r_headers_yaml_location_vnfd,
2363 "yaml",
2364 )
2365 self.vnfds_id.append(engine.last_id)
2366 else:
2367 # vnfd CREATE AND UPLOAD ZIP
2368 engine.test(
2369 "Onboard VNFD step 1",
2370 "POST",
2371 "/vnfpkgm/v1/vnf_packages",
2372 headers_json,
2373 None,
2374 201,
2375 {
2376 "Location": "/vnfpkgm/v1/vnf_packages/",
2377 "Content-Type": "application/json",
2378 },
2379 "json",
2380 )
2381 self.vnfds_id.append(engine.last_id)
2382 engine.test(
2383 "Onboard VNFD step 2 as ZIP",
2384 "PUT",
2385 "/vnfpkgm/v1/vnf_packages/<>/package_content" + self.qforce,
2386 headers,
2387 "@b" + vnfd_filename_path,
2388 204,
2389 None,
2390 0,
2391 )
2392
2393 if self.descriptor_edit:
2394 if "vnfd{}".format(vnfd_index) in self.descriptor_edit:
2395 # Modify VNFD
2396 engine.test(
2397 "Edit VNFD ",
2398 "PATCH",
2399 "/vnfpkgm/v1/vnf_packages/{}".format(self.vnfds_id[-1]),
2400 headers_yaml,
2401 self.descriptor_edit["vnfd{}".format(vnfd_index)],
2402 204,
2403 None,
2404 None,
2405 )
2406
2407 if "/" in self.nsd_filename:
2408 nsd_filename_path = self.nsd_filename
2409 if not os.path.exists(nsd_filename_path):
2410 raise TestException(
2411 "File '{}' does not exist".format(nsd_filename_path)
2412 )
2413 else:
2414 nsd_filename_path = temp_dir + self.nsd_filename
2415 if not os.path.exists(nsd_filename_path):
2416 with open(nsd_filename_path, "wb") as file:
2417 response = requests.get(self.descriptor_url + self.nsd_filename)
2418 if response.status_code >= 300:
2419 raise TestException(
2420 "Error downloading descriptor from '{}': {}".format(
2421 self.descriptor_url + self.nsd_filename,
2422 response.status_code,
2423 )
2424 )
2425 file.write(response.content)
2426 if nsd_filename_path.endswith(".yaml"):
2427 headers = headers_yaml
2428 else:
2429 headers = headers_zip_yaml
2430
2431 if randint(0, 1) == 0:
2432 # nsd CREATE AND UPLOAD in one step:
2433 engine.test(
2434 "Onboard NSD in one step",
2435 "POST",
2436 "/nsd/v1/ns_descriptors_content" + self.qforce,
2437 headers,
2438 "@b" + nsd_filename_path,
2439 201,
2440 r_headers_yaml_location_nsd,
2441 yaml,
2442 )
2443 self.nsd_id = engine.last_id
2444 else:
2445 # nsd CREATE AND UPLOAD ZIP
2446 engine.test(
2447 "Onboard NSD step 1",
2448 "POST",
2449 "/nsd/v1/ns_descriptors",
2450 headers_json,
2451 None,
2452 201,
2453 {
2454 "Location": "/nsd/v1/ns_descriptors/",
2455 "Content-Type": "application/json",
2456 },
2457 "json",
2458 )
2459 self.nsd_id = engine.last_id
2460 engine.test(
2461 "Onboard NSD step 2 as ZIP",
2462 "PUT",
2463 "/nsd/v1/ns_descriptors/<>/nsd_content" + self.qforce,
2464 headers,
2465 "@b" + nsd_filename_path,
2466 204,
2467 None,
2468 0,
2469 )
2470
2471 if self.descriptor_edit and "nsd" in self.descriptor_edit:
2472 # Modify NSD
2473 engine.test(
2474 "Edit NSD ",
2475 "PATCH",
2476 "/nsd/v1/ns_descriptors/{}".format(self.nsd_id),
2477 headers_yaml,
2478 self.descriptor_edit["nsd"],
2479 204,
2480 None,
2481 None,
2482 )
2483
2484 def delete_descriptors(self, engine):
2485 # delete descriptors
2486 engine.test(
2487 "Delete NSSD SOL005",
2488 "DELETE",
2489 "/nsd/v1/ns_descriptors/{}".format(self.nsd_id),
2490 headers_yaml,
2491 None,
2492 204,
2493 None,
2494 0,
2495 )
2496 for vnfd_id in self.vnfds_id:
2497 engine.test(
2498 "Delete VNFD SOL005",
2499 "DELETE",
2500 "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_id),
2501 headers_yaml,
2502 None,
2503 204,
2504 None,
2505 0,
2506 )
2507
2508 def instantiate(self, engine, ns_data):
2509 ns_data_text = yaml.safe_dump(ns_data, default_flow_style=True, width=256)
2510 # create NS Two steps
2511 r = engine.test(
2512 "Create NS step 1",
2513 "POST",
2514 "/nslcm/v1/ns_instances",
2515 headers_yaml,
2516 ns_data_text,
2517 (201, 202),
2518 {"Location": "nslcm/v1/ns_instances/", "Content-Type": "application/yaml"},
2519 "yaml",
2520 )
2521 if not r:
2522 return
2523 self.ns_id = engine.last_id
2524 engine.test(
2525 "Instantiate NS step 2",
2526 "POST",
2527 "/nslcm/v1/ns_instances/{}/instantiate".format(self.ns_id),
2528 headers_yaml,
2529 ns_data_text,
2530 (201, 202),
2531 r_headers_yaml_location_nslcmop,
2532 "yaml",
2533 )
2534 nslcmop_id = engine.last_id
2535
2536 if test_osm:
2537 # Wait until status is Ok
2538 timeout = timeout_configure if self.uses_configuration else timeout_deploy
2539 engine.wait_operation_ready("ns", nslcmop_id, timeout)
2540
2541 def terminate(self, engine):
2542 # remove deployment
2543 if test_osm:
2544 engine.test(
2545 "Terminate NS",
2546 "POST",
2547 "/nslcm/v1/ns_instances/{}/terminate".format(self.ns_id),
2548 headers_yaml,
2549 None,
2550 (201, 202),
2551 r_headers_yaml_location_nslcmop,
2552 "yaml",
2553 )
2554 nslcmop2_id = engine.last_id
2555 # Wait until status is Ok
2556 engine.wait_operation_ready("ns", nslcmop2_id, timeout_deploy)
2557
2558 engine.test(
2559 "Delete NS",
2560 "DELETE",
2561 "/nslcm/v1/ns_instances/{}".format(self.ns_id),
2562 headers_yaml,
2563 None,
2564 204,
2565 None,
2566 0,
2567 )
2568 else:
2569 engine.test(
2570 "Delete NS with FORCE",
2571 "DELETE",
2572 "/nslcm/v1/ns_instances/{}?FORCE=True".format(self.ns_id),
2573 headers_yaml,
2574 None,
2575 204,
2576 None,
2577 0,
2578 )
2579
2580 # check all it is deleted
2581 engine.test(
2582 "Check NS is deleted",
2583 "GET",
2584 "/nslcm/v1/ns_instances/{}".format(self.ns_id),
2585 headers_yaml,
2586 None,
2587 404,
2588 None,
2589 "yaml",
2590 )
2591 r = engine.test(
2592 "Check NSLCMOPs are deleted",
2593 "GET",
2594 "/nslcm/v1/ns_lcm_op_occs?nsInstanceId={}".format(self.ns_id),
2595 headers_json,
2596 None,
2597 200,
2598 None,
2599 "json",
2600 )
2601 if not r:
2602 return
2603 nslcmops = r.json()
2604 if not isinstance(nslcmops, list) or nslcmops:
2605 raise TestException(
2606 "NS {} deleted but with ns_lcm_op_occ active: {}".format(
2607 self.ns_id, nslcmops
2608 )
2609 )
2610
2611 def test_ns(
2612 self,
2613 engine,
2614 test_osm,
2615 commands=None,
2616 users=None,
2617 passwds=None,
2618 keys=None,
2619 timeout=0,
2620 ):
2621
2622 r = engine.test(
2623 "GET VNFR IDs",
2624 "GET",
2625 "/nslcm/v1/ns_instances/{}".format(self.ns_id),
2626 headers_json,
2627 None,
2628 200,
2629 r_header_json,
2630 "json",
2631 )
2632 if not r:
2633 return
2634 ns_data = r.json()
2635
2636 vnfr_list = ns_data["constituent-vnfr-ref"]
2637 time = 0
2638 _commands = commands if commands is not None else self.commands
2639 _users = users if users is not None else self.users
2640 _passwds = passwds if passwds is not None else self.passwords
2641 _keys = keys if keys is not None else self.keys
2642 _timeout = timeout if timeout != 0 else self.timeout
2643
2644 # vnfr_list=[d8272263-6bd3-4680-84ca-6a4be23b3f2d, 88b22e2f-994a-4b61-94fd-4a3c90de3dc4]
2645 for vnfr_id in vnfr_list:
2646 r = engine.test(
2647 "Get VNFR to get IP_ADDRESS",
2648 "GET",
2649 "/nslcm/v1/vnfrs/{}".format(vnfr_id),
2650 headers_json,
2651 None,
2652 200,
2653 r_header_json,
2654 "json",
2655 )
2656 if not r:
2657 continue
2658 vnfr_data = r.json()
2659
2660 vnf_index = str(vnfr_data["member-vnf-index-ref"])
2661
2662 ip_address = self.get_vnfr_ip(engine, vnf_index)
2663 description = "Exec command='{}' at VNFR={} IP={}".format(
2664 _commands.get(vnf_index)[0], vnf_index, ip_address
2665 )
2666 engine.step += 1
2667 test_description = "{}{} {}".format(
2668 engine.test_name, engine.step, description
2669 )
2670 logger.warning(test_description)
2671 while _timeout >= time:
2672 result, message = self.do_checks(
2673 [ip_address],
2674 vnf_index=vnfr_data["member-vnf-index-ref"],
2675 commands=_commands.get(vnf_index),
2676 user=_users.get(vnf_index),
2677 passwd=_passwds.get(vnf_index),
2678 key=_keys.get(vnf_index),
2679 )
2680 if result == 1:
2681 engine.passed_tests += 1
2682 logger.debug(message)
2683 break
2684 elif result == 0:
2685 time += 20
2686 sleep(20)
2687 elif result == -1:
2688 engine.failed_tests += 1
2689 logger.error(message)
2690 break
2691 else:
2692 time -= 20
2693 engine.failed_tests += 1
2694 logger.error(message)
2695 else:
2696 engine.failed_tests += 1
2697 logger.error(
2698 "VNFR {} has not mgmt address. Check failed".format(vnf_index)
2699 )
2700
2701 def do_checks(self, ip, vnf_index, commands=[], user=None, passwd=None, key=None):
2702 try:
2703 import urllib3
2704 from pssh.clients import ParallelSSHClient
2705 from pssh.utils import load_private_key
2706 from ssh2 import exceptions as ssh2Exception
2707 except ImportError as e:
2708 logger.critical(
2709 "Package <pssh> or/and <urllib3> is not installed. Please add them with 'pip3 install "
2710 "parallel-ssh urllib3': {}".format(e)
2711 )
2712 return -1, "install needed packages 'pip3 install parallel-ssh urllib3'"
2713 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
2714 try:
2715 p_host = os.environ.get("PROXY_HOST")
2716 p_user = os.environ.get("PROXY_USER")
2717 p_password = os.environ.get("PROXY_PASSWD")
2718
2719 if key:
2720 pkey = load_private_key(key)
2721 else:
2722 pkey = None
2723
2724 client = ParallelSSHClient(
2725 ip,
2726 user=user,
2727 password=passwd,
2728 pkey=pkey,
2729 proxy_host=p_host,
2730 proxy_user=p_user,
2731 proxy_password=p_password,
2732 timeout=10,
2733 num_retries=0,
2734 )
2735 for cmd in commands:
2736 output = client.run_command(cmd)
2737 client.join(output)
2738 if output[ip[0]].exit_code:
2739 return -1, "VNFR {} command '{}' returns error: '{}'".format(
2740 ip[0], cmd, "\n".join(output[ip[0]].stderr)
2741 )
2742 else:
2743 return 1, "VNFR {} command '{}' successful".format(ip[0], cmd)
2744 except (
2745 ssh2Exception.ChannelFailure,
2746 ssh2Exception.SocketDisconnectError,
2747 ssh2Exception.SocketTimeout,
2748 ssh2Exception.SocketRecvError,
2749 ) as e:
2750 return 0, "Timeout accessing the VNFR {}: {}".format(ip[0], str(e))
2751 except Exception as e:
2752 return -1, "ERROR checking the VNFR {}: {}".format(ip[0], str(e))
2753
2754 def additional_operations(self, engine, test_osm, manual_check):
2755 pass
2756
2757 def run(self, engine, test_osm, manual_check, test_params=None):
2758 engine.set_test_name(self.test_name)
2759 engine.get_autorization()
2760 nsname = os.environ.get("OSMNBITEST_NS_NAME", "OSMNBITEST")
2761 if test_params:
2762 if "vnfd-files" in test_params:
2763 self.vnfd_filenames = test_params["vnfd-files"].split(",")
2764 if "nsd-file" in test_params:
2765 self.nsd_filename = test_params["nsd-file"]
2766 if test_params.get("ns-name"):
2767 nsname = test_params["ns-name"]
2768 self.create_descriptors(engine)
2769
2770 # create real VIM if not exist
2771 self.vim_id = engine.get_create_vim(test_osm)
2772 ns_data = {
2773 "nsDescription": "default description",
2774 "nsName": nsname,
2775 "nsdId": self.nsd_id,
2776 "vimAccountId": self.vim_id,
2777 }
2778 if self.ns_params:
2779 ns_data.update(self.ns_params)
2780 if test_params and test_params.get("ns-config"):
2781 if isinstance(test_params["ns-config"], str):
2782 ns_data.update(yaml.load(test_params["ns-config"]), Loader=yaml.Loader)
2783 else:
2784 ns_data.update(test_params["ns-config"])
2785 self.instantiate(engine, ns_data)
2786
2787 if manual_check:
2788 input(
2789 "NS has been deployed. Perform manual check and press enter to resume"
2790 )
2791 if test_osm and self.commands:
2792 self.test_ns(engine, test_osm)
2793 self.additional_operations(engine, test_osm, manual_check)
2794 self.terminate(engine)
2795 self.delete_descriptors(engine)
2796
2797 def get_first_ip(self, ip_string):
2798 # When using a floating IP, the vnfr_data['ip-address'] contains a semicolon-separated list of IP:s.
2799 first_ip = ip_string.split(";")[0] if ip_string else ""
2800 return first_ip
2801
2802 def get_vnfr_ip(self, engine, vnfr_index_wanted):
2803 # If the IP address list has been obtained before, it has been stored in 'vnfr_ip_list'
2804 ip = self.vnfr_ip_list.get(vnfr_index_wanted, "")
2805 if ip:
2806 return self.get_first_ip(ip)
2807 r = engine.test(
2808 "Get VNFR to get IP_ADDRESS",
2809 "GET",
2810 "/nslcm/v1/vnfrs?member-vnf-index-ref={}&nsr-id-ref={}".format(
2811 vnfr_index_wanted, self.ns_id
2812 ),
2813 headers_json,
2814 None,
2815 200,
2816 r_header_json,
2817 "json",
2818 )
2819 if not r:
2820 return ""
2821 vnfr_data = r.json()
2822 if not (vnfr_data and vnfr_data[0]):
2823 return ""
2824 # Store the IP (or list of IPs) in 'vnfr_ip_list'
2825 ip_list = vnfr_data[0].get("ip-address", "")
2826 if ip_list:
2827 self.vnfr_ip_list[vnfr_index_wanted] = ip_list
2828 ip = self.get_first_ip(ip_list)
2829 return ip
2830
2831
2832 class TestDeployHackfestCirros(TestDeploy):
2833 description = "Load and deploy Hackfest cirros_2vnf_ns example"
2834
2835 def __init__(self):
2836 super().__init__()
2837 self.test_name = "CIRROS"
2838 self.vnfd_filenames = ("cirros_vnf.tar.gz",)
2839 self.nsd_filename = "cirros_2vnf_ns.tar.gz"
2840 self.commands = {
2841 "1": [
2842 "ls -lrt",
2843 ],
2844 "2": [
2845 "ls -lrt",
2846 ],
2847 }
2848 self.users = {"1": "cirros", "2": "cirros"}
2849 self.passwords = {"1": "cubswin:)", "2": "cubswin:)"}
2850
2851 def terminate(self, engine):
2852 # Make a delete in one step, overriding the normal two step of TestDeploy that launched terminate and delete
2853 if test_osm:
2854 engine.test(
2855 "Terminate and delete NS in one step",
2856 "DELETE",
2857 "/nslcm/v1/ns_instances_content/{}".format(self.ns_id),
2858 headers_yaml,
2859 None,
2860 202,
2861 None,
2862 "yaml",
2863 )
2864
2865 engine.wait_until_delete(
2866 "/nslcm/v1/ns_instances/{}".format(self.ns_id), timeout_deploy
2867 )
2868 else:
2869 engine.test(
2870 "Delete NS with FORCE",
2871 "DELETE",
2872 "/nslcm/v1/ns_instances/{}?FORCE=True".format(self.ns_id),
2873 headers_yaml,
2874 None,
2875 204,
2876 None,
2877 0,
2878 )
2879
2880 # check all it is deleted
2881 engine.test(
2882 "Check NS is deleted",
2883 "GET",
2884 "/nslcm/v1/ns_instances/{}".format(self.ns_id),
2885 headers_yaml,
2886 None,
2887 404,
2888 None,
2889 "yaml",
2890 )
2891 r = engine.test(
2892 "Check NSLCMOPs are deleted",
2893 "GET",
2894 "/nslcm/v1/ns_lcm_op_occs?nsInstanceId={}".format(self.ns_id),
2895 headers_json,
2896 None,
2897 200,
2898 None,
2899 "json",
2900 )
2901 if not r:
2902 return
2903 nslcmops = r.json()
2904 if not isinstance(nslcmops, list) or nslcmops:
2905 raise TestException(
2906 "NS {} deleted but with ns_lcm_op_occ active: {}".format(
2907 self.ns_id, nslcmops
2908 )
2909 )
2910
2911
2912 class TestDeployHackfest1(TestDeploy):
2913 description = "Load and deploy Hackfest_1_vnfd example"
2914
2915 def __init__(self):
2916 super().__init__()
2917 self.test_name = "HACKFEST1-"
2918 self.vnfd_filenames = ("hackfest_1_vnfd.tar.gz",)
2919 self.nsd_filename = "hackfest_1_nsd.tar.gz"
2920 # self.commands = {'1': ['ls -lrt', ], '2': ['ls -lrt', ]}
2921 # self.users = {'1': "cirros", '2': "cirros"}
2922 # self.passwords = {'1': "cubswin:)", '2': "cubswin:)"}
2923
2924
2925 class TestDeployHackfestCirrosScaling(TestDeploy):
2926 description = (
2927 "Load and deploy Hackfest cirros_2vnf_ns example with scaling modifications"
2928 )
2929
2930 def __init__(self):
2931 super().__init__()
2932 self.test_name = "CIRROS-SCALE"
2933 self.vnfd_filenames = ("cirros_vnf.tar.gz",)
2934 self.nsd_filename = "cirros_2vnf_ns.tar.gz"
2935 # Modify VNFD to add scaling and count=2
2936 self.descriptor_edit = {
2937 "vnfd0": {
2938 "vdu": {"$id: 'cirros_vnfd-VM'": {"count": 2}},
2939 "scaling-group-descriptor": [
2940 {
2941 "name": "scale_cirros",
2942 "max-instance-count": 2,
2943 "vdu": [{"vdu-id-ref": "cirros_vnfd-VM", "count": 2}],
2944 }
2945 ],
2946 }
2947 }
2948
2949 def additional_operations(self, engine, test_osm, manual_check):
2950 if not test_osm:
2951 return
2952 # 2 perform scale out twice
2953 payload = (
2954 "{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_OUT, scaleByStepData: "
2955 '{scaling-group-descriptor: scale_cirros, member-vnf-index: "1"}}}'
2956 )
2957 for i in range(0, 2):
2958 engine.test(
2959 "Execute scale action over NS",
2960 "POST",
2961 "/nslcm/v1/ns_instances/{}/scale".format(self.ns_id),
2962 headers_yaml,
2963 payload,
2964 (201, 202),
2965 r_headers_yaml_location_nslcmop,
2966 "yaml",
2967 )
2968 nslcmop2_scale_out = engine.last_id
2969 engine.wait_operation_ready("ns", nslcmop2_scale_out, timeout_deploy)
2970 if manual_check:
2971 input("NS scale out done. Check that two more vdus are there")
2972 # TODO check automatic
2973
2974 # 2 perform scale in
2975 payload = (
2976 "{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_IN, scaleByStepData: "
2977 '{scaling-group-descriptor: scale_cirros, member-vnf-index: "1"}}}'
2978 )
2979 for i in range(0, 2):
2980 engine.test(
2981 "Execute scale IN action over NS",
2982 "POST",
2983 "/nslcm/v1/ns_instances/{}/scale".format(self.ns_id),
2984 headers_yaml,
2985 payload,
2986 (201, 202),
2987 r_headers_yaml_location_nslcmop,
2988 "yaml",
2989 )
2990 nslcmop2_scale_in = engine.last_id
2991 engine.wait_operation_ready("ns", nslcmop2_scale_in, timeout_deploy)
2992 if manual_check:
2993 input("NS scale in done. Check that two less vdus are there")
2994 # TODO check automatic
2995
2996 # perform scale in that must fail as reached limit
2997 engine.test(
2998 "Execute scale IN out of limit action over NS",
2999 "POST",
3000 "/nslcm/v1/ns_instances/{}/scale".format(self.ns_id),
3001 headers_yaml,
3002 payload,
3003 (201, 202),
3004 r_headers_yaml_location_nslcmop,
3005 "yaml",
3006 )
3007 nslcmop2_scale_in = engine.last_id
3008 engine.wait_operation_ready(
3009 "ns", nslcmop2_scale_in, timeout_deploy, expected_fail=True
3010 )
3011
3012
3013 class TestDeployIpMac(TestDeploy):
3014 description = "Load and deploy descriptor examples setting mac, ip address at descriptor and instantiate params"
3015
3016 def __init__(self):
3017 super().__init__()
3018 self.test_name = "SetIpMac"
3019 self.vnfd_filenames = (
3020 "vnfd_2vdu_set_ip_mac2.yaml",
3021 "vnfd_2vdu_set_ip_mac.yaml",
3022 )
3023 self.nsd_filename = "scenario_2vdu_set_ip_mac.yaml"
3024 self.descriptor_url = "https://osm.etsi.org/gitweb/?p=osm/RO.git;a=blob_plain;f=test/RO_tests/v3_2vdu_set_ip_mac/"
3025 self.commands = {
3026 "1": [
3027 "ls -lrt",
3028 ],
3029 "2": [
3030 "ls -lrt",
3031 ],
3032 }
3033 self.users = {"1": "osm", "2": "osm"}
3034 self.passwords = {"1": "osm4u", "2": "osm4u"}
3035 self.timeout = 360
3036
3037 def run(self, engine, test_osm, manual_check, test_params=None):
3038 # super().run(engine, test_osm, manual_check, test_params)
3039 # run again setting IPs with instantiate parameters
3040 instantiation_params = {
3041 "vnf": [
3042 {
3043 "member-vnf-index": "1",
3044 "internal-vld": [
3045 {
3046 "name": "internal_vld1", # net_internal
3047 "ip-profile": {
3048 "ip-version": "ipv4",
3049 "subnet-address": "10.9.8.0/24",
3050 "dhcp-params": {
3051 "count": 100,
3052 "start-address": "10.9.8.100",
3053 },
3054 },
3055 "internal-connection-point": [
3056 {
3057 "id-ref": "eth2",
3058 "ip-address": "10.9.8.2",
3059 },
3060 {
3061 "id-ref": "eth3",
3062 "ip-address": "10.9.8.3",
3063 },
3064 ],
3065 },
3066 ],
3067 "vdu": [
3068 {
3069 "id": "VM1",
3070 "interface": [
3071 # {
3072 # "name": "iface11",
3073 # "floating-ip-required": True,
3074 # },
3075 {"name": "iface13", "mac-address": "52:33:44:55:66:13"},
3076 ],
3077 },
3078 {
3079 "id": "VM2",
3080 "interface": [
3081 {
3082 "name": "iface21",
3083 "ip-address": "10.31.31.22",
3084 "mac-address": "52:33:44:55:66:21",
3085 },
3086 ],
3087 },
3088 ],
3089 },
3090 ]
3091 }
3092
3093 super().run(
3094 engine,
3095 test_osm,
3096 manual_check,
3097 test_params={"ns-config": instantiation_params},
3098 )
3099
3100
3101 class TestDeployHackfest4(TestDeploy):
3102 description = "Load and deploy Hackfest 4 example."
3103
3104 def __init__(self):
3105 super().__init__()
3106 self.test_name = "HACKFEST4-"
3107 self.vnfd_filenames = ("hackfest_4_vnfd.tar.gz",)
3108 self.nsd_filename = "hackfest_4_nsd.tar.gz"
3109 self.uses_configuration = True
3110 self.commands = {
3111 "1": [
3112 "ls -lrt",
3113 ],
3114 "2": [
3115 "ls -lrt",
3116 ],
3117 }
3118 self.users = {"1": "ubuntu", "2": "ubuntu"}
3119 self.passwords = {"1": "osm4u", "2": "osm4u"}
3120 # Modify VNFD to add scaling
3121 # self.descriptor_edit = {
3122 # "vnfd0": {
3123 # 'vnf-configuration': {
3124 # 'config-primitive': [{
3125 # 'name': 'touch',
3126 # 'parameter': [{
3127 # 'name': 'filename',
3128 # 'data-type': 'STRING',
3129 # 'default-value': '/home/ubuntu/touched'
3130 # }]
3131 # }]
3132 # },
3133 # 'scaling-group-descriptor': [{
3134 # 'name': 'scale_dataVM',
3135 # 'scaling-policy': [{
3136 # 'threshold-time': 0,
3137 # 'name': 'auto_cpu_util_above_threshold',
3138 # 'scaling-type': 'automatic',
3139 # 'scaling-criteria': [{
3140 # 'name': 'cpu_util_above_threshold',
3141 # 'vnf-monitoring-param-ref': 'all_aaa_cpu_util',
3142 # 'scale-out-relational-operation': 'GE',
3143 # 'scale-in-threshold': 15,
3144 # 'scale-out-threshold': 60,
3145 # 'scale-in-relational-operation': 'LE'
3146 # }],
3147 # 'cooldown-time': 60
3148 # }],
3149 # 'max-instance-count': 10,
3150 # 'scaling-config-action': [
3151 # {'vnf-config-primitive-name-ref': 'touch',
3152 # 'trigger': 'post-scale-out'},
3153 # {'vnf-config-primitive-name-ref': 'touch',
3154 # 'trigger': 'pre-scale-in'}
3155 # ],
3156 # 'vdu': [{
3157 # 'vdu-id-ref': 'dataVM',
3158 # 'count': 1
3159 # }]
3160 # }]
3161 # }
3162 # }
3163
3164
3165 class TestDeployHackfest3Charmed(TestDeploy):
3166 description = "Load and deploy Hackfest 3charmed_ns example"
3167
3168 def __init__(self):
3169 super().__init__()
3170 self.test_name = "HACKFEST3-"
3171 self.vnfd_filenames = ("hackfest_3charmed_vnfd.tar.gz",)
3172 self.nsd_filename = "hackfest_3charmed_nsd.tar.gz"
3173 self.uses_configuration = True
3174 self.commands = {
3175 "1": ["ls -lrt /home/ubuntu/first-touch"],
3176 "2": ["ls -lrt /home/ubuntu/first-touch"],
3177 }
3178 self.users = {"1": "ubuntu", "2": "ubuntu"}
3179 self.passwords = {"1": "osm4u", "2": "osm4u"}
3180 self.descriptor_edit = {
3181 "vnfd0": yaml.safe_load(
3182 """
3183 vnf-configuration:
3184 terminate-config-primitive:
3185 - seq: '1'
3186 name: touch
3187 parameter:
3188 - name: filename
3189 value: '/home/ubuntu/last-touch1'
3190 - seq: '3'
3191 name: touch
3192 parameter:
3193 - name: filename
3194 value: '/home/ubuntu/last-touch3'
3195 - seq: '2'
3196 name: touch
3197 parameter:
3198 - name: filename
3199 value: '/home/ubuntu/last-touch2'
3200 """
3201 )
3202 }
3203
3204 def additional_operations(self, engine, test_osm, manual_check):
3205 if not test_osm:
3206 return
3207 # 1 perform action
3208 vnfr_index_selected = "2"
3209 payload = '{member_vnf_index: "2", primitive: touch, primitive_params: { filename: /home/ubuntu/OSMTESTNBI }}'
3210 engine.test(
3211 "Exec service primitive over NS",
3212 "POST",
3213 "/nslcm/v1/ns_instances/{}/action".format(self.ns_id),
3214 headers_yaml,
3215 payload,
3216 (201, 202),
3217 r_headers_yaml_location_nslcmop,
3218 "yaml",
3219 )
3220 nslcmop2_action = engine.last_id
3221 # Wait until status is Ok
3222 engine.wait_operation_ready("ns", nslcmop2_action, timeout_deploy)
3223 vnfr_ip = self.get_vnfr_ip(engine, vnfr_index_selected)
3224 if manual_check:
3225 input(
3226 "NS service primitive has been executed."
3227 "Check that file /home/ubuntu/OSMTESTNBI is present at {}".format(
3228 vnfr_ip
3229 )
3230 )
3231 if test_osm:
3232 commands = {
3233 "1": [""],
3234 "2": [
3235 "ls -lrt /home/ubuntu/OSMTESTNBI",
3236 ],
3237 }
3238 self.test_ns(engine, test_osm, commands=commands)
3239
3240 # # 2 perform scale out
3241 # payload = '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_OUT, scaleByStepData: ' \
3242 # '{scaling-group-descriptor: scale_dataVM, member-vnf-index: "1"}}}'
3243 # engine.test("Execute scale action over NS", "POST",
3244 # "/nslcm/v1/ns_instances/{}/scale".format(self.ns_id), headers_yaml, payload,
3245 # (201, 202), r_headers_yaml_location_nslcmop, "yaml")
3246 # nslcmop2_scale_out = engine.last_id
3247 # engine.wait_operation_ready("ns", nslcmop2_scale_out, timeout_deploy)
3248 # if manual_check:
3249 # input('NS scale out done. Check that file /home/ubuntu/touched is present and new VM is created')
3250 # # TODO check automatic
3251 #
3252 # # 2 perform scale in
3253 # payload = '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_IN, scaleByStepData: ' \
3254 # '{scaling-group-descriptor: scale_dataVM, member-vnf-index: "1"}}}'
3255 # engine.test("Execute scale action over NS", "POST",
3256 # "/nslcm/v1/ns_instances/{}/scale".format(self.ns_id), headers_yaml, payload,
3257 # (201, 202), r_headers_yaml_location_nslcmop, "yaml")
3258 # nslcmop2_scale_in = engine.last_id
3259 # engine.wait_operation_ready("ns", nslcmop2_scale_in, timeout_deploy)
3260 # if manual_check:
3261 # input('NS scale in done. Check that file /home/ubuntu/touched is updated and new VM is deleted')
3262 # # TODO check automatic
3263
3264
3265 class TestDeployHackfest3Charmed2(TestDeployHackfest3Charmed):
3266 description = (
3267 "Load and deploy Hackfest 3charmed_ns example modified version of descriptors to have dots in "
3268 "ids and member-vnf-index."
3269 )
3270
3271 def __init__(self):
3272 super().__init__()
3273 self.test_name = "HACKFEST3v2-"
3274 self.qforce = "?FORCE=True"
3275 self.descriptor_edit = {
3276 "vnfd0": {
3277 "vdu": {
3278 "$[0]": {
3279 "interface": {
3280 "$[0]": {"external-connection-point-ref": "pdu-mgmt"}
3281 }
3282 },
3283 "$[1]": None,
3284 },
3285 "vnf-configuration": None,
3286 "connection-point": {
3287 "$[0]": {
3288 "id": "pdu-mgmt",
3289 "name": "pdu-mgmt",
3290 "short-name": "pdu-mgmt",
3291 },
3292 "$[1]": None,
3293 },
3294 "mgmt-interface": {"cp": "pdu-mgmt"},
3295 "description": "A vnf single vdu to be used as PDU",
3296 "id": "vdu-as-pdu",
3297 "internal-vld": {
3298 "$[0]": {
3299 "id": "pdu_internal",
3300 "name": "pdu_internal",
3301 "internal-connection-point": {"$[1]": None},
3302 "short-name": "pdu_internal",
3303 "type": "ELAN",
3304 }
3305 },
3306 },
3307 # Modify NSD accordingly
3308 "nsd": {
3309 "constituent-vnfd": {
3310 "$[0]": {"vnfd-id-ref": "vdu-as-pdu"},
3311 "$[1]": None,
3312 },
3313 "description": "A nsd to deploy the vnf to act as as PDU",
3314 "id": "nsd-as-pdu",
3315 "name": "nsd-as-pdu",
3316 "short-name": "nsd-as-pdu",
3317 "vld": {
3318 "$[0]": {
3319 "id": "mgmt_pdu",
3320 "name": "mgmt_pdu",
3321 "short-name": "mgmt_pdu",
3322 "vnfd-connection-point-ref": {
3323 "$[0]": {
3324 "vnfd-connection-point-ref": "pdu-mgmt",
3325 "vnfd-id-ref": "vdu-as-pdu",
3326 },
3327 "$[1]": None,
3328 },
3329 "type": "ELAN",
3330 },
3331 "$[1]": None,
3332 },
3333 },
3334 }
3335
3336
3337 class TestDeployHackfest3Charmed3(TestDeployHackfest3Charmed):
3338 description = "Load and deploy Hackfest 3charmed_ns example modified version to test scaling and NS parameters"
3339
3340 def __init__(self):
3341 super().__init__()
3342 self.test_name = "HACKFEST3v3-"
3343 self.commands = {
3344 "1": ["ls -lrt /home/ubuntu/first-touch-1"],
3345 "2": ["ls -lrt /home/ubuntu/first-touch-2"],
3346 }
3347 self.descriptor_edit = {
3348 "vnfd0": yaml.load(
3349 """
3350 scaling-group-descriptor:
3351 - name: "scale_dataVM"
3352 max-instance-count: 10
3353 scaling-policy:
3354 - name: "auto_cpu_util_above_threshold"
3355 scaling-type: "automatic"
3356 threshold-time: 0
3357 cooldown-time: 60
3358 scaling-criteria:
3359 - name: "cpu_util_above_threshold"
3360 scale-in-threshold: 15
3361 scale-in-relational-operation: "LE"
3362 scale-out-threshold: 60
3363 scale-out-relational-operation: "GE"
3364 vnf-monitoring-param-ref: "monitor1"
3365 vdu:
3366 - vdu-id-ref: dataVM
3367 count: 1
3368 scaling-config-action:
3369 - trigger: post-scale-out
3370 vnf-config-primitive-name-ref: touch
3371 - trigger: pre-scale-in
3372 vnf-config-primitive-name-ref: touch
3373 vdu:
3374 "$id: dataVM":
3375 monitoring-param:
3376 - id: "dataVM_cpu_util"
3377 nfvi-metric: "cpu_utilization"
3378
3379 monitoring-param:
3380 - id: "monitor1"
3381 name: "monitor1"
3382 aggregation-type: AVERAGE
3383 vdu-monitoring-param:
3384 vdu-ref: "dataVM"
3385 vdu-monitoring-param-ref: "dataVM_cpu_util"
3386 vnf-configuration:
3387 initial-config-primitive:
3388 "$[1]":
3389 parameter:
3390 "$[0]":
3391 value: "<touch_filename>" # default-value: /home/ubuntu/first-touch
3392 config-primitive:
3393 "$[0]":
3394 parameter:
3395 "$[0]":
3396 default-value: "<touch_filename2>"
3397 """,
3398 Loader=yaml.Loader,
3399 )
3400 }
3401 self.ns_params = {
3402 "additionalParamsForVnf": [
3403 {
3404 "member-vnf-index": "1",
3405 "additionalParams": {
3406 "touch_filename": "/home/ubuntu/first-touch-1",
3407 "touch_filename2": "/home/ubuntu/second-touch-1",
3408 },
3409 },
3410 {
3411 "member-vnf-index": "2",
3412 "additionalParams": {
3413 "touch_filename": "/home/ubuntu/first-touch-2",
3414 "touch_filename2": "/home/ubuntu/second-touch-2",
3415 },
3416 },
3417 ]
3418 }
3419
3420 def additional_operations(self, engine, test_osm, manual_check):
3421 super().additional_operations(engine, test_osm, manual_check)
3422 if not test_osm:
3423 return
3424
3425 # 2 perform scale out
3426 payload = (
3427 "{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_OUT, scaleByStepData: "
3428 '{scaling-group-descriptor: scale_dataVM, member-vnf-index: "1"}}}'
3429 )
3430 engine.test(
3431 "Execute scale action over NS",
3432 "POST",
3433 "/nslcm/v1/ns_instances/{}/scale".format(self.ns_id),
3434 headers_yaml,
3435 payload,
3436 (201, 202),
3437 r_headers_yaml_location_nslcmop,
3438 "yaml",
3439 )
3440 nslcmop2_scale_out = engine.last_id
3441 engine.wait_operation_ready("ns", nslcmop2_scale_out, timeout_deploy)
3442 if manual_check:
3443 input(
3444 "NS scale out done. Check that file /home/ubuntu/second-touch-1 is present and new VM is created"
3445 )
3446 if test_osm:
3447 commands = {
3448 "1": [
3449 "ls -lrt /home/ubuntu/second-touch-1",
3450 ]
3451 }
3452 self.test_ns(engine, test_osm, commands=commands)
3453 # TODO check automatic connection to scaled VM
3454
3455 # 2 perform scale in
3456 payload = (
3457 "{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_IN, scaleByStepData: "
3458 '{scaling-group-descriptor: scale_dataVM, member-vnf-index: "1"}}}'
3459 )
3460 engine.test(
3461 "Execute scale action over NS",
3462 "POST",
3463 "/nslcm/v1/ns_instances/{}/scale".format(self.ns_id),
3464 headers_yaml,
3465 payload,
3466 (201, 202),
3467 r_headers_yaml_location_nslcmop,
3468 "yaml",
3469 )
3470 nslcmop2_scale_in = engine.last_id
3471 engine.wait_operation_ready("ns", nslcmop2_scale_in, timeout_deploy)
3472 if manual_check:
3473 input(
3474 "NS scale in done. Check that file /home/ubuntu/second-touch-1 is updated and new VM is deleted"
3475 )
3476 # TODO check automatic
3477
3478
3479 class TestDeploySimpleCharm(TestDeploy):
3480 description = "Deploy hackfest-4 hackfest_simplecharm example"
3481
3482 def __init__(self):
3483 super().__init__()
3484 self.test_name = "HACKFEST-SIMPLE"
3485 self.descriptor_url = (
3486 "https://osm-download.etsi.org/ftp/osm-4.0-four/4th-hackfest/packages/"
3487 )
3488 self.vnfd_filenames = ("hackfest_simplecharm_vnf.tar.gz",)
3489 self.nsd_filename = "hackfest_simplecharm_ns.tar.gz"
3490 self.uses_configuration = True
3491 self.commands = {
3492 "1": [""],
3493 "2": [
3494 "ls -lrt /home/ubuntu/first-touch",
3495 ],
3496 }
3497 self.users = {"1": "ubuntu", "2": "ubuntu"}
3498 self.passwords = {"1": "osm4u", "2": "osm4u"}
3499
3500
3501 class TestDeploySimpleCharm2(TestDeploySimpleCharm):
3502 description = (
3503 "Deploy hackfest-4 hackfest_simplecharm example changing naming to contain dots on ids and "
3504 "vnf-member-index"
3505 )
3506
3507 def __init__(self):
3508 super().__init__()
3509 self.test_name = "HACKFEST-SIMPLE2-"
3510 self.qforce = "?FORCE=True"
3511 self.descriptor_edit = {
3512 "vnfd0": {"id": "hackfest.simplecharm.vnf"},
3513 "nsd": {
3514 "id": "hackfest.simplecharm.ns",
3515 "constituent-vnfd": {
3516 "$[0]": {
3517 "vnfd-id-ref": "hackfest.simplecharm.vnf",
3518 "member-vnf-index": "$1",
3519 },
3520 "$[1]": {
3521 "vnfd-id-ref": "hackfest.simplecharm.vnf",
3522 "member-vnf-index": "$2",
3523 },
3524 },
3525 "vld": {
3526 "$[0]": {
3527 "vnfd-connection-point-ref": {
3528 "$[0]": {
3529 "member-vnf-index-ref": "$1",
3530 "vnfd-id-ref": "hackfest.simplecharm.vnf",
3531 },
3532 "$[1]": {
3533 "member-vnf-index-ref": "$2",
3534 "vnfd-id-ref": "hackfest.simplecharm.vnf",
3535 },
3536 },
3537 },
3538 "$[1]": {
3539 "vnfd-connection-point-ref": {
3540 "$[0]": {
3541 "member-vnf-index-ref": "$1",
3542 "vnfd-id-ref": "hackfest.simplecharm.vnf",
3543 },
3544 "$[1]": {
3545 "member-vnf-index-ref": "$2",
3546 "vnfd-id-ref": "hackfest.simplecharm.vnf",
3547 },
3548 },
3549 },
3550 },
3551 },
3552 }
3553
3554
3555 class TestDeploySingleVdu(TestDeployHackfest3Charmed):
3556 description = (
3557 "Generate a single VDU base on editing Hackfest3Charmed descriptors and deploy"
3558 )
3559
3560 def __init__(self):
3561 super().__init__()
3562 self.test_name = "SingleVDU"
3563 self.qforce = "?FORCE=True"
3564 self.descriptor_edit = {
3565 # Modify VNFD to remove one VDU
3566 "vnfd0": {
3567 "vdu": {
3568 "$[0]": {
3569 "interface": {
3570 "$[0]": {"external-connection-point-ref": "pdu-mgmt"}
3571 }
3572 },
3573 "$[1]": None,
3574 },
3575 "vnf-configuration": None,
3576 "connection-point": {
3577 "$[0]": {
3578 "id": "pdu-mgmt",
3579 "name": "pdu-mgmt",
3580 "short-name": "pdu-mgmt",
3581 },
3582 "$[1]": None,
3583 },
3584 "mgmt-interface": {"cp": "pdu-mgmt"},
3585 "description": "A vnf single vdu to be used as PDU",
3586 "id": "vdu-as-pdu",
3587 "internal-vld": {
3588 "$[0]": {
3589 "id": "pdu_internal",
3590 "name": "pdu_internal",
3591 "internal-connection-point": {"$[1]": None},
3592 "short-name": "pdu_internal",
3593 "type": "ELAN",
3594 }
3595 },
3596 },
3597 # Modify NSD accordingly
3598 "nsd": {
3599 "constituent-vnfd": {
3600 "$[0]": {"vnfd-id-ref": "vdu-as-pdu"},
3601 "$[1]": None,
3602 },
3603 "description": "A nsd to deploy the vnf to act as as PDU",
3604 "id": "nsd-as-pdu",
3605 "name": "nsd-as-pdu",
3606 "short-name": "nsd-as-pdu",
3607 "vld": {
3608 "$[0]": {
3609 "id": "mgmt_pdu",
3610 "name": "mgmt_pdu",
3611 "short-name": "mgmt_pdu",
3612 "vnfd-connection-point-ref": {
3613 "$[0]": {
3614 "vnfd-connection-point-ref": "pdu-mgmt",
3615 "vnfd-id-ref": "vdu-as-pdu",
3616 },
3617 "$[1]": None,
3618 },
3619 "type": "ELAN",
3620 },
3621 "$[1]": None,
3622 },
3623 },
3624 }
3625
3626
3627 class TestDeployHnfd(TestDeployHackfest3Charmed):
3628 description = (
3629 "Generate a HNFD base on editing Hackfest3Charmed descriptors and deploy"
3630 )
3631
3632 def __init__(self):
3633 super().__init__()
3634 self.test_name = "HNFD"
3635 self.pduDeploy = TestDeploySingleVdu()
3636 self.pdu_interface_0 = {}
3637 self.pdu_interface_1 = {}
3638
3639 self.pdu_id = None
3640 # self.vnf_to_pdu = """
3641 # vdu:
3642 # "$[0]":
3643 # pdu-type: PDU-TYPE-1
3644 # interface:
3645 # "$[0]":
3646 # name: mgmt-iface
3647 # "$[1]":
3648 # name: pdu-iface-internal
3649 # id: hfn1
3650 # description: HFND, one PDU + One VDU
3651 # name: hfn1
3652 # short-name: hfn1
3653 #
3654 # """
3655
3656 self.pdu_descriptor = {
3657 "name": "my-PDU",
3658 "type": "PDU-TYPE-1",
3659 "vim_accounts": "to-override",
3660 "interfaces": [
3661 {
3662 "name": "mgmt-iface",
3663 "mgmt": True,
3664 "type": "overlay",
3665 "ip-address": "to override",
3666 "mac-address": "mac_address",
3667 "vim-network-name": "mgmt",
3668 },
3669 {
3670 "name": "pdu-iface-internal",
3671 "mgmt": False,
3672 "type": "overlay",
3673 "ip-address": "to override",
3674 "mac-address": "mac_address",
3675 "vim-network-name": "pdu_internal", # OSMNBITEST-PDU-pdu_internal
3676 },
3677 ],
3678 }
3679 self.vnfd_filenames = (
3680 "hackfest_3charmed_vnfd.tar.gz",
3681 "hackfest_3charmed_vnfd.tar.gz",
3682 )
3683
3684 self.descriptor_edit = {
3685 "vnfd0": {
3686 "id": "hfnd1",
3687 "name": "hfn1",
3688 "short-name": "hfn1",
3689 "vdu": {
3690 "$[0]": {
3691 "pdu-type": "PDU-TYPE-1",
3692 "interface": {
3693 "$[0]": {"name": "mgmt-iface"},
3694 "$[1]": {"name": "pdu-iface-internal"},
3695 },
3696 }
3697 },
3698 },
3699 "nsd": {
3700 "constituent-vnfd": {"$[1]": {"vnfd-id-ref": "hfnd1"}},
3701