+ def test_ns(self, engine, test_osm, commands=None, users=None, passwds=None, keys=None, timeout=0):
+
+ n = 0
+ r = engine.test("TEST_NS{}".format(n), "GET VNFR_IDs", "GET",
+ "/nslcm/v1/ns_instances/{}".format(self.ns_id), headers_json, None,
+ 200, r_header_json, "json")
+ n += 1
+ ns_data = r.json()
+
+ vnfr_list = ns_data['constituent-vnfr-ref']
+ time = 0
+
+ for vnfr_id in vnfr_list:
+ self.total_tests += 1
+ r = engine.test("TEST_NS{}".format(n), "GET IP_ADDRESS OF VNFR", "GET",
+ "/nslcm/v1/vnfrs/{}".format(vnfr_id), headers_json, None,
+ 200, r_header_json, "json")
+ n += 1
+ vnfr_data = r.json()
+
+ if vnfr_data.get("ip-address"):
+ name = "TEST_NS{}".format(n)
+ description = "Run tests in VNFR with IP {}".format(vnfr_data['ip-address'])
+ n += 1
+ test_description = "Test {} {}".format(name, description)
+ logger.warning(test_description)
+ vnf_index = str(vnfr_data["member-vnf-index-ref"])
+ while timeout >= time:
+ result, message = self.do_checks([vnfr_data["ip-address"]],
+ vnf_index=vnfr_data["member-vnf-index-ref"],
+ commands=commands.get(vnf_index), user=users.get(vnf_index),
+ passwd=passwds.get(vnf_index), key=keys.get(vnf_index))
+ if result == 1:
+ logger.warning(message)
+ break
+ elif result == 0:
+ time += 20
+ sleep(20)
+ elif result == -1:
+ logger.critical(message)
+ break
+ else:
+ time -= 20
+ logger.critical(message)
+ else:
+ logger.critical("VNFR {} has not mgmt address. Check failed".format(vnfr_id))
+
+ def do_checks(self, ip, vnf_index, commands=[], user=None, passwd=None, key=None):
+ try:
+ import urllib3
+ from pssh.clients import ParallelSSHClient
+ from pssh.utils import load_private_key
+ from ssh2 import exceptions as ssh2Exception
+ except ImportError as e:
+ logger.critical("package <pssh> or/and <urllib3> is not installed. Please add it with 'pip3 install "
+ "parallel-ssh' and/or 'pip3 install urllib3': {}".format(e))
+ urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
+ try:
+ p_host = os.environ.get("PROXY_HOST")
+ p_user = os.environ.get("PROXY_USER")
+ p_password = os.environ.get("PROXY_PASSWD")
+
+ if key:
+ pkey = load_private_key(key)
+ else:
+ pkey = None
+
+ client = ParallelSSHClient(ip, user=user, password=passwd, pkey=pkey, proxy_host=p_host,
+ proxy_user=p_user, proxy_password=p_password, timeout=10, num_retries=0)
+ for cmd in commands:
+ output = client.run_command(cmd)
+ client.join(output)
+ if output[ip[0]].exit_code:
+ return -1, " VNFR {} could not be checked: {}".format(ip[0], output[ip[0]].stderr)
+ else:
+ self.passed_tests += 1
+ return 1, " Test successful"
+ except (ssh2Exception.ChannelFailure, ssh2Exception.SocketDisconnectError, ssh2Exception.SocketTimeout,
+ ssh2Exception.SocketRecvError) as e:
+ return 0, "Timeout accessing the VNFR {}: {}".format(ip[0], str(e))
+ except Exception as e:
+ return -1, "ERROR checking the VNFR {}: {}".format(ip[0], str(e))