tests/vid/resources/simulators/SO.py

   1 # ============LICENSE_START=======================================================
   2 # INTEGRATION CSIT
   3 # ================================================================================
   4 # Copyright (C) 2018 Nokia Intellectual Property. All rights reserved.
   5 # ================================================================================
   6 # Licensed under the Apache License, Version 2.0 (the "License");
   7 # you may not use this file except in compliance with the License.
   8 # You may obtain a copy of the License at
   9 #
  10 #      http://www.apache.org/licenses/LICENSE-2.0
  11 #
  12 # Unless required by applicable law or agreed to in writing, software
  13 # distributed under the License is distributed on an "AS IS" BASIS,
  14 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15 # See the License for the specific language governing permissions and
  16 # limitations under the License.
  17 # ============LICENSE_END=========================================================
  18
  19 import json
  20 import logging
  21 from functools import partial
  22 from sys import argv
  23 from http.server import BaseHTTPRequestHandler, HTTPServer
  24
  25 DEFAULT_PORT = 8443
  26
  27
  28 class SOHandler(BaseHTTPRequestHandler):
  29     def __init__(self, expected_requests, expected_responses, requests, responses, *args, **kwargs):
  30
  31         self._expected_requests = expected_requests
  32         self._expected_responses = expected_responses
  33         self._requests = requests["requests"]
  34         self._responses = responses["responses"]
  35         self._known_endpoints = self._get_known_endpoints()
  36         super().__init__(*args, **kwargs)
  37
  38     def do_POST(self):
  39         logging.info(
  40             'POST called. Expected POST REQUEST: ' + json.dumps(
  41                 self._expected_requests["post"]) + '\nExpected POST response: ' +
  42             json.dumps(self._expected_responses["post"]))
  43         self.send_response(200)
  44         self._set_headers()
  45
  46         self.wfile.write(json.dumps(self._expected_responses["post"]).encode("utf-8"))
  47         return
  48
  49     def do_GET(self):
  50         if self._does_path_exist_as_endpoint():
  51             self.rest_method("GET")
  52         else:
  53             logging.info('GET called. Expected GET REQUEST: '
  54                          + json.dumps(self._expected_requests["get"])
  55                          + '\nExpected GET response: '
  56                          + json.dumps(self._expected_responses["get"]))
  57             self.send_response(200)
  58             self._set_headers()
  59             self.wfile.write(json.dumps(self._expected_responses["get"]).encode("utf-8"))
  60
  61     def do_PUT(self):
  62         if self._does_path_exist_as_endpoint():
  63             self.rest_method("PUT")
  64         else:
  65             request_body_json = self._get_request_body()
  66             if request_body_json is not None:
  67                 self._apply_expected_data(request_body_json)
  68                 logging.info("EXPECTED RESPONSES: " + str(self._expected_responses))
  69                 logging.info("EXPECTED REQUESTS: " + str(self._expected_requests))
  70                 response_status = 200
  71             else:
  72                 response_status = 400
  73             self.send_response(response_status)
  74             self._set_headers()
  75
  76     def rest_method(self,method):
  77         expected_response = self._get_response_by_endpoint(self.path,method)
  78         if expected_response:
  79             self.send_response(expected_response["responseCode"])
  80             self._set_headers()
  81             self.wfile.write(json.dumps(expected_response["body"]).encode("utf-8"))
  82         else:
  83             response_body = "{\"message:\" \"no response for endpoint\"}"
  84             self.send_response(400)
  85             self._set_headers()
  86             self.wfile.write(json.dumps(response_body).encode("utf-8"))
  87
  88     def _get_request_body(self):
  89         content_len = int(self.headers['Content-Length'], 0)
  90         parsed_req_body = None
  91         if content_len > 0:
  92             body = self.rfile.read(content_len)
  93             body_decoded = body.decode('utf8')
  94             logging.info("BODY: %s type: %s  body decoded: %s type: %s", str(body), type(body), str(body_decoded),
  95                          type(body_decoded))
  96             parsed_req_body = json.loads(body_decoded)
  97         return parsed_req_body
  98
  99     def _apply_expected_data(self, request_body_json):
 100         if self.path == '/setResponse':
 101             logging.info("IN PUT /setResponse: " + str(request_body_json))
 102             self._expected_responses.update(request_body_json)
 103         elif self.path == '/setRequest':
 104             logging.info("IN PUT /setRequest: " + str(request_body_json))
 105             self._expected_requests.update(request_body_json)
 106
 107     def _set_headers(self):
 108         self.send_header('Content-Type', 'application/json')
 109         self.end_headers()
 110
 111     def _get_response_by_endpoint(self, endpoint, method):
 112         for response in self._responses:
 113             if response["path"] == endpoint and response["method"] == method:
 114                 return response
 115
 116     def _does_path_exist_as_endpoint(self):
 117         does_exist = False
 118         for ep in self._known_endpoints:
 119             if ep == self.path:
 120                 does_exist = True
 121                 break
 122         return does_exist
 123
 124     def _create_path_from_request(self, request):
 125         base_uri = request["path"]
 126         query_params = request["queryParams"]
 127         return base_uri + self._get_params_uri(query_params)
 128
 129     def _get_known_endpoints(self):
 130         endpoints = []
 131         for request in self._requests:
 132             endpoints.append(self._create_path_from_request(request))
 133         return endpoints
 134
 135     @staticmethod
 136     def _get_params_uri(query_params):
 137         params_uri = "?"
 138         for parameter in query_params:
 139             params_uri += parameter + '=' + query_params.get(parameter, '') + '&'
 140         return params_uri[:-1]
 141
 142
 143 class JsonFileToDictReader(object):
 144
 145     @staticmethod
 146     def read_expected_test_data(expected_responses_filename):
 147         with open(expected_responses_filename, 'r') as file:
 148             return json.load(file)
 149
 150
 151 def init_so_simulator():
 152     requests = JsonFileToDictReader.read_expected_test_data("test_data_assets/requests.json")
 153     responses = JsonFileToDictReader.read_expected_test_data("test_data_assets/responses.json")
 154
 155     expected_so_requests = JsonFileToDictReader.read_expected_test_data(argv[1])
 156     expected_so_responses = JsonFileToDictReader.read_expected_test_data(argv[2])
 157     logging.basicConfig(level=logging.INFO)
 158     handler = partial(SOHandler, expected_so_requests, expected_so_responses, requests, responses)
 159     handler.protocol_version = "HTTP/1.0"
 160     httpd = HTTPServer(('', DEFAULT_PORT), handler)
 161     logging.info("serving on: " + str(httpd.socket.getsockname()))
 162     httpd.serve_forever()
 163
 164
 165 if __name__ == '__main__':
 166     init_so_simulator()