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

   1 import json
   2
   3 import docker
   4 import time
   5
   6
   7 class PrhLibrary(object):
   8
   9     def __init__(self):
  10         pass
  11
  12     @staticmethod
  13     def check_for_log(search_for):
  14         client = docker.from_env()
  15         container = client.containers.get('prh')
  16         print ("Check for log searches for pattern: ", search_for )
  17         for line in container.logs(stream=True):
  18             print ("Check for log analysis line: ", line )
  19             if search_for in line.strip():
  20                 return True
  21         else:
  22             return False
  23
  24     @staticmethod
  25     def create_invalid_notification(json_file):
  26         json_to_python = json.loads(json_file)
  27         correlation_id = PrhLibrary.extract_correlation_id_value(json_to_python, "correlationId")
  28         ipv4 = PrhLibrary.extract_value_from_pnfRegistrationFields(json_to_python, "oamV4IpAddress", "oamV4IpAddress")
  29         ipv6 = PrhLibrary.extract_value_from_pnfRegistrationFields(json_to_python, "oamV6IpAddress", "oamV6IpAddress")
  30         serial_number = PrhLibrary.extract_value_from_pnfRegistrationFields(json_to_python, "serialNumber", "serialNumber")
  31         vendor_name = PrhLibrary.extract_value_from_pnfRegistrationFields(json_to_python, "vendorName", "vendorName")
  32         model_number = PrhLibrary.extract_value_from_pnfRegistrationFields(json_to_python, "modelNumber", "modelNumber")
  33         unit_type = PrhLibrary.extract_value_from_pnfRegistrationFields(json_to_python, "unitType", "unitType")
  34
  35         additional_fields = PrhLibrary.extract_additional_fields(json_to_python)
  36
  37         str_json = '{' + correlation_id + ipv4 + ipv6 + serial_number + vendor_name + model_number + unit_type + '"nfNamingCode":""' + "," + '"softwareVersion":"",' + additional_fields
  38         return json.dumps(str_json).replace("\\", "")[1:-1].replace("\":", "\": ").rstrip(',') + '\\n}'
  39
  40     @staticmethod
  41     def create_pnf_ready_notification_as_pnf_ready(json_file):
  42         json_to_python = json.loads(json_file)
  43         correlation_id = PrhLibrary.extract_correlation_id_value(json_to_python, "correlationId")
  44         serial_number = PrhLibrary.extract_value_from_pnfRegistrationFields(json_to_python, "serial-number", "serialNumber")
  45         vendor_name = PrhLibrary.extract_value_from_pnfRegistrationFields(json_to_python, "equip-vendor", "vendorName")
  46         model_number = PrhLibrary.extract_value_from_pnfRegistrationFields(json_to_python, "equip-model", "modelNumber")
  47         unit_type = PrhLibrary.extract_value_from_pnfRegistrationFields(json_to_python, "equip-type", "unitType")
  48
  49         additional_fields = PrhLibrary.extract_additional_fields_value(json_to_python)
  50
  51         nf_role  = json_to_python.get("event").get("commonEventHeader").get("nfNamingCode") if "nfNamingCode" in json_to_python["event"]["commonEventHeader"] else ""
  52
  53         str_json = '{' + correlation_id + serial_number + vendor_name + model_number + unit_type + '"nf-role":"' + nf_role + '","sw-version":"",' + additional_fields
  54
  55         return json.dumps(str_json.rstrip(',') + '}').replace("\\", "")[1:-1]
  56
  57     @staticmethod
  58     def extract_additional_fields_value(content):
  59         fields = PrhLibrary.get_additional_fields_as_key_value_pairs(content)
  60         if len(fields) == 0:
  61             return ""
  62         return PrhLibrary.build_additional_fields_json(fields)
  63
  64     @staticmethod
  65     def extract_additional_fields(content):
  66         fields = PrhLibrary.get_additional_fields_as_key_value_pairs(content)
  67         if fields == []:
  68             return '"additionalFields":null'
  69         return PrhLibrary.build_additional_fields_json(fields)
  70
  71     @staticmethod
  72     def get_additional_fields_as_key_value_pairs(content):
  73         return content.get("event").get("pnfRegistrationFields").get(
  74             "additionalFields") if "additionalFields" in content["event"]["pnfRegistrationFields"] else []
  75
  76     @staticmethod
  77     def build_additional_fields_json(fields):
  78         res = '"additionalFields":{'
  79         for f in fields:
  80             res += '"' + f + '":"' + fields.get(f) + '",'
  81         return res.rstrip(',') + '},'
  82
  83     @staticmethod
  84     def extract_value_from_pnfRegistrationFields(content, name, key):
  85         return '"' + name + '":"' + (content.get("event").get("pnfRegistrationFields").get(key) + '",' if key in content["event"]["pnfRegistrationFields"] else '",')
  86
  87     @staticmethod
  88     def extract_correlation_id_value(content, name):
  89         return '"' + name + '":"' + (content.get("event").get("commonEventHeader").get("sourceName") + '",' if "sourceName" in content["event"]["commonEventHeader"] else '",')
  90
  91     @staticmethod
  92     def create_pnf_name(json_file):
  93         json_to_python = json.loads(json_file)
  94         correlation_id = json_to_python.get("event").get("commonEventHeader").get("sourceName") + '",' if "sourceName" in json_to_python["event"]["commonEventHeader"] else '",'
  95         return correlation_id
  96
  97     @staticmethod
  98     def ensure_container_is_running(name):
  99         client = docker.from_env()
 100
 101         if not PrhLibrary.is_in_status(client, name, "running"):
 102             print ("starting container", name)
 103             container = client.containers.get(name)
 104             container.start()
 105             PrhLibrary.wait_for_status(client, name, "running")
 106
 107         PrhLibrary.print_status(client)
 108
 109     @staticmethod
 110     def ensure_container_is_exited(name):
 111         client = docker.from_env()
 112
 113         if not PrhLibrary.is_in_status(client, name, "exited"):
 114             print ("stopping container", name)
 115             container = client.containers.get(name)
 116             container.stop()
 117             PrhLibrary.wait_for_status(client, name, "exited")
 118
 119         PrhLibrary.print_status(client)
 120
 121     @staticmethod
 122     def print_status(client):
 123         print("containers status")
 124         for c in client.containers.list(all=True):
 125             print(c.name, "   ", c.status)
 126
 127     @staticmethod
 128     def wait_for_status(client, name, status):
 129         while not PrhLibrary.is_in_status(client, name, status):
 130             print ("waiting for container: ", name, "to be in status: ", status)
 131             time.sleep(3)
 132
 133     @staticmethod
 134     def is_in_status(client, name, status):
 135         return len(client.containers.list(all=True, filters={"name": "^/"+name+"$", "status": status})) == 1