utils/test/testapi/opnfv_testapi/tests/unit/test_base.py

   1 ##############################################################################
   2 # Copyright (c) 2016 ZTE Corporation
   3 # feng.xiaowei@zte.com.cn
   4 # All rights reserved. This program and the accompanying materials
   5 # are made available under the terms of the Apache License, Version 2.0
   6 # which accompanies this distribution, and is available at
   7 # http://www.apache.org/licenses/LICENSE-2.0
   8 ##############################################################################
   9 import json
  10 from os import path
  11
  12 import mock
  13 from tornado import testing
  14
  15 from opnfv_testapi.cmd import server
  16 from opnfv_testapi.resources import models
  17 from opnfv_testapi.tests.unit import fake_pymongo
  18
  19
  20 class TestBase(testing.AsyncHTTPTestCase):
  21     headers = {'Content-Type': 'application/json; charset=UTF-8'}
  22
  23     def setUp(self):
  24         self._patch_server()
  25         self.basePath = ''
  26         self.create_res = models.CreateResponse
  27         self.get_res = None
  28         self.list_res = None
  29         self.update_res = None
  30         self.req_d = None
  31         self.req_e = None
  32         self.addCleanup(self._clear)
  33         super(TestBase, self).setUp()
  34
  35     def tearDown(self):
  36         self.db_patcher.stop()
  37
  38     def _patch_server(self):
  39         server.parse_config([
  40             '--config-file',
  41             path.join(path.dirname(__file__), 'common/normal.ini')
  42         ])
  43         self.db_patcher = mock.patch('opnfv_testapi.cmd.server.get_db',
  44                                      self._fake_pymongo)
  45         self.db_patcher.start()
  46
  47     @staticmethod
  48     def _fake_pymongo():
  49         return fake_pymongo
  50
  51     def get_app(self):
  52         return server.make_app()
  53
  54     def create_d(self, *args):
  55         return self.create(self.req_d, *args)
  56
  57     def create_e(self, *args):
  58         return self.create(self.req_e, *args)
  59
  60     def create(self, req=None, *args):
  61         return self.create_help(self.basePath, req, *args)
  62
  63     def create_help(self, uri, req, *args):
  64         if req and not isinstance(req, str) and hasattr(req, 'format'):
  65             req = req.format()
  66         res = self.fetch(self._update_uri(uri, *args),
  67                          method='POST',
  68                          body=json.dumps(req),
  69                          headers=self.headers)
  70
  71         return self._get_return(res, self.create_res)
  72
  73     def get(self, *args):
  74         res = self.fetch(self._get_uri(*args),
  75                          method='GET',
  76                          headers=self.headers)
  77
  78         def inner():
  79             new_args, num = self._get_valid_args(*args)
  80             return self.get_res \
  81                 if num != self._need_arg_num(self.basePath) else self.list_res
  82         return self._get_return(res, inner())
  83
  84     def query(self, query):
  85         res = self.fetch(self._get_query_uri(query),
  86                          method='GET',
  87                          headers=self.headers)
  88         return self._get_return(res, self.list_res)
  89
  90     def update(self, new=None, *args):
  91         if new:
  92             new = new.format()
  93         res = self.fetch(self._get_uri(*args),
  94                          method='PUT',
  95                          body=json.dumps(new),
  96                          headers=self.headers)
  97         return self._get_return(res, self.update_res)
  98
  99     def delete(self, *args):
 100         res = self.fetch(self._get_uri(*args),
 101                          method='DELETE',
 102                          headers=self.headers)
 103         return res.code, res.body
 104
 105     @staticmethod
 106     def _get_valid_args(*args):
 107         new_args = tuple(['%s' % arg for arg in args if arg is not None])
 108         return new_args, len(new_args)
 109
 110     def _need_arg_num(self, uri):
 111         return uri.count('%s')
 112
 113     def _get_query_uri(self, query):
 114         return self.basePath + '?' + query if query else self.basePath
 115
 116     def _get_uri(self, *args):
 117         return self._update_uri(self.basePath, *args)
 118
 119     def _update_uri(self, uri, *args):
 120         r_uri = uri
 121         new_args, num = self._get_valid_args(*args)
 122         if num != self._need_arg_num(uri):
 123             r_uri += '/%s'
 124
 125         return r_uri % tuple(['%s' % arg for arg in new_args])
 126
 127     def _get_return(self, res, cls):
 128         code = res.code
 129         body = res.body
 130         return code, self._get_return_body(code, body, cls)
 131
 132     @staticmethod
 133     def _get_return_body(code, body, cls):
 134         return cls.from_dict(json.loads(body)) if code < 300 and cls else body
 135
 136     def assert_href(self, body):
 137         self.assertIn(self.basePath, body.href)
 138
 139     def assert_create_body(self, body, req=None, *args):
 140         import inspect
 141         if not req:
 142             req = self.req_d
 143         resource_name = ''
 144         if inspect.isclass(req):
 145             resource_name = req.name
 146         elif isinstance(req, dict):
 147             resource_name = req['name']
 148         elif isinstance(req, str):
 149             resource_name = json.loads(req)['name']
 150         new_args = args + tuple([resource_name])
 151         self.assertIn(self._get_uri(*new_args), body.href)
 152
 153     @staticmethod
 154     def _clear():
 155         fake_pymongo.pods.clear()
 156         fake_pymongo.projects.clear()
 157         fake_pymongo.testcases.clear()
 158         fake_pymongo.results.clear()
 159         fake_pymongo.scenarios.clear()