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"]["pnfRegistrationFields"]["oamV4IpAddress"]
  25         ipv6 = json_to_python["event"]["pnfRegistrationFields"]["oamV6IpAddress"]
  26         correlationId = json_to_python["event"]["commonEventHeader"]["sourceName"]
  27         str_json = '{"correlationId":"' + correlationId + '","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         json_to_python = json.loads(json_file)
  34         correlationId = json_to_python["event"]["commonEventHeader"]["sourceName"]
  35         return correlationId
  36
  37     @staticmethod
  38     def stop_aai():
  39         client = docker.from_env()
  40         container = client.containers.get('aai_simulator')
  41         container.stop()
  42
  43     def create_invalid_notification(self, json_file):
  44         return self.create_pnf_ready_notification(json_file).replace("\":", "\": ")\
  45             .replace("ipaddress-v4-oam", "oamV4IpAddress").replace("ipaddress-v6-oam", "oamV6IpAddress")\
  46             .replace("}", "\\n}")