1 # ============LICENSE_START=======================================================
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
10 # http://www.apache.org/licenses/LICENSE-2.0
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=========================================================
21 from functools import partial
23 from http.server import BaseHTTPRequestHandler, HTTPServer
28 class SOHandler(BaseHTTPRequestHandler):
29 def __init__(self, expected_requests, expected_responses, requests, responses, *args, **kwargs):
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)
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)
46 self.wfile.write(json.dumps(self._expected_responses["post"]).encode("utf-8"))
50 print(self._known_endpoints)
51 if self._does_path_exist_as_endpoint():
52 expected_response = self._get_response_by_path()
54 self.send_response(expected_response["responseCode"])
56 self.wfile.write(json.dumps(expected_response["body"]).encode("utf-8"))
58 response_body = "{\"message:\" \"no response for endpoint\"}"
59 self.send_response(400)
61 self.wfile.write(json.dumps(response_body).encode("utf-8"))
63 logging.info('GET called. Expected GET REQUEST: '
64 + json.dumps(self._expected_requests["get"])
65 + '\nExpected GET response: '
66 + json.dumps(self._expected_responses["get"]))
67 self.send_response(200)
69 self.wfile.write(json.dumps(self._expected_responses["get"]).encode("utf-8"))
72 request_body_json = self._get_request_body()
73 if request_body_json is not None:
74 self._apply_expected_data(request_body_json)
75 logging.info("EXPECTED RESPONSES: " + str(self._expected_responses))
76 logging.info("EXPECTED REQUESTS: " + str(self._expected_requests))
80 self.send_response(response_status)
83 def _get_request_body(self):
84 content_len = int(self.headers['Content-Length'], 0)
85 parsed_req_body = None
87 body = self.rfile.read(content_len)
88 body_decoded = body.decode('utf8')
89 logging.info("BODY: %s type: %s body decoded: %s type: %s", str(body), type(body), str(body_decoded),
91 parsed_req_body = json.loads(body_decoded)
92 return parsed_req_body
94 def _apply_expected_data(self, request_body_json):
95 if self.path == '/setResponse':
96 logging.info("IN PUT /setResponse: " + str(request_body_json))
97 self._expected_responses.update(request_body_json)
98 elif self.path == '/setRequest':
99 logging.info("IN PUT /setRequest: " + str(request_body_json))
100 self._expected_requests.update(request_body_json)
102 def _set_headers(self):
103 self.send_header('Content-Type', 'application/json')
106 def _get_response_by_path(self):
107 for response in self._responses:
108 if response["path"] == self.path:
111 def _does_path_exist_as_endpoint(self):
113 for ep in self._known_endpoints:
119 def _create_path_from_request(self, request):
120 base_uri = request["path"]
121 query_params = request["queryParams"]
122 return base_uri + self._get_params_uri(query_params)
124 def _get_known_endpoints(self):
126 for request in self._requests:
127 endpoints.append(self._create_path_from_request(request))
131 def _get_params_uri(query_params):
133 for parameter in query_params:
134 params_uri += parameter + '=' + query_params.get(parameter, '') + '&'
135 return params_uri[:-1]
138 class JsonFileToDictReader(object):
141 def read_expected_test_data(expected_responses_filename):
142 with open(expected_responses_filename, 'r') as file:
143 return json.load(file)
146 def init_so_simulator():
147 requests = JsonFileToDictReader.read_expected_test_data("test_data_assets/requests.json")
148 responses = JsonFileToDictReader.read_expected_test_data("test_data_assets/responses.json")
150 expected_so_requests = JsonFileToDictReader.read_expected_test_data(argv[1])
151 expected_so_responses = JsonFileToDictReader.read_expected_test_data(argv[2])
152 logging.basicConfig(level=logging.INFO)
153 handler = partial(SOHandler, expected_so_requests, expected_so_responses, requests, responses)
154 handler.protocol_version = "HTTP/1.0"
155 httpd = HTTPServer(('', DEFAULT_PORT), handler)
156 logging.info("serving on: " + str(httpd.socket.getsockname()))
157 httpd.serve_forever()
160 if __name__ == '__main__':