tests/dcaegen2/prh-testcases/resources/PrhLibrary.py

import json
import re
import docker
import time
import datetime


class PrhLibrary(object):
    ROBOT_LIBRARY_SCOPE = 'TEST SUITE'
    ROBOT_LISTENER_API_VERSION = 2

    def __init__(self):
        self.ROBOT_LIBRARY_LISTENER = self

    def _start_test(self, name, attrs):
        # http://robotframework.org/robotframework/latest/RobotFrameworkUserGuide.html#test-libraries-as-listeners
        self.test_start_time = self.get_current_utc_datetime()

    @staticmethod
    def get_current_utc_datetime():
        return datetime.datetime.utcnow()

    def get_docker_logs_since_test_start(self, container_id):
        return self.get_docker_logs(container_id, self.test_start_time)

    @staticmethod
    def get_docker_logs(container_id, since=None):
        container = PrhLibrary.__get_docker_container(container_id)
        return container.logs(stream=False, since=since)

    def wait_for_one_of_docker_log_entries(self, container_id, searched_entries):
        print("Looking for: %s" % searched_entries)
        container = PrhLibrary.__get_docker_container(container_id)
        print("Log lines:")
        for line in container.logs(stream=True, since=self.test_start_time):
            print(line)
            for searched_entry in searched_entries:
                if searched_entry in line.strip():
                    return True
        else:
            return False

    def wait_for_log_entry_with_json_message(self, prefix, json_message):
        print("Looking for:")
        print("Prefix: " + str(prefix))
        print("Json: " + str(json_message))
        try:
            decoded_message = json.loads(json_message)
        except json.JSONDecodeError:
            print("Could not decode given message")
            return False
        pattern = re.compile(prefix + "(.*)$")
        container = PrhLibrary.__get_docker_container('prh')
        print("Log lines:")
        for line in container.logs(stream=True, since=self.test_start_time):
            print(line)
            if PrhLibrary.__same_json_in_log(decoded_message, line, pattern):
                return True
        else:
            return False

    @staticmethod
    def create_invalid_notification(json_file):
        output = {}
        input = json.loads(json_file)
        output["correlationId"] = PrhLibrary.__extract_correlation_id_value(input)
        output["oamV4IpAddress"] = PrhLibrary.__extract_value_from_pnfRegistrationFields(input, "oamV4IpAddress")
        output["oamV6IpAddress"] = PrhLibrary.__extract_value_from_pnfRegistrationFields(input, "oamV6IpAddress")
        output["serialNumber"] = PrhLibrary.__extract_value_from_pnfRegistrationFields(input, "serialNumber")
        output["vendorName"] = PrhLibrary.__extract_value_from_pnfRegistrationFields(input, "vendorName")
        output["modelNumber"] = PrhLibrary.__extract_value_from_pnfRegistrationFields(input, "modelNumber")
        output["unitType"] = PrhLibrary.__extract_value_from_pnfRegistrationFields(input, "unitType")
        output['nfNamingCode'] = ''
        output['softwareVersion'] = ''

        output["additionalFields"] = PrhLibrary.__get_additional_fields_as_key_value_pairs(input)

        return json.dumps(output)

    @staticmethod
    def create_pnf_ready_notification_as_pnf_ready(json_file):
        output = {}
        input = json.loads(json_file)[0]

        output["correlationId"] = PrhLibrary.__extract_correlation_id_value(input)
        output["serialNumber"] = PrhLibrary.__extract_value_from_pnfRegistrationFields(input, "serialNumber")
        output["equip-vendor"] = PrhLibrary.__extract_value_from_pnfRegistrationFields(input, "vendorName")
        output["equip-model"] = PrhLibrary.__extract_value_from_pnfRegistrationFields(input, "modelNumber")
        output["equip-type"] = PrhLibrary.__extract_value_from_pnfRegistrationFields(input, "unitType")
        output["nf-role"] = PrhLibrary.__extract_nf_role(input)
        output["sw-version"] = ""

        output["additionalFields"] = PrhLibrary.__get_additional_fields_as_key_value_pairs(input)

        return json.dumps(output)

    @staticmethod
    def ensure_container_is_running(name):
        client = docker.from_env()

        if not PrhLibrary.is_in_status(client, name, "running"):
            print ("starting container", name)
            container = client.containers.get(name)
            container.start()
            PrhLibrary.wait_for_status(client, name, "running")

        PrhLibrary.print_status(client)

    @staticmethod
    def ensure_container_is_exited(name):
        client = docker.from_env()

        if not PrhLibrary.is_in_status(client, name, "exited"):
            print ("stopping container", name)
            container = client.containers.get(name)
            container.stop()
            PrhLibrary.wait_for_status(client, name, "exited")

        PrhLibrary.print_status(client)

    @staticmethod
    def print_status(client):
        print("containers status")
        for c in client.containers.list(all=True):
            print(c.name, "   ", c.status)

    @staticmethod
    def wait_for_status(client, name, status):
        while not PrhLibrary.is_in_status(client, name, status):
            print ("waiting for container: ", name, "to be in status: ", status)
            time.sleep(3)

    @staticmethod
    def is_in_status(client, name, status):
        return len(client.containers.list(all=True, filters={"name": "^/"+name+"$", "status": status})) == 1

    @staticmethod
    def create_pnf_name(json_file):
        json_to_python = json.loads(json_file)
        correlation_id = json_to_python.get("event").get("commonEventHeader").get("sourceName") + '",' if "sourceName" in json_to_python["event"]["commonEventHeader"] else '",'
        return correlation_id

    @staticmethod
    def __get_additional_fields_as_key_value_pairs(content):
        return content.get("event").get("pnfRegistrationFields").get(
            "additionalFields") if "additionalFields" in content["event"]["pnfRegistrationFields"] else {}

    @staticmethod
    def __extract_value_from_pnfRegistrationFields(content, key):
        return content["event"]["pnfRegistrationFields"][key] if key in content["event"]["pnfRegistrationFields"] else ''

    @staticmethod
    def __extract_correlation_id_value(content):
        return content["event"]["commonEventHeader"]["sourceName"] if "sourceName" in content["event"]["commonEventHeader"] else ''

    @staticmethod
    def __extract_nf_role(content):
        return content["event"]["commonEventHeader"]["nfNamingCode"] if "nfNamingCode" in content["event"]["commonEventHeader"] else ''

    @staticmethod
    def __same_json_in_log(decoded_message, line, pattern):
        extracted_json = PrhLibrary.__extract_json(line, pattern)
        if extracted_json is not None:
            print("Found json: " + extracted_json)
            try:
                if json.loads(extracted_json) == decoded_message:
                    return True
            except json.JSONDecodeError:
                print("Could not decode")
        return False

    @staticmethod
    def __extract_json(line, pattern):
        full_message = PrhLibrary.__extract_full_message_from_line(line)
        if full_message is not None:
            match = pattern.match(full_message)
            if match:
                return match.group(1).replace("\\n", "\n").replace("\\t", "\t")
        return None

    @staticmethod
    def __extract_full_message_from_line(line):
        split = line.split("|")
        if len(split) > 3:
            return split[3]
        return None

    @staticmethod
    def __get_docker_container(container_id):
        docker_client = docker.from_env()
        return docker_client.containers.get(container_id)