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
|
import io
import os
import sys
from unittest.mock import patch
try:
import psycopg2 # noqa: F401
postgresql_scheme = 'postgresql'
except ImportError:
# http://docs.sqlalchemy.org/en/latest/dialects/postgresql.html#module-sqlalchemy.dialects.postgresql.psycopg2cffi
postgresql_scheme = 'postgresql+psycopg2cffi'
from csvkit.utilities.csvsql import CSVSQL
from csvkit.utilities.sql2csv import SQL2CSV, launch_new_instance
from tests.utils import CSVKitTestCase, EmptyFileTests, stdin_as_string
class TestSQL2CSV(CSVKitTestCase, EmptyFileTests):
Utility = SQL2CSV
def test_launch_new_instance(self):
with patch.object(sys, 'argv', [self.Utility.__name__.lower(), '--query', 'select 1']):
launch_new_instance()
def setUp(self):
self.db_file = 'foo.db'
def tearDown(self):
try:
os.remove(self.db_file)
except OSError:
pass
def csvsql(self, csv_file, db=None):
"""
Load test data into the DB and return it as a string for comparison.
"""
if not db:
db = 'sqlite:///' + self.db_file
args = ['--db', db, '--tables', 'foo', '--insert', csv_file, '--no-inference']
utility = CSVSQL(args)
utility.run()
with open(csv_file) as f:
text = f.read()
return text.strip()
def test_encoding(self):
# Test that csvstat doesn't error on Latin-1 input.
self.get_output(['-e', 'latin1', 'examples/test.sql'])
def test_query(self):
csv = self.get_output(['--query', 'select 6*9 as question'])
self.assertTrue('question' in csv)
self.assertTrue('54' in csv)
def test_file(self):
csv = self.get_output(['examples/test.sql'])
self.assertTrue('question' in csv)
self.assertTrue('36' in csv)
def test_file_with_query(self):
csv = self.get_output(['examples/test.sql', '--query', 'select 6*9 as question'])
self.assertTrue('question' in csv)
self.assertTrue('54' in csv)
def test_stdin(self):
input_file = io.BytesIO(b'select cast(3.1415 * 13.37 as integer) as answer')
with stdin_as_string(input_file):
csv = self.get_output([])
self.assertTrue('answer' in csv)
self.assertTrue('42' in csv)
input_file.close()
def test_stdin_with_query(self):
input_file = io.BytesIO(b'select cast(3.1415 * 13.37 as integer) as answer')
with stdin_as_string(input_file):
csv = self.get_output(['--query', 'select 6*9 as question'])
self.assertTrue('question' in csv)
self.assertTrue('54' in csv)
input_file.close()
def test_stdin_with_file(self):
input_file = io.BytesIO(b'select cast(3.1415 * 13.37 as integer) as answer')
with stdin_as_string(input_file):
csv = self.get_output(['examples/test.sql'])
self.assertTrue('question' in csv)
self.assertTrue('36' in csv)
input_file.close()
def test_stdin_with_file_and_query(self):
input_file = io.BytesIO(b'select cast(3.1415 * 13.37 as integer) as answer')
with stdin_as_string(input_file):
csv = self.get_output(['examples/test.sql', '--query', 'select 6*9 as question'])
self.assertTrue('question' in csv)
self.assertTrue('54' in csv)
input_file.close()
def test_unicode(self):
expected = self.csvsql('examples/test_utf8.csv')
csv = self.get_output(['--db', 'sqlite:///' + self.db_file, '--query', 'select * from foo'])
self.assertEqual(csv.strip(), expected)
def test_no_header_row(self):
self.csvsql('examples/dummy.csv')
csv = self.get_output(['--db', 'sqlite:///' + self.db_file, '--no-header-row', '--query', 'select * from foo'])
self.assertTrue('a,b,c' not in csv)
self.assertTrue('1,2,3' in csv)
def test_linenumbers(self):
self.csvsql('examples/dummy.csv')
csv = self.get_output(['--db', 'sqlite:///' + self.db_file, '--linenumbers', '--query', 'select * from foo'])
self.assertTrue('line_number,a,b,c' in csv)
self.assertTrue('1,1,2,3' in csv)
def test_wildcard_on_sqlite(self):
self.csvsql('examples/iris.csv')
csv = self.get_output(['--db', 'sqlite:///' + self.db_file, '--query',
"select * from foo where species LIKE '%'"])
self.assertTrue('sepal_length,sepal_width,petal_length,petal_width,species' in csv)
self.assertTrue('5.1,3.5,1.4,0.2,Iris-setosa' in csv)
# def test_wildcard_on_postgresql(self):
# db = postgresql_scheme + ':///dummy_test'
# self.csvsql('examples/iris.csv', db)
# csv = self.get_output(['--db', db, '--query', "select * from foo where species LIKE '%'"])
# self.assertTrue('sepal_length,sepal_width,petal_length,petal_width,species' in csv)
# self.assertTrue('5.1,3.5,1.4,0.2,Iris-setosa' in csv)
|
