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