4 from robot.api import logger
7 XNF_SIMULATOR_NAME = "xNF Simulator"
8 HV_VES_COLLECTOR_NAMESPACE="onap"
9 HV_VES_GROUP_ID="org.onap.dcaegen2.collectors.hv-ves"
10 SIMULATOR_IMAGE_NAME = HV_VES_COLLECTOR_NAMESPACE + "/" + HV_VES_GROUP_ID + ".hv-collector-xnf-simulator"
11 HV_VES_VERSION="1.1-SNAPSHOT"
12 SIMULATOR_IMAGE_FULL_NAME = os.getenv("DOCKER_REGISTRY_PREFIX") + SIMULATOR_IMAGE_NAME + ":" + HV_VES_VERSION
13 WORKSPACE_ENV = os.getenv("WORKSPACE")
14 certificates_dir_path = WORKSPACE_ENV + "/plans/dcaegen2-collectors-hv-ves/testsuites/ssl/"
15 collector_certs_lookup_dir = "/etc/ves-hv/"
16 ONE_SECOND_IN_NANOS = 10 ** 9
19 class XnfSimulatorLibrary:
21 def start_xnf_simulators(self, list_of_ports,
22 should_use_valid_certs=True,
23 should_disable_ssl=False,
24 should_connect_to_unencrypted_hv_ves=False):
25 logger.info("Creating " + str(len(list_of_ports)) + " xNF Simulator containers")
26 dockerClient = docker.from_env()
28 self.pullImageIfAbsent(dockerClient)
29 logger.info("Using image: " + SIMULATOR_IMAGE_FULL_NAME)
31 simulators_addresses = self.create_containers(dockerClient,
33 should_use_valid_certs,
35 should_connect_to_unencrypted_hv_ves)
37 self.assert_containers_startup_was_successful(dockerClient)
39 return simulators_addresses
41 def pullImageIfAbsent(self, dockerClient):
43 dockerClient.images.get(SIMULATOR_IMAGE_FULL_NAME)
45 logger.console("Image " + SIMULATOR_IMAGE_FULL_NAME + " will be pulled from repository. "
46 "This can take a while.")
47 dockerClient.images.pull(SIMULATOR_IMAGE_FULL_NAME)
49 def create_containers(self,
52 should_use_valid_certs,
54 should_connect_to_unencrypted_hv_ves):
55 simulators_addresses = []
56 for port in list_of_ports:
57 xnf = XnfSimulator(port, should_use_valid_certs, should_disable_ssl, should_connect_to_unencrypted_hv_ves)
58 container = self.run_simulator(dockerClient, xnf)
59 logger.info("Started container: " + container.name + " " + container.id)
60 simulators_addresses.append(container.name + ":" + xnf.port)
61 return simulators_addresses
63 def run_simulator(self, dockerClient, xnf):
64 xNF_startup_command = xnf.get_startup_command()
65 xNF_healthcheck_command = xnf.get_healthcheck_command()
67 logger.info("Startup command: " + str(xNF_startup_command))
68 logger.info("Healthcheck command: " + str(xNF_healthcheck_command))
69 return dockerClient.containers.run(SIMULATOR_IMAGE_FULL_NAME,
70 command=xNF_startup_command,
71 healthcheck=xNF_healthcheck_command,
73 network="ves-hv-default",
74 ports={port + "/tcp": port},
75 volumes=self.container_volumes(),
76 name=xnf.container_name_prefix + port)
78 def container_volumes(self):
79 return {certificates_dir_path: {"bind": collector_certs_lookup_dir, "mode": 'rw'}}
81 def assert_containers_startup_was_successful(self, dockerClient):
83 check_interval_in_seconds = 5
84 for _ in range(checks_amount):
85 sleep(check_interval_in_seconds)
86 all_containers_healthy = True
87 for container in self.get_simulators_list(dockerClient):
88 all_containers_healthy = all_containers_healthy and self.is_container_healthy(container)
89 if (all_containers_healthy):
91 raise ContainerException("One of xNF simulators containers did not pass the healthcheck.")
93 def is_container_healthy(self, container):
94 container_health = container.attrs['State']['Health']['Status']
95 return container_health == 'healthy' and container.status == 'running'
97 def stop_and_remove_all_xnf_simulators(self, suite_name):
98 dockerClient = docker.from_env()
99 for container in self.get_simulators_list(dockerClient):
100 logger.info("Stopping and removing container: " + container.id)
101 log_filename = WORKSPACE_ENV + "/archives/containers_logs/" + \
102 suite_name.split(".")[-1] + "_" + container.name + ".log"
103 file = open(log_filename, "w+")
104 file.write(container.logs())
110 def get_simulators_list(self, dockerClient):
111 return dockerClient.containers.list(filters={"ancestor": SIMULATOR_IMAGE_FULL_NAME}, all=True)
113 def send_messages(self, simulator_url, message_filepath):
114 logger.info("Reading message to simulator from: " + message_filepath)
116 file = open(message_filepath, "rb")
120 logger.info("POST at: " + simulator_url)
121 resp = HttpRequests.session_without_env().post(simulator_url, data=data, timeout=5)
122 HttpRequests.checkStatusCode(resp.status_code, XNF_SIMULATOR_NAME)
126 container_name_prefix = "ves-hv-collector-xnf-simulator"
130 should_use_valid_certs,
132 should_connect_to_unencrypted_hv_ves):
134 self.healthcheck_server_port = "6063"
135 cert_name_prefix = "" if should_use_valid_certs else "untrusted"
136 certificates_path_with_file_prefix = collector_certs_lookup_dir + cert_name_prefix
137 self.key_store_path = certificates_path_with_file_prefix + "client.p12"
138 self.trust_store_path = certificates_path_with_file_prefix + "trust.p12"
139 self.sec_store_passwd = "onaponap"
140 self.disable_ssl = should_disable_ssl
141 self.hv_collector_host = "unencrypted-ves-hv-collector" \
142 if should_connect_to_unencrypted_hv_ves else "ves-hv-collector"
144 def get_startup_command(self):
145 startup_command = ["--listen-port", self.port,
146 "--health-check-api-port", self.healthcheck_server_port,
147 "--ves-host", self.hv_collector_host,
148 "--ves-port", "6061",
149 "--key-store", self.key_store_path,
150 "--trust-store", self.trust_store_path,
151 "--key-store-password", self.sec_store_passwd,
152 "--trust-store-password", self.sec_store_passwd
155 startup_command.append("--ssl-disable")
156 return startup_command
158 def get_healthcheck_command(self):
160 "interval": 5 * ONE_SECOND_IN_NANOS,
161 "timeout": 3 * ONE_SECOND_IN_NANOS,
163 "test": ["CMD", "curl", "--request", "GET",
164 "--fail", "--silent", "--show-error",
165 "localhost:" + self.healthcheck_server_port + "/health/ready"]
169 class ContainerException(Exception):
170 def __init__(self, message):
171 super(ContainerException, self).__init__(message)