utils/test/testapi/opnfv_testapi/tests/unit/resources/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.resources import models
  16 from opnfv_testapi.tests.unit import fake_pymongo
  17
  18
  19 class TestBase(testing.AsyncHTTPTestCase):
  20     headers = {'Content-Type': 'application/json; charset=UTF-8'}
  21
  22     def setUp(self):
  23         self._patch_server()
  24         self.basePath = ''
  25         self.create_res = models.CreateResponse
  26         self.get_res = None
  27         self.list_res = None
  28         self.update_res = None
  29         self.req_d = None
  30         self.req_e = None
  31         self.addCleanup(self._clear)
  32         super(TestBase, self).setUp()
  33
  34     def tearDown(self):
  35         self.db_patcher.stop()
  36         self.config_patcher.stop()
  37
  38     def _patch_server(self):
  39         import argparse
  40         config = path.join(path.dirname(__file__), '../common/normal.ini')
  41         self.config_patcher = mock.patch(
  42             'argparse.ArgumentParser.parse_known_args',
  43             return_value=(argparse.Namespace(config_file=config), None))
  44         self.db_patcher = mock.patch('opnfv_testapi.db.api.DB',
  45                                      fake_pymongo)
  46         self.config_patcher.start()
  47         self.db_patcher.start()
  48
  49     def set_config_file(self):
  50         self.config_file = 'normal.ini'
  51
  52     def get_app(self):
  53         from opnfv_testapi.cmd import server
  54         return server.make_app()
  55
  56     def create_d(self, *args):
  57         return self.create(self.req_d, *args)
  58
  59     def create_e(self, *args):
  60         return self.create(self.req_e, *args)
  61
  62     def create(self, req=None, *args):
  63         return self.create_help(self.basePath, req, *args)
  64
  65     def create_help(self, uri, req, *args):
  66         return self.post_direct_url(self._update_uri(uri, *args), req)
  67
  68     def post_direct_url(self, url, req):
  69         if req and not isinstance(req, str) and hasattr(req, 'format'):
  70             req = req.format()
  71         res = self.fetch(url,
  72                          method='POST',
  73                          body=json.dumps(req),
  74                          headers=self.headers)
  75
  76         return self._get_return(res, self.create_res)
  77
  78     def get(self, *args):
  79         res = self.fetch(self._get_uri(*args),
  80                          method='GET',
  81                          headers=self.headers)
  82
  83         def inner():
  84             new_args, num = self._get_valid_args(*args)
  85             return self.get_res \
  86                 if num != self._need_arg_num(self.basePath) else self.list_res
  87         return self._get_return(res, inner())
  88
  89     def query(self, query):
  90         res = self.fetch(self._get_query_uri(query),
  91                          method='GET',
  92                          headers=self.headers)
  93         return self._get_return(res, self.list_res)
  94
  95     def update_direct_url(self, url, new=None):
  96         if new and hasattr(new, 'format'):
  97             new = new.format()
  98         res = self.fetch(url,
  99                          method='PUT',
 100                          body=json.dumps(new),
 101                          headers=self.headers)
 102         return self._get_return(res, self.update_res)
 103
 104     def update(self, new=None, *args):
 105         return self.update_direct_url(self._get_uri(*args), new)
 106
 107     def delete_direct_url(self, url, body):
 108         if body:
 109             res = self.fetch(url,
 110                              method='DELETE',
 111                              body=json.dumps(body),
 112                              headers=self.headers,
 113                              allow_nonstandard_methods=True)
 114         else:
 115             res = self.fetch(url,
 116                              method='DELETE',
 117                              headers=self.headers)
 118
 119         return res.code, res.body
 120
 121     def delete(self, *args):
 122         return self.delete_direct_url(self._get_uri(*args), None)
 123
 124     @staticmethod
 125     def _get_valid_args(*args):
 126         new_args = tuple(['%s' % arg for arg in args if arg is not None])
 127         return new_args, len(new_args)
 128
 129     def _need_arg_num(self, uri):
 130         return uri.count('%s')
 131
 132     def _get_query_uri(self, query):
 133         return self.basePath + '?' + query if query else self.basePath
 134
 135     def _get_uri(self, *args):
 136         return self._update_uri(self.basePath, *args)
 137
 138     def _update_uri(self, uri, *args):
 139         r_uri = uri
 140         new_args, num = self._get_valid_args(*args)
 141         if num != self._need_arg_num(uri):
 142             r_uri += '/%s'
 143
 144         return r_uri % tuple(['%s' % arg for arg in new_args])
 145
 146     def _get_return(self, res, cls):
 147         code = res.code
 148         body = res.body
 149         if body:
 150             return code, self._get_return_body(code, body, cls)
 151         else:
 152             return code, None
 153
 154     @staticmethod
 155     def _get_return_body(code, body, cls):
 156         return cls.from_dict(json.loads(body)) if code < 300 and cls else body
 157
 158     def assert_href(self, body):
 159         self.assertIn(self.basePath, body.href)
 160
 161     def assert_create_body(self, body, req=None, *args):
 162         import inspect
 163         if not req:
 164             req = self.req_d
 165         resource_name = ''
 166         if inspect.isclass(req):
 167             resource_name = req.name
 168         elif isinstance(req, dict):
 169             resource_name = req['name']
 170         elif isinstance(req, str):
 171             resource_name = json.loads(req)['name']
 172         new_args = args + tuple([resource_name])
 173         self.assertIn(self._get_uri(*new_args), body.href)
 174
 175     @staticmethod
 176     def _clear():
 177         fake_pymongo.pods.clear()
 178         fake_pymongo.projects.clear()
 179         fake_pymongo.testcases.clear()
 180         fake_pymongo.results.clear()
 181         fake_pymongo.scenarios.clear()