forked from rspeer/dominionstats
-
Notifications
You must be signed in to change notification settings - Fork 17
/
test_incremental_analyze.py
36 lines (28 loc) · 1.04 KB
/
test_incremental_analyze.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
#!/usr/bin/python
import os
import pymongo
import pprint
range_test_datas = [dict(name='r1', ranges=[('20101020', '20101021'),
('20101021', '20101022')]),
dict(name='r2', ranges=[('20101020', '20101022')])]
c = pymongo.Connection()
def test_incremental_build(range_test_data):
ranges = range_test_data['ranges']
col_name = range_test_data['name']
print 'testing', ranges, col_name
c.test.drop_collection(col_name)
for mn, mx in ranges:
cmd = ('python analyze.py --min_date=%s --max_date=%s '
'--output_collection_name=%s' % (mn, mx, col_name))
print cmd
os.system(cmd)
for range_test_data in range_test_datas:
test_incremental_build(range_test_data)
query = {'_id':''}
first_out = c.test[range_test_datas[0]['name']].find_one(query)
assert first_out
for additional_data in range_test_datas[1:]:
name = additional_data['name']
this_out = c.test[name].find_one(query)
assert this_out
assert first_out == this_out