This repository has been archived by the owner on Mar 24, 2021. It is now read-only.
/
api_test.py
126 lines (84 loc) · 4.01 KB
/
api_test.py
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
import test_helper
import os
import unittest
from dateutil import parser
from flask import json
import pytz
from performance_platform.read import api
DATABASE_NAME = 'performance_platform_test'
def load_fixture(collection_name, fixture_name):
test_path = os.path.join(os.path.dirname(__file__), '..')
fixture_path = os.path.join(test_path, 'fixtures', fixture_name)
with open(fixture_path) as fixture:
for document in json.load(fixture):
if '_timestamp' in document:
document['_timestamp'] = parser.parse(document['_timestamp'])\
.astimezone(pytz.utc)
api.mongo[DATABASE_NAME][collection_name].save(document)
class IntegrationTests(unittest.TestCase):
def setUp(self):
api.app.config['DATABASE_NAME'] = DATABASE_NAME
load_fixture('licencing', 'licence.json')
self.app = api.app.test_client()
def tearDown(self):
api.mongo[DATABASE_NAME]['licencing'].drop()
def create_query(self, **kwargs):
params = '&'.join('{0}={1}'.format(k, v) for k, v in kwargs.items())
return '/licencing?{0}'.format(params)
def test_that_with_no_arguments_all_documents_are_returned(self):
response = self.app.get('/licencing')
self.assertEqual(response.status_code, 200)
response_data = json.loads(response.data)
self.assertEqual(len(response_data['data']), 4)
def test_that_events_later_than_start_at_are_returned(self):
query = self.create_query(start_at='2012-12-12T01:01:02%2B00:00')
response = self.app.get(query)
self.assertEqual(response.status_code, 200)
response_data = json.loads(response.data)
self.assertEqual(len(response_data['data']), 2)
def test_that_events_equal_to_start_at_are_returned(self):
query = self.create_query(start_at='2012-12-13T01:01:01%2B00:00')
response = self.app.get(query)
self.assertEqual(response.status_code, 200)
response_data = json.loads(response.data)
self.assertEqual(len(response_data['data']), 2)
def test_that_events_earlier_than_end_at_are_returned(self):
query = self.create_query(end_at='2012-12-12T01:01:02%2B00:00')
response = self.app.get(query)
self.assertEqual(response.status_code, 200)
response_data = json.loads(response.data)
self.assertEqual(len(response_data['data']), 2)
def test_that_events_equal_to_that_end_at_are_not_returned(self):
query = self.create_query(end_at='2012-12-12T01:01:01%2B00:00')
response = self.app.get(query)
self.assertEqual(response.status_code, 200)
response_data = json.loads(response.data)
self.assertEqual(len(response_data['data']), 0)
def test_that_events_between_start_at_and_end_at_are_returned(self):
start_at = '2012-12-12T01:01:02%2B00:00'
end_at = '2012-12-14T00:00:00%2B00:00'
query = self.create_query(start_at=start_at, end_at=end_at)
response = self.app.get(query)
self.assertEqual(response.status_code, 200)
response_data = json.loads(response.data)
self.assertEqual(len(response_data['data']), 1)
def test_that_events_are_filtered(self):
query = self.create_query(filter_by='authority:Camden')
response = self.app.get(query)
self.assertEqual(response.status_code, 200)
response_data = json.loads(response.data)
self.assertEqual(len(response_data['data']), 1)
def test_that_events_are_grouped(self):
query = self.create_query(group_by='authority')
response = self.app.get(query)
self.assertEqual(response.status_code, 200)
response_data = json.loads(response.data)
self.assertEqual(len(response_data['data']), 2)
self.assertEqual(response_data['data'][0], {
'authority': 'Westminster',
'count': 3
})
self.assertEqual(response_data['data'][1], {
'authority': 'Camden',
'count': 1
})