tests/dcaegen2/prh-testcases/resources/simulator/AAI.py

   1 from http.server import BaseHTTPRequestHandler
   2 from http.server import HTTPServer
   3 import _thread, ssl, time, re
   4
   5 pnfs = 'Empty'
   6
   7
   8 class AAISetup(BaseHTTPRequestHandler):
   9
  10     def do_PUT(self):
  11         if re.search('/set_pnfs', self.path):
  12             global pnfs
  13             content_length = int(self.headers['Content-Length'])
  14             pnfs = self.rfile.read(content_length)
  15             _header_200_and_json(self)
  16
  17         return
  18
  19     def do_POST(self):
  20         if re.search('/reset', self.path):
  21             global pnfs
  22             pnfs = 'Empty'
  23             _header_200_and_json(self)
  24
  25         return
  26
  27 class AAIHandler(BaseHTTPRequestHandler):
  28
  29     def do_PATCH(self):
  30         pnfs_name = '/aai/v12/network/pnfs/pnf/' + pnfs.decode()
  31         if re.search('wrong_aai_record', self.path):
  32             self.send_response(400)
  33             self.end_headers()
  34         elif re.search(pnfs_name, self.path):
  35             self.send_response(200)
  36             self.end_headers()
  37
  38         return
  39
  40
  41 def _header_200_and_json(self):
  42     self.send_response(200)
  43     self.send_header('Content-Type', 'application/json')
  44     self.end_headers()
  45
  46
  47 def _main_(handler_class=AAIHandler, protocol="HTTP/1.0"):
  48     handler_class.protocol_version = protocol
  49     _thread.start_new_thread(_init_http_endpoints, (3333, AAIHandler))
  50     _thread.start_new_thread(_init_https_endpoints, (3334, AAIHandler))
  51     _thread.start_new_thread(_init_http_endpoints, (3335, AAISetup))
  52     while 1:
  53         time.sleep(10)
  54
  55
  56 def _init_http_endpoints(port, handler_class, server_class=HTTPServer):
  57     server = server_class(('', port), handler_class)
  58     sa = server.socket.getsockname()
  59     print("Serving HTTP on", sa[0], "port", sa[1], "for", handler_class, "...")
  60     server.serve_forever()
  61
  62
  63 def _init_https_endpoints(port, handler_class, server_class=HTTPServer):
  64     server = server_class(('', port), handler_class)
  65     server.socket = ssl.wrap_socket(server.socket,
  66                                     keyfile="certs/server.key", certfile="certs/server.crt",
  67                                     ca_certs="certs/client.crt", server_side=True)
  68     sa = server.socket.getsockname()
  69     print("Serving HTTPS on", sa[0], "port", sa[1], "for", handler_class, "...")
  70     server.serve_forever()
  71
  72
  73 if __name__ == '__main__':
  74     _main_()