scripts/multicloud-ocata/generic_sim/generic_sim.py

   1 # Copyright 2018 Intel Corporation, Inc
   2 # Licensed under the Apache License, Version 2.0 (the "License");
   3 # you may not use this file except in compliance with the License.
   4 # You may obtain a copy of the License at
   5 #
   6 #     http://www.apache.org/licenses/LICENSE-2.0
   7 #
   8 # Unless required by applicable law or agreed to in writing, software
   9 # distributed under the License is distributed on an "AS IS" BASIS,
  10 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  11 # See the License for the specific language governing permissions and
  12 # limitations under the License.
  13
  14 import json
  15 import logging
  16
  17 import web
  18 from web import webapi
  19 import yaml
  20
  21 urls = (
  22   '/(.*)','MockController'
  23 )
  24
  25 def setup_logger(name, log_file, level=logging.DEBUG):
  26     print("Configuring the logger...")
  27     handler = logging.FileHandler(log_file)
  28     formatter = logging.Formatter('%(message)s')
  29     handler.setFormatter(formatter)
  30
  31     logger = logging.getLogger(name)
  32     logger.setLevel(level)
  33     logger.addHandler(handler)
  34
  35     return logger
  36
  37
  38 class MockResponse:
  39     def __init__(self, http_verb, status_code,
  40                  content_type="application/json", body="{}",
  41                  headers={}):
  42         self.http_verb = http_verb.lower()
  43         self.status_code = status_code
  44         self.content_type = content_type
  45         self.body = body
  46         self.headers = headers
  47
  48 def _parse_responses(parsed_responses):
  49     result = {}
  50     for path, responses in parsed_responses.iteritems():
  51         new_path = path
  52         if path.startswith("/"):
  53             new_path = path[1:]
  54
  55         result[new_path] = []
  56         for http_verb, response in responses.iteritems():
  57             result[new_path].append(MockResponse(http_verb, **response))
  58     return result
  59
  60 def load_responses(filename):
  61     print("Loading responses from configuration file..")
  62     with open(filename) as yaml_file:
  63         responses_file = yaml.safe_load(yaml_file)
  64     responses_map = _parse_responses(responses_file)
  65     return responses_map
  66
  67
  68 class MockController:
  69
  70     def _do_action(self, action):
  71         logger.info('{}'.format(web.ctx.env.get('wsgi.input').read()))
  72         action = action.lower()
  73         url = web.ctx['fullpath']
  74         try:
  75             if url.startswith("/"):
  76                 url = url[1:]
  77             response = [ r for r in responses_map[url] if r.http_verb == action][0]
  78             for header, value in response.headers.iteritems():
  79                 web.header(header, value)
  80             web.header('Content-Type', response.content_type)
  81             print(response.body)
  82             return response.body
  83         except:
  84             webapi.NotFound()
  85
  86     def DELETE(self, url):
  87         return self._do_action("delete")
  88
  89     def HEAD(self, url):
  90         return self._do_action("head")
  91
  92     def PUT(self, url):
  93         return self._do_action("put")
  94
  95     def GET(self, url):
  96         return self._do_action("get")
  97
  98     def POST(self, url):
  99         return self._do_action("post")
 100
 101     def PATCH(self, url):
 102         return self._do_action("patch")
 103
 104
 105 logger = setup_logger('mock_controller', '/tmp/generic_sim/output.log')
 106 responses_map = load_responses('/etc/generic_sim/responses.yml')
 107 app = web.application(urls, globals())
 108 if __name__ == "__main__":
 109     app.run()