FilesExpand file tree

 master

/

test_split.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

205 lines (154 loc) · 5.85 KB

 master

/

test_split.py

Copy path

Top

File metadata and controls

• Code
• Blame

205 lines (154 loc) · 5.85 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

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

# 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