| # ============LICENSE_START======================================================= |
| # INTEGRATION CSIT |
| # ================================================================================ |
| # Copyright (C) 2018 Nokia Intellectual Property. All rights reserved. |
| # ================================================================================ |
| # Licensed under the Apache License, Version 2.0 (the "License"); |
| # you may not use this file except in compliance with the License. |
| # You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| # ============LICENSE_END========================================================= |
| |
| import json |
| import logging |
| from functools import partial |
| from sys import argv |
| from http.server import BaseHTTPRequestHandler, HTTPServer |
| |
| DEFAULT_PORT = 8443 |
| |
| |
| class SOHandler(BaseHTTPRequestHandler): |
| def __init__(self, expected_requests, expected_responses, *args, **kwargs): |
| |
| self._expected_requests = expected_requests |
| self._expected_responses = expected_responses |
| super().__init__(*args, **kwargs) |
| |
| def do_POST(self): |
| logging.info( |
| 'POST called. Expected POST REQUEST: ' + json.dumps( |
| self._expected_requests["post"]) + '\nExpected POST response: ' + |
| json.dumps(self._expected_responses["post"])) |
| self.send_response(200) |
| self._set_headers() |
| |
| self.wfile.write(json.dumps(self._expected_responses["post"]).encode("utf-8")) |
| return |
| |
| def do_GET(self): |
| logging.info( |
| 'GET called. Expected GET REQUEST: ' + json.dumps( |
| self._expected_requests["get"]) + '\nExpected GET response: ' + |
| json.dumps(self._expected_responses["get"])) |
| self.send_response(200) |
| self._set_headers() |
| |
| self.wfile.write(json.dumps(self._expected_responses["get"]).encode("utf-8")) |
| return self._expected_responses["get"] |
| |
| def do_PUT(self): |
| request_body_json = self._get_request_body() |
| if request_body_json is not None: |
| self._apply_expected_data(request_body_json) |
| logging.info("EXPECTED RESPONSES: " + str(self._expected_responses)) |
| logging.info("EXPECTED REQUESTS: " + str(self._expected_requests)) |
| response_status = 200 |
| else: |
| response_status = 400 |
| self.send_response(response_status) |
| self._set_headers() |
| |
| def _get_request_body(self): |
| content_len = int(self.headers['Content-Length'], 0) |
| parsed_req_body = None |
| if content_len > 0: |
| body = self.rfile.read(content_len) |
| body_decoded = body.decode('utf8') |
| logging.info("BODY: %s type: %s body decoded: %s type: %s", str(body), type(body), str(body_decoded), |
| type(body_decoded)) |
| parsed_req_body = json.loads(body_decoded) |
| return parsed_req_body |
| |
| def _apply_expected_data(self, request_body_json): |
| if self.path == '/setResponse': |
| logging.info("IN PUT /setResponse: " + str(request_body_json)) |
| print("TYPE: %s and text: %s", type(request_body_json), str(request_body_json)) |
| self._expected_responses.update(request_body_json) |
| print("TYPE: %s", type(request_body_json)) |
| elif self.path == '/setRequest': |
| logging.info("IN PUT /setRequest: " + str(request_body_json)) |
| self._expected_requests.update(request_body_json) |
| |
| def _set_headers(self): |
| self.send_header('Content-Type', 'application/json') |
| self.end_headers() |
| |
| |
| class JsonFileToDictReader(object): |
| |
| @staticmethod |
| def read_expected_test_data(expected_responses_filename): |
| with open(expected_responses_filename, 'r') as file: |
| return json.load(file) |
| |
| |
| def init_so_simulator(): |
| expected_so_requests = JsonFileToDictReader.read_expected_test_data(argv[1]) |
| expected_so_responses = JsonFileToDictReader.read_expected_test_data(argv[2]) |
| logging.basicConfig(level=logging.INFO) |
| handler = partial(SOHandler, expected_so_requests, expected_so_responses) |
| handler.protocol_version = "HTTP/1.0" |
| httpd = HTTPServer(('', DEFAULT_PORT), handler) |
| logging.info("serving on: " + str(httpd.socket.getsockname())) |
| httpd.serve_forever() |
| |
| |
| if __name__ == '__main__': |
| init_so_simulator() |