*** Settings ***
Library DcaeAppSimulatorLibrary
+Library XnfSimulatorLibrary
+Library VesHvContainersUtilsLibrary
+Library Collections
+
+Suite Setup Message Routing Suite Setup
+Suite Teardown VES-HV Collector Suite Teardown
+Test Teardown VES-HV Collector Test Shutdown
*** Test Cases ***
-Initial testcase
- [Documentation] Testing dcae app connection
+Correct Messages Routing
+ [Documentation] VES-HV Collector should route all valid messages to topics specified in configuration
+ ... and do not change message payload generated in XNF simulator
+
+ ${SIMULATORS_LIST}= Get xNF Simulators 1
+ Send Messages From xNF Simulators ${SIMULATORS_LIST} ${XNF_FIXED_PAYLOAD_REQUEST}
+
Wait until keyword succeeds 60 sec 5 sec
- ... Assert Dcae App Consumed ${DCAE_APP_API_MESSAGES_COUNT_URL} 0
+ ... Assert Dcae App Consumed ${DCAE_APP_API_MESSAGES_COUNT_URL} ${AMOUNT_25000}
+
+*** Keywords ***
+Message Routing Suite Setup
+ Log Started Suite: VES-HV Message Routing
+ ${XNF_PORTS_LIST}= Create List 7000
+ Configure xNF Simulators On Ports ${XNF_PORTS_LIST}
+ Log Suite setup finished
+
+Configure xNF Simulators On Ports
+ [Arguments] ${XNF_PORTS_LIST}
+ ${XNF_SIMULATORS_ADDRESSES}= Start Xnf Simulators ${XNF_PORTS_LIST} True
+ Set Suite Variable ${XNF_SIMULATORS_ADDRESSES}
+
+
+Get xNF Simulators
+ [Arguments] ${AMOUNT}
+ ${SIMULATORS}= Get Slice From List ${XNF_SIMULATORS_ADDRESSES} 0 ${AMOUNT}
+ [Return] ${SIMULATORS}
+
+
+Send Messages From xNF Simulators
+ [Arguments] ${XNF_HOSTS_LIST} ${MESSAGE_FILEPATH}
+ :FOR ${HOST} IN @{XNF_HOSTS_LIST}
+ \ ${XNF_SIM_API_ACCESS}= Get xNF Sim Api Access Url ${HTTP_METHOD_URL} ${HOST}
+ \ ${XNF_SIM_API_URL}= Catenate SEPARATOR= ${XNF_SIM_API_ACCESS} ${XNF_SIM_API_PATH}
+ \ Send messages ${XNF_SIM_API_URL} ${MESSAGE_FILEPATH}
+
+
+VES-HV Collector Test Shutdown
+ Reset DCAE App Simulator ${DCAE_APP_API_MESSAGE_RESET_URL}
+
+
+VES-HV Collector Suite Teardown
+ Stop And Remove All Xnf Simulators
+
+*** Variables ***
+${HTTP_METHOD_URL} http://
+
+${XNF_SIM_API_PATH} /simulator/async
+
+${VES_HV_SCENARIOS} %{WORKSPACE}/test/csit/tests/dcaegen2-collectors-hv-ves/testcases/resources/scenarios
+${XNF_FIXED_PAYLOAD_REQUEST} ${VES_HV_SCENARIOS}/fixed-payload/xnf-fixed-payload-request.json
+
+${AMOUNT_25000} 25000
\ No newline at end of file
from robot.api import logger
import os.path
+import docker
+from io import BytesIO
+from os.path import basename
+from tarfile import TarFile, TarInfo
LOCALHOST = "localhost"
self.get_instance_address(image_name, port)
)
+ def get_xnf_sim_api_access_url(self, method, host):
+ if is_running_inside_docker():
+ return self.create_url(method, host)
+ else:
+ logger.info("File `/.dockerenv` not found. Assuming local environment and using localhost.")
+ port_from_container_name = str(host)[-4:]
+ return self.create_url(method, LOCALHOST + ":" + port_from_container_name)
+
def get_dcae_app_api_access_url(self, method, image_name, port):
return self.create_url(
method,
def is_running_inside_docker():
return os.path.isfile("/.dockerenv")
+
+def copy_to_container(container_id, filepaths, path='/etc/ves-hv'):
+ with create_archive(filepaths) as archive:
+ docker.APIClient('unix:///var/run/docker.sock') \
+ .put_archive(container=container_id, path=(path), data=archive)
+
+
+def create_archive(filepaths):
+ tarstream = BytesIO()
+ tarfile = TarFile(fileobj=tarstream, mode='w')
+ for filepath in filepaths:
+ file = open(filepath, 'r')
+ file_data = file.read()
+
+ tarinfo = TarInfo(name=basename(file.name))
+ tarinfo.size = len(file_data)
+ tarinfo.mtime = time()
+
+ tarfile.addfile(tarinfo, BytesIO(file_data))
+
+ tarfile.close()
+ tarstream.seek(0)
+ return tarstream
--- /dev/null
+from VesHvContainersUtilsLibrary import copy_to_container
+import HttpRequests
+import os
+import docker
+from robot.api import logger
+from time import sleep
+
+XNF_SIMULATOR_NAME = "xNF Simulator"
+SIMULATOR_IMAGE_NAME = "onap/org.onap.dcaegen2.collectors.hv-ves.hv-collector-xnf-simulator"
+SIMULATOR_IMAGE_FULL_NAME = os.getenv("DOCKER_REGISTRY") + "/" + SIMULATOR_IMAGE_NAME + ":latest"
+certificates_dir_path = os.getenv("WORKSPACE") + "/test/csit/plans/dcaegen2-collectors-hv-ves/testsuites/ssl/"
+ONE_SECOND_IN_NANOS = 10 ** 9
+
+class XnfSimulatorLibrary:
+
+ def start_xnf_simulators(self, list_of_ports, valid_certs=True):
+ logger.info("Creating " + str(len(list_of_ports)) + " xNF Simulator containers")
+ dockerClient = docker.from_env()
+ cert_name_prefix = "" if valid_certs else "invalid_"
+ self.pullImageIfAbsent(dockerClient)
+ logger.info("Using image: " + SIMULATOR_IMAGE_FULL_NAME)
+ simulators_addresses = self.create_simulators(dockerClient, list_of_ports, cert_name_prefix)
+ self.assert_containers_startup_was_successful(dockerClient)
+ dockerClient.close()
+ return simulators_addresses
+
+ def pullImageIfAbsent(self, dockerClient):
+ try:
+ dockerClient.images.get(SIMULATOR_IMAGE_FULL_NAME)
+ except:
+ logger.console("Image " + SIMULATOR_IMAGE_FULL_NAME + " will be pulled from repository. "
+ "This can take a while.")
+ dockerClient.images.pull(SIMULATOR_IMAGE_FULL_NAME)
+
+ def create_simulators(self, dockerClient, list_of_ports, cert_name_prefix):
+ simulators_addresses = []
+ for port in list_of_ports:
+ container = self.run_simulator(dockerClient, port,
+ "/etc/ves-hv/" + cert_name_prefix + "client.crt",
+ "/etc/ves-hv/" + cert_name_prefix + "client.key",
+ "/etc/ves-hv/" + cert_name_prefix + "trust.crt"
+ )
+
+ self.copy_required_certificates_into_simulator(container)
+ logger.info("Started container: " + container.name + " " + container.id)
+ simulators_addresses.append(container.name + ":" + port)
+ return simulators_addresses
+
+ def run_simulator(self, dockerClient, port, client_crt_path, client_key_path, client_trust_store):
+ return dockerClient.containers.run(SIMULATOR_IMAGE_FULL_NAME,
+ command=["--listen-port", port,
+ "--ves-host", "ves-hv-collector",
+ "--ves-port", "6061",
+ "--cert-file", client_crt_path,
+ "--private-key-file", client_key_path,
+ "--trust-cert-file", client_trust_store
+ ],
+ healthcheck={
+ "interval": 5 * ONE_SECOND_IN_NANOS,
+ "timeout": 3 * ONE_SECOND_IN_NANOS,
+ "retries": 1,
+ "test": ["CMD", "curl", "--request", "GET",
+ "--fail", "--silent", "--show-error",
+ "localhost:" + port + "/healthcheck"]
+ },
+ detach=True,
+ network="ves-hv-default",
+ ports={port + "/tcp": port},
+ name="ves-hv-collector-xnf-simulator" + port)
+
+ def copy_required_certificates_into_simulator(self, container):
+ container.exec_run("mkdir -p /etc/ves-hv")
+ copy_to_container(container.id, [
+ certificates_dir_path + "client.crt",
+ certificates_dir_path + "client.key",
+ certificates_dir_path + "trust.crt",
+ certificates_dir_path + "invalid_client.crt",
+ certificates_dir_path + "invalid_client.key",
+ certificates_dir_path + "invalid_trust.crt",
+ ])
+
+ def assert_containers_startup_was_successful(self, dockerClient):
+ checks_amount = 6
+ check_interval_in_seconds = 5
+ for _ in range(checks_amount):
+ sleep(check_interval_in_seconds)
+ all_containers_healthy = True
+ for container in self.get_simulators_list(dockerClient):
+ all_containers_healthy = all_containers_healthy and self.is_container_healthy(container)
+ if (all_containers_healthy):
+ return
+ raise ContainerException("One of xNF simulators containers did not pass the healthcheck.")
+
+ def is_container_healthy(self, container):
+ container_health = container.attrs['State']['Health']['Status']
+ return container_health == 'healthy' and container.status == 'running'
+
+ def stop_and_remove_all_xnf_simulators(self):
+ dockerClient = docker.from_env()
+ for container in self.get_simulators_list(dockerClient):
+ logger.info("Stopping and removing container: " + container.id)
+ logger.debug(container.logs())
+ container.stop()
+ container.remove()
+ dockerClient.close()
+
+ def get_simulators_list(self, dockerClient):
+ return dockerClient.containers.list(filters={"ancestor": SIMULATOR_IMAGE_FULL_NAME}, all=True)
+
+ def send_messages(self, simulator_url, message_filepath):
+ logger.info("Reading message to simulator from: " + message_filepath)
+
+ file = open(message_filepath, "rb")
+ data = file.read()
+ file.close()
+
+ logger.info("POST at: " + simulator_url)
+ resp = HttpRequests.session_without_env().post(simulator_url, data=data, timeout=5)
+ HttpRequests.checkStatusCode(resp.status_code, XNF_SIMULATOR_NAME)
+
+
+class ContainerException(Exception):
+ def __init__(self, message):
+ super(ContainerException, self).__init__(message)
Code Review / integration.git / commitdiff