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

   1 import json
   2
   3 import docker
   4
   5
   6 class PrhLibrary(object):
   7
   8     def __init__(self):
   9         pass
  10
  11     @staticmethod
  12     def check_for_log(search_for):
  13         client = docker.from_env()
  14         container = client.containers.get('prh')
  15         for line in container.logs(stream=True):
  16             if search_for in line.strip():
  17                 return True
  18         else:
  19             return False
  20
  21     @staticmethod
  22     def create_pnf_ready_notification(json_file):
  23         json_to_python = json.loads(json_file)
  24         ipv4 = json_to_python["event"]["otherFields"]["pnfOamIpv4Address"]
  25         ipv6 = json_to_python["event"]["otherFields"]["pnfOamIpv6Address"]
  26         pnf_name = _create_pnf_name(json_file)
  27         str_json = '{"pnf-name":"' + pnf_name + '","ipaddress-v4-oam":"' + ipv4 + '","ipaddress-v6-oam":"' + ipv6 + '"}'
  28         python_to_json = json.dumps(str_json)
  29         return python_to_json.replace("\\", "")[1:-1]
  30
  31     @staticmethod
  32     def create_pnf_name(json_file):
  33         return _create_pnf_name(json_file)
  34
  35     @staticmethod
  36     def stop_aai():
  37         client = docker.from_env()
  38         container = client.containers.get('aai_simulator')
  39         container.stop()
  40
  41
  42 def _create_pnf_name(json_file):
  43     json_to_python = json.loads(json_file)
  44     vendor = json_to_python["event"]["otherFields"]["pnfVendorName"]
  45     serial_number = json_to_python["event"]["otherFields"]["pnfSerialNumber"]
  46     return vendor[:3].upper() + serial_number