forked from turbolytics/sql-flow
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathtest_configs.py
53 lines (46 loc) · 1.62 KB
/
test_configs.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
import os
import unittest
from sqlflow import invoke
dev_dir = os.path.join(
os.path.dirname(__file__),
'..',
'dev',
)
conf_dir = os.path.join(dev_dir, 'config')
fixtures_dir = os.path.join(dev_dir, 'fixtures')
class ExamplesTestCase(unittest.TestCase):
def test_basic_agg(self):
out = invoke(
config=os.path.join(conf_dir, 'examples', 'basic.agg.yml'),
fixture=os.path.join(fixtures_dir, 'simple.json'),
)
self.assertEqual([
'{"city":"New York","city_count":28672}',
'{"city":"Baltimore","city_count":28672}',
], out)
def test_csv_filesystem_join(self):
out = invoke(
config=os.path.join(conf_dir, 'examples', 'csv.filesystem.join.yml'),
fixture=os.path.join(fixtures_dir, 'simple.json'),
setting_overrides={
'STATIC_ROOT': dev_dir,
}
)
self.assertEqual([
'{"state_full":"Ohio","city_count":57344}',
'{"state_full":"New York","city_count":1777664}',
'{"state_full":"Maryland","city_count":1232896}',
], out)
def test_csv_mem_join(self):
out = invoke(
config=os.path.join(conf_dir, 'examples', 'csv.mem.join.yml'),
fixture=os.path.join(fixtures_dir, 'simple.json'),
setting_overrides={
'STATIC_ROOT': dev_dir,
}
)
self.assertEqual([
'{"state_full":"Ohio","city_count":57344}',
'{"state_full":"New York","city_count":1777664}',
'{"state_full":"Maryland","city_count":1232896}',
], out)