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
|
# Tests splitting functions.
import types
from io import StringIO
import pytest
import sqlparse
def test_split_semicolon():
sql1 = 'select * from foo;'
sql2 = "select * from foo where bar = 'foo;bar';"
stmts = sqlparse.parse(''.join([sql1, sql2]))
assert len(stmts) == 2
assert str(stmts[0]) == sql1
assert str(stmts[1]) == sql2
def test_split_backslash():
stmts = sqlparse.parse("select '\'; select '\'';")
assert len(stmts) == 2
@pytest.mark.parametrize('fn', ['function.sql',
'function_psql.sql',
'function_psql2.sql',
'function_psql3.sql',
'function_psql4.sql'])
def test_split_create_function(load_file, fn):
sql = load_file(fn)
stmts = sqlparse.parse(sql)
assert len(stmts) == 1
assert str(stmts[0]) == sql
def test_split_dashcomments(load_file):
sql = load_file('dashcomment.sql')
stmts = sqlparse.parse(sql)
assert len(stmts) == 3
assert ''.join(str(q) for q in stmts) == sql
@pytest.mark.parametrize('s', ['select foo; -- comment\n',
'select foo; -- comment\r',
'select foo; -- comment\r\n',
'select foo; -- comment'])
def test_split_dashcomments_eol(s):
stmts = sqlparse.parse(s)
assert len(stmts) == 1
def test_split_begintag(load_file):
sql = load_file('begintag.sql')
stmts = sqlparse.parse(sql)
assert len(stmts) == 3
assert ''.join(str(q) for q in stmts) == sql
def test_split_begintag_2(load_file):
sql = load_file('begintag_2.sql')
stmts = sqlparse.parse(sql)
assert len(stmts) == 1
assert ''.join(str(q) for q in stmts) == sql
def test_split_dropif():
sql = 'DROP TABLE IF EXISTS FOO;\n\nSELECT * FROM BAR;'
stmts = sqlparse.parse(sql)
assert len(stmts) == 2
assert ''.join(str(q) for q in stmts) == sql
def test_split_comment_with_umlaut():
sql = ('select * from foo;\n'
'-- Testing an umlaut: ä\n'
'select * from bar;')
stmts = sqlparse.parse(sql)
assert len(stmts) == 2
assert ''.join(str(q) for q in stmts) == sql
def test_split_comment_end_of_line():
sql = ('select * from foo; -- foo\n'
'select * from bar;')
stmts = sqlparse.parse(sql)
assert len(stmts) == 2
assert ''.join(str(q) for q in stmts) == sql
# make sure the comment belongs to first query
assert str(stmts[0]) == 'select * from foo; -- foo\n'
def test_split_casewhen():
sql = ("SELECT case when val = 1 then 2 else null end as foo;\n"
"comment on table actor is 'The actor table.';")
stmts = sqlparse.split(sql)
assert len(stmts) == 2
def test_split_casewhen_procedure(load_file):
# see issue580
stmts = sqlparse.split(load_file('casewhen_procedure.sql'))
assert len(stmts) == 2
def test_split_cursor_declare():
sql = ('DECLARE CURSOR "foo" AS SELECT 1;\n'
'SELECT 2;')
stmts = sqlparse.split(sql)
assert len(stmts) == 2
def test_split_if_function(): # see issue 33
# don't let IF as a function confuse the splitter
sql = ('CREATE TEMPORARY TABLE tmp '
'SELECT IF(a=1, a, b) AS o FROM one; '
'SELECT t FROM two')
stmts = sqlparse.split(sql)
assert len(stmts) == 2
def test_split_stream():
stream = StringIO("SELECT 1; SELECT 2;")
stmts = sqlparse.parsestream(stream)
assert isinstance(stmts, types.GeneratorType)
assert len(list(stmts)) == 2
def test_split_encoding_parsestream():
stream = StringIO("SELECT 1; SELECT 2;")
stmts = list(sqlparse.parsestream(stream))
assert isinstance(stmts[0].tokens[0].value, str)
def test_split_unicode_parsestream():
stream = StringIO('SELECT ö')
stmts = list(sqlparse.parsestream(stream))
assert str(stmts[0]) == 'SELECT ö'
def test_split_simple():
stmts = sqlparse.split('select * from foo; select * from bar;')
assert len(stmts) == 2
assert stmts[0] == 'select * from foo;'
assert stmts[1] == 'select * from bar;'
def test_split_ignores_empty_newlines():
stmts = sqlparse.split('select foo;\nselect bar;\n')
assert len(stmts) == 2
assert stmts[0] == 'select foo;'
assert stmts[1] == 'select bar;'
def test_split_quotes_with_new_line():
stmts = sqlparse.split('select "foo\nbar"')
assert len(stmts) == 1
assert stmts[0] == 'select "foo\nbar"'
stmts = sqlparse.split("select 'foo\n\bar'")
assert len(stmts) == 1
assert stmts[0] == "select 'foo\n\bar'"
def test_split_mysql_handler_for(load_file):
# see issue581
stmts = sqlparse.split(load_file('mysql_handler.sql'))
assert len(stmts) == 2
@pytest.mark.parametrize('sql, expected', [
('select * from foo;', ['select * from foo']),
('select * from foo', ['select * from foo']),
('select * from foo; select * from bar;', [
'select * from foo',
'select * from bar',
]),
(' select * from foo;\n\nselect * from bar;\n\n\n\n', [
'select * from foo',
'select * from bar',
]),
('select * from foo\n\n; bar', ['select * from foo', 'bar']),
])
def test_split_strip_semicolon(sql, expected):
stmts = sqlparse.split(sql, strip_semicolon=True)
assert len(stmts) == len(expected)
for idx, expectation in enumerate(expected):
assert stmts[idx] == expectation
def test_split_strip_semicolon_procedure(load_file):
stmts = sqlparse.split(load_file('mysql_handler.sql'),
strip_semicolon=True)
assert len(stmts) == 2
assert stmts[0].endswith('end')
assert stmts[1].endswith('end')
@pytest.mark.parametrize('sql, num', [
('USE foo;\nGO\nSELECT 1;\nGO', 4),
('SELECT * FROM foo;\nGO', 2),
('USE foo;\nGO 2\nSELECT 1;', 3)
])
def test_split_go(sql, num): # issue762
stmts = sqlparse.split(sql)
assert len(stmts) == num
def test_split_multiple_case_in_begin(load_file): # issue784
stmts = sqlparse.split(load_file('multiple_case_in_begin.sql'))
assert len(stmts) == 1
|
