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
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
|
import json
import os
from app import manager
from flask import Blueprint
from flask_testing import TestCase
manager.app.tests_blueprint = Blueprint("tests", __name__)
manager.load_views(manager.app.tests_blueprint, "pysite/views/tests")
manager.app.register_blueprint(manager.app.tests_blueprint)
app = manager.app
class SiteTest(TestCase):
""" extend TestCase with flask app instantiation """
def create_app(self):
""" add flask app configuration settings """
server_name = 'pytest.local'
app.config['TESTING'] = True
app.config['LIVESERVER_TIMEOUT'] = 10
app.config['SERVER_NAME'] = server_name
app.config['API_SUBDOMAIN'] = f'http://api.{server_name}'
app.config['STAFF_SUBDOMAIN'] = f'http://staff.{server_name}'
app.allow_subdomain_redirects = True
return app
class BaseEndpoints(SiteTest):
""" test cases for the base endpoints """
def test_index(self):
""" Check the root path reponds with 200 OK """
response = self.client.get('/', 'http://pytest.local')
self.assertEqual(response.status_code, 200)
def test_help(self):
""" Check the help path responds with 200 OK """
response = self.client.get('/help')
self.assertEqual(response.status_code, 200)
def test_not_found(self):
""" Check paths without handlers returns 404 Not Found """
response = self.client.get('/nonexistentpath')
self.assertEqual(response.status_code, 404)
def test_invite(self):
""" Check invite redirects """
response = self.client.get('/invite')
self.assertEqual(response.status_code, 302)
def test_ws_test(self):
""" check ws_test responds """
response = self.client.get('/ws_test')
self.assertEqual(response.status_code, 200)
def test_datadog_redirect(self):
""" Check datadog path redirects """
response = self.client.get('/datadog')
self.assertEqual(response.status_code, 302)
class ApiEndpoints(SiteTest):
""" test cases for the api subdomain """
def test_api_unknown_route(self):
""" Check api unknown route """
response = self.client.get('/', app.config['API_SUBDOMAIN'])
self.assertEqual(response.json, {'error_code': 0, 'error_message': 'Unknown API route'})
self.assertEqual(response.status_code, 404)
def test_api_healthcheck(self):
""" Check healthcheck url responds """
response = self.client.get('/healthcheck', app.config['API_SUBDOMAIN'])
self.assertEqual(response.json, {'status': 'ok'})
self.assertEqual(response.status_code, 200)
def test_api_tag(self):
""" Check tag api """
os.environ['BOT_API_KEY'] = 'abcdefg'
headers = {'X-API-Key': 'abcdefg', 'Content-Type': 'application/json'}
good_data = json.dumps({
'tag_name': 'testing',
'tag_content': 'testing',
'tag_category': 'testing'})
bad_data = json.dumps({
'not_a_valid_key': 'testing',
'tag_content': 'testing',
'tag_category': 'testing'})
response = self.client.get('/tag', app.config['API_SUBDOMAIN'])
self.assertEqual(response.status_code, 401)
response = self.client.get('/tag', app.config['API_SUBDOMAIN'], headers=headers)
self.assertEqual(response.status_code, 200)
response = self.client.post('/tag', app.config['API_SUBDOMAIN'], headers=headers, data=bad_data)
self.assertEqual(response.status_code, 400)
response = self.client.post('/tag', app.config['API_SUBDOMAIN'], headers=headers, data=good_data)
self.assertEqual(response.json, {'success': True})
response = self.client.get('/tag', app.config['API_SUBDOMAIN'], headers=headers, data=good_data)
self.assertEqual(response.json, [{'tag_name': 'testing'}])
self.assertEqual(response.status_code, 200)
def test_api_user(self):
""" Check insert user """
os.environ['BOT_API_KEY'] = 'abcdefg'
headers = {'X-API-Key': 'abcdefg', 'Content-Type': 'application/json'}
bad_data = json.dumps({'user_id': 1234, 'role': 5678})
good_data = json.dumps([{'user_id': 1234, 'role': 5678}])
response = self.client.get('/user', app.config['API_SUBDOMAIN'], headers=headers)
self.assertEqual(response.status_code, 405)
response = self.client.post('/user', app.config['API_SUBDOMAIN'], headers=headers, data=bad_data)
self.assertEqual(response.json, {'error_code': 3, 'error_message': 'Incorrect parameters provided'})
response = self.client.post('/user', app.config['API_SUBDOMAIN'], headers=headers, data=good_data)
self.assertEqual(True, "inserted" in response.json)
def test_api_route_errors(self):
""" Check api route errors """
from pysite.base_route import APIView
from pysite.constants import ErrorCodes
av = APIView()
av.error(ErrorCodes.unauthorized)
av.error(ErrorCodes.bad_data_format)
def test_not_found(self):
""" Check paths without handlers returns 404 Not Found """
response = self.client.get('/nonexistentpath')
self.assertEqual(response.status_code, 404)
class StaffEndpoints(SiteTest):
""" Test cases for staff subdomain """
def test_staff_view(self):
""" Check staff view renders """
from pysite.views.staff.index import StaffView
sv = StaffView()
result = sv.get()
self.assertEqual(result.startswith('<!DOCTYPE html>'), True)
response = self.client.get('/', app.config['STAFF_SUBDOMAIN'])
self.assertEqual(response.status_code, 200)
class Utilities(SiteTest):
""" Test cases for internal utility code """
def test_logging_runtime_error(self):
""" Check that a wrong value for log level raises runtime error """
os.environ['LOG_LEVEL'] = 'wrong value'
try:
import pysite.__init__ # noqa: F401
except RuntimeError:
return True
finally:
os.environ['LOG_LEVEL'] = 'info'
raise Exception('Expected a failure due to wrong LOG_LEVEL attribute name')
def test_error_view_runtime_error(self):
""" Check that wrong values for error view setup raises runtime error """
import pysite.base_route
ev = pysite.base_route.ErrorView()
try:
ev.setup('sdf', 'sdfsdf')
except RuntimeError:
return True
raise Exception('Expected runtime error on setup() when giving wrongful arguments')
class MixinTests(SiteTest):
""" Test cases for mixins """
def test_dbmixin_runtime_error(self):
""" Check that wrong values for error view setup raises runtime error """
from pysite.mixins import DBMixin
dbm = DBMixin()
try:
dbm.setup('sdf', 'sdfsdf')
except RuntimeError:
return True
raise Exception('Expected runtime error on setup() when giving wrongful arguments')
def test_dbmixin_table_property(self):
""" Check the table property returns correctly """
from pysite.mixins import DBMixin
try:
dbm = DBMixin()
dbm.table_name = 'Table'
self.assertEquals(dbm.table, 'Table')
except AttributeError:
pass
def test_handler_5xx(self):
""" Check error view returns error message """
from werkzeug.exceptions import InternalServerError
from pysite.views.error_handlers import http_5xx
error_view = http_5xx.Error404View()
error_message = error_view.get(InternalServerError)
self.assertEqual(error_message, ('Internal server error. Please try again later!', 500))
def test_route_view_runtime_error(self):
""" Check that wrong values for route view setup raises runtime error """
from pysite.base_route import RouteView
rv = RouteView()
try:
rv.setup('sdf', 'sdfsdf')
except RuntimeError:
return True
raise Exception('Expected runtime error on setup() when giving wrongful arguments')
def test_route_manager(self):
""" Check route manager """
from pysite.route_manager import RouteManager
os.environ['WEBPAGE_SECRET_KEY'] = 'super_secret'
rm = RouteManager()
self.assertEqual(rm.app.secret_key, 'super_secret')
class DecoratorTests(SiteTest):
def test_decorator_api_json(self):
""" Check the json validation decorator """
from pysite.decorators import api_params
from pysite.constants import ValidationTypes
from schema import Schema
SCHEMA = Schema([{"user_id": int, "role": int}])
@api_params(schema=SCHEMA, validation_type=ValidationTypes.json)
def try_json_type(data):
return data
try:
try_json_type("not json")
except Exception as error_message:
self.assertEqual(type(error_message), AttributeError)
def test_decorator_params(self):
""" Check the params validation decorator """
response = self.client.post('/testparams?test=params')
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json, [{'test': 'params'}])
class DatabaseTests(SiteTest):
""" Test cases for the database module """
def test_table_actions(self):
import string
import secrets
from pysite.database import RethinkDB
alphabet = string.ascii_letters
generated_table_name = ''.join(secrets.choice(alphabet) for i in range(8))
rdb = RethinkDB()
# Create table name and expect it to work
result = rdb.create_table(generated_table_name)
self.assertEquals(result, True)
# Create the same table name and expect it to already exist
result = rdb.create_table(generated_table_name)
self.assertEquals(result, False)
# Drop table and expect it to work
result = rdb.drop_table(generated_table_name)
self.assertEquals(result, True)
# Drop the same table and expect it to already be gone
result = rdb.drop_table(generated_table_name)
self.assertEquals(result, False)
# This is to get some more code coverage
self.assertEquals(rdb.teardown_request('_'), None)
class TestWebsocketEcho(SiteTest):
""" Test cases for the echo endpoint """
def testEcho(self):
""" Check rudimentary websockets handlers work """
from geventwebsocket.websocket import WebSocket
from pysite.views.ws.echo import EchoWebsocket
ew = EchoWebsocket(WebSocket)
ew.on_open()
ew.on_message('message')
ew.on_close()
|