1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
|
import json
from tornado.web import Application
from tornado.testing import AsyncHTTPTestCase
from opnfv_testapi.router import url_mappings
from opnfv_testapi.resources.models import CreateResponse
import fake_pymongo
class TestBase(AsyncHTTPTestCase):
headers = {'Content-Type': 'application/json; charset=UTF-8'}
def setUp(self):
self.basePath = ''
self.create_res = CreateResponse
self.get_res = None
self.list_res = None
self.update_res = None
self.req_d = None
self.req_e = None
self.addCleanup(self._clear)
super(TestBase, self).setUp()
def get_app(self):
return Application(
url_mappings.mappings,
db=fake_pymongo,
debug=True,
)
def create_d(self, *args):
return self.create(self.req_d, *args)
def create_e(self, *args):
return self.create(self.req_e, *args)
def create(self, req=None, *args):
return self.create_help(self.basePath, req, *args)
def create_help(self, uri, req, *args):
if req:
req = req.format()
res = self.fetch(self._update_uri(uri, *args),
method='POST',
body=json.dumps(req),
headers=self.headers)
return self._get_return(res, self.create_res)
def get(self, *args):
res = self.fetch(self._get_uri(*args),
method='GET',
headers=self.headers)
def inner():
new_args, num = self._get_valid_args(*args)
return self.get_res \
if num != self._need_arg_num(self.basePath) else self.list_res
return self._get_return(res, inner())
def query(self, query):
res = self.fetch(self._get_query_uri(query),
method='GET',
headers=self.headers)
return self._get_return(res, self.list_res)
def update(self, new=None, *args):
if new:
new = new.format()
res = self.fetch(self._get_uri(*args),
method='PUT',
body=json.dumps(new),
headers=self.headers)
return self._get_return(res, self.update_res)
def delete(self, *args):
res = self.fetch(self._get_uri(*args),
method='DELETE',
headers=self.headers)
return res.code, res.body
@staticmethod
def _get_valid_args(*args):
new_args = tuple(['%s' % arg for arg in args if arg is not None])
return new_args, len(new_args)
def _need_arg_num(self, uri):
return uri.count('%s')
def _get_query_uri(self, query):
return self.basePath + '?' + query
def _get_uri(self, *args):
return self._update_uri(self.basePath, *args)
def _update_uri(self, uri, *args):
r_uri = uri
new_args, num = self._get_valid_args(*args)
if num != self._need_arg_num(uri):
r_uri += '/%s'
return r_uri % tuple(['%s' % arg for arg in new_args])
def _get_return(self, res, cls):
code = res.code
body = res.body
return code, self._get_return_body(code, body, cls)
@staticmethod
def _get_return_body(code, body, cls):
return cls.from_dict(json.loads(body)) if code < 300 and cls else body
def assert_href(self, body):
self.assertIn(self.basePath, body.href)
def assert_create_body(self, body, req=None, *args):
if not req:
req = self.req_d
new_args = args + tuple([req.name])
self.assertIn(self._get_uri(*new_args), body.href)
@staticmethod
def _clear():
fake_pymongo.pods.clear()
fake_pymongo.projects.clear()
fake_pymongo.testcases.clear()
fake_pymongo.results.clear()
|