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


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