FilesExpand file tree

 master

/

test_split.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

151 lines (110 loc) · 4.26 KB

 master

/

test_split.py

Copy path

Top

File metadata and controls

• Code
• Blame

151 lines (110 loc) · 4.26 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

# -*- coding: utf-8 -*-

# Tests splitting functions.

import types

import pytest

import sqlparse

from sqlparse.compat import StringIO, text_type

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(r"select '\\'; select '\''; select '\\\'';")

assert len(stmts) == 3

@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 text_type(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 = (u'select * from foo;\n'

u'-- Testing an umlaut: ä\n'

u'select * from bar;')

stmts = sqlparse.parse(sql)

assert len(stmts) == 2

assert ''.join(text_type(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_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, text_type)

def test_split_unicode_parsestream():

stream = StringIO(u'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_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'"