Repository navigation
Expand file tree
/
Copy pathhelpers.py
More file actions
164 lines (139 loc) · 5.64 KB
/
Copy pathhelpers.py
File metadata and controls
164 lines (139 loc) · 5.64 KB
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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
import sys
import os
import argparse
import signal
import unittest
import flask
import time
import requests
import multiprocessing
import traceback
from multiprocessing.util import register_after_fork
from quantifiedcode.app import get_app
from quantifiedcode.settings import settings
from quantifiedcode.backend.models import User, AccessToken, Project
class ApplicationProcess(multiprocessing.Process):
def __init__(self, host, port, get_app, queue):
super(ApplicationProcess,self).__init__()
self.host = host
self.queue = queue
self.port = port
self.get_app = get_app
def run(self):
#we replace the engine getter, as the old one comes from another process
try:
settings.backend = None
application = flask.Flask(__name__)
application.wsgi_app = self.get_app(settings)
except:
traceback.print_exc()
raise
try:
self.queue.put(True)
application.run(
debug=settings.get('debug',True),
host=self.host,
port=self.port,
processes=1,
use_reloader=False
)
except KeyboardInterrupt:
print "Exiting..."
class DatabaseTest(unittest.TestCase):
"""
:attribute recreate_db: Completely recreate the database from scratch
:attribute create_db: Create all tables and triggers if they don't yet exist.
:attribute delete_data: Delete all data currently in the database.
:atttribute models: The models that are relevant for this test
(only they will get deleted from the DB if specified).
Normally, `create_db` and `delete_data` should be sufficient to ensure that the
database is in a good state for running the tests.
However, if the test database has been "messed up" by the user or a test, it should
be reinitialized either manually or by using the `recreate_db` option.
Manual reinitialization can be done via the `reinit-db` management command.
"""
create_db = True
delete_data = True
recreate_db = False
delete_data = True
fixtures = []
models = [AccessToken, User, Project]
def setUp(self):
self.backend = settings.backend
if not TEST:
raise AttributeError("You try to run tests in a non-TEST environment!")
if self.recreate_db:
settings.backend.drop_schema()
if self.create_db or self.recreate_db:
settings.backend.create_schema()
if not self.recreate_db and self.delete_data:
#we delete existing objects from the database...
with settings.backend.transaction():
for model in self.models:
settings.backend.filter(model,{}).delete()
self._create_fixtures()
def _create_fixtures(self):
fixture_objs = {}
for fixture_dict in self.fixtures:
for key,fixture in fixture_dict.items():
objs = fixture(self,fixture_objs)
fixture_objs[key] = objs
setattr(self,key,objs)
def tearDown(self):
"""
we delete the objects from the database to make sure other tests don't fail because of them...
"""
with settings.backend.transaction():
for model in self.models:
settings.backend.filter(model,{}).delete()
class ApplicationTest(DatabaseTest):
"""
"""
get_app = staticmethod(get_app)
host = 'localhost'
port = 5555
protocol = 'http'
base_url = ''
recreate_db = False
create_db = True
delete_data = True
def url(self,name):
return '{}://{}:{}{}{}'.format(self.protocol,
self.host,
self.port,
self.base_url,
name)
def setUp(self):
settings.initialize()
super(ApplicationTest,self).setUp()
self.queue = multiprocessing.Queue()
self.application_process = ApplicationProcess(self.host,self.port,self.get_app,self.queue)
self.application_process.start()
#this parameter might need some tuning to be sure the API is up when the tests start
value = self.queue.get(True, 1.0)
if not value:
raise IOError("Problem starting application!")
time.sleep(0.2)
def tearDown(self):
start = time.time()
self.application_process.terminate()
super(ApplicationTest, self).tearDown()
def _authenticated_request(self, user, func, url, *args, **kwargs):
access_token = user.access_tokens[0]
headers = {
'Authorization' : 'bearer {}'.format(access_token.token)
}
if 'headers' in kwargs:
headers.update(kwargs['headers'])
del kwargs['headers']
return func(self.url(url), *args, headers = headers, **kwargs)
def authenticated_get(self, user, *args, **kwargs):
return self._authenticated_request(user, requests.get, *args, **kwargs)
def authenticated_post(self, user, *args, **kwargs):
return self._authenticated_request(user, requests.post, *args, **kwargs)
def authenticated_delete(self, user, *args, **kwargs):
return self._authenticated_request(user, requests.delete, *args, **kwargs)
def authenticated_put(self, user, *args, **kwargs):
return self._authenticated_request(user, requests.put, *args, **kwargs)
def authenticated_patch(self, user, *args, **kwargs):
return self._authenticated_request(user, requests.patch, *args, **kwargs)