test/csit/tests/dcaegen2/prh_testcases/resources/PrhLibrary.py

   1 '''
   2 Created on Apr 27, 2018
   3
   4 @author: mwagner9
   5 '''
   6
   7 import threading
   8 import time
   9
  10 from robot.api import logger
  11
  12 import PrhVariables
  13 import Simulator
  14
  15 st = '{"event":{"correlationID":"NOKQTFCOC540002E", "pnfOamIpv4Address":"10.16.123.234", "pnfOamIpv6Address":"2001:0db8:85a3:0000:0000:8a2e:0370:7334"}}'
  16
  17 class PrhLibrary(object):
  18     def __init__(self):
  19         pass
  20
  21     def setup_dmaap_server(self):
  22         return _setup(PrhVariables.DMaaPHttpServerThread, 'DMaaP', PrhVariables.DMaaPIsRobotRun, Simulator, 3904)
  23
  24     def setup_aai_server(self):
  25         return _setup(PrhVariables.AAIHttpServerThread, 'AAI', PrhVariables.AAIIsRobotRun, Simulator, 3905)
  26
  27     def shutdown_dmaap_server(self):
  28         return _shutdown(PrhVariables.DMaaPHTTPD, 'DMaaP')
  29
  30     def shutdown_aai_server(self):
  31         return _shutdown(PrhVariables.AAIHTTPD, 'AAI')
  32
  33     def is_json_empty(self, response):
  34         logger.info("Enter is_json_empty: response.text: " + response.text)
  35         if response.text is None or len(response.text) < 2:
  36             return 'True'
  37         return 'False'
  38
  39     def dmaap_collectorTimeStamp_receive(self, search, response):
  40         return _find_element(search, response)
  41
  42     def AAI_Ipv4_receive(self, search, response):
  43         return _find_element(search, response)
  44
  45     def AAI_Ipv6_receive(self, search, response):
  46         return _find_element(search, response)
  47
  48     def check_pnf_ready(self):
  49         if st in Simulator.prh_ready:
  50             return 'true'
  51         return 'false'
  52
  53 def _setup(serverthread, servername, isrobotrun, module, portNum):
  54     if serverthread is not None:
  55         logger.console('{} Mockup Sever started'.format(servername))
  56         return "true"
  57
  58     isrobotrun = True
  59
  60     module_handler = module.AAIHandler if servername is 'AAI' else module.DMaaPHandler
  61
  62     module.run_server(module_handler, portNum)
  63     try:
  64         serverthread = threading.Thread(name='{}_HTTPServer'.format(servername), target=module.Httpd.serve_forever)
  65         serverthread.start()
  66         logger.console('{}  Mockup Sever started'.format(servername))
  67         time.sleep(2)
  68         return "true"
  69     except Exception as e:
  70         print (str(e))
  71         return "false"
  72
  73 def _shutdown(server, name):
  74     if server is not None:
  75         server.shutdown()
  76         logger.console("{} Server shut down".format(name))
  77         time.sleep(3)
  78         return "true"
  79     else:
  80         return "false"
  81
  82 def _find_element(search, response):
  83     while response is not None:
  84         json_data = str(response)
  85         if search in json_data:
  86             return 'true'
  87     return 'false'
  88
  89 if __name__ == '__main__':
  90     lib = PrhLibrary()
  91     ret = lib.setup_dmaap_server()
  92     print ret
  93     ret = lib.setup_aai_server()
  94     print ret
  95     time.sleep(10)