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
|
# -*- coding: utf-8 -*-
# :Project: pglast -- Test the parser.pyx module
# :Created: ven 04 ago 2017 08:37:10 CEST
# :Author: Lele Gaifax <lele@metapensiero.it>
# :License: GNU General Public License version 3 or later
# :Copyright: © 2017, 2018, 2019, 2021 Lele Gaifax
#
import pytest
from pglast import Error, ast, parse_plpgsql, parse_sql
from pglast.parser import ParseError, fingerprint, get_postgresql_version, scan, split
from pglast.parser import deparse_protobuf, parse_sql_protobuf
def test_parse_sql():
assert parse_sql('') == ()
assert parse_sql('-- nothing') == ()
with pytest.raises(ParseError):
parse_sql('foo')
ptree = parse_sql('SELECT 1')
assert isinstance(ptree, tuple)
assert len(ptree) == 1
rawstmt = ptree[0]
assert isinstance(rawstmt, ast.RawStmt)
def test_parse_plpgsql():
ptree = parse_plpgsql('CREATE FUNCTION add (a integer, b integer)'
' RETURNS integer AS $$ BEGIN RETURN a + b; END; $$'
' LANGUAGE plpgsql')
assert len(ptree) == 1
function = ptree[0]
assert isinstance(function, dict)
assert function.keys() == {'PLpgSQL_function'}
# See https://github.com/pganalyze/libpg_query/issues/122
ptree = parse_plpgsql("""\
CREATE OR REPLACE FUNCTION test_parse (
p_time_start timestamptz,
p_time_end timestamptz,
p_time_interval interval default NULL
) RETURNS TABLE (
ts timestamptz,
arbitrary_return bigint
) AS $$
BEGIN
IF p_time_interval IS NULL
THEN p_time_interval := INTERVAL '1 hour';
END IF;
RETURN QUERY
SELECT p_time_start + p_time_interval, 1234::bigint;
END; $$ LANGUAGE plpgsql;""")
function = ptree[0]
assert isinstance(function, dict)
assert function.keys() == {'PLpgSQL_function'}
def test_fingerprint():
sql1 = "SELECT a as b, c as d FROM atable AS btable WHERE a = 1 AND b in (1, 2)"
sql2 = "SELECT a, c FROM atable WHERE a = 2 AND b IN (2, 3, 4) "
assert fingerprint(sql1) == fingerprint(sql2)
def test_errors():
with pytest.raises(Error) as exc:
parse_sql('FooBar')
assert exc.typename == 'ParseError'
errmsg, index = exc.value.args
assert errmsg == 'syntax error at or near "FooBar"'
assert index == 0
with pytest.raises(Error) as exc:
parse_sql('SELECT foo FRON bar')
assert exc.typename == 'ParseError'
errmsg, index = exc.value.args
assert errmsg == 'syntax error at or near "bar"'
assert index == 16
assert str(exc.value) == f'{errmsg}, at index {index}'
with pytest.raises(Error) as exc:
parse_plpgsql('CREATE FUMCTION add (a integer, b integer)'
' RETURNS integer AS $$ BEGIN RETURN a + b; END; $$'
' LANGUAGE plpgsql')
assert exc.typename == 'ParseError'
errmsg, index = exc.value.args
assert errmsg == 'syntax error at or near "FUMCTION"'
assert index == 7
with pytest.raises(Error) as exc:
fingerprint('SELECT foo FRON bar')
assert exc.typename == 'ParseError'
errmsg, index = exc.value.args
assert errmsg == 'syntax error at or near "bar"'
assert index == 16
def test_unicode():
ptree = parse_sql('SELECT 1 AS "Naïve"')
target = ptree[0].stmt.targetList[0]
assert target.name == "Naïve"
def test_locations_fixup():
sql = 'SELECT 1 AS "Naïve" /* there is an "ı" with a \u0308 above */ FROM somewhere'
sql3 = ';'.join([sql]*3)
ptree = parse_sql(sql3)
assert len(ptree) == 3
raw = ptree[0]
assert raw.stmt_location == 0
assert raw.stmt_len == len(sql)
fromc = raw.stmt.fromClause[0]
assert sql3[fromc.location:].startswith('somewhere')
raw = ptree[1]
assert raw.stmt_location == len(sql)+1
assert raw.stmt_len == len(sql)
fromc = raw.stmt.fromClause[0]
assert sql3[fromc.location:].startswith('somewhere')
raw = ptree[2]
assert raw.stmt_location == len(sql)*2+2
# For the last stmt, the stmt_len is 0...
assert raw.stmt_len == 0
fromc = raw.stmt.fromClause[0]
assert sql3[fromc.location:].startswith('somewhere')
def test_pg_version():
pg_version = get_postgresql_version()
assert isinstance(pg_version, tuple)
assert len(pg_version) == 2
def test_clone():
from pglast import ast
stmts = parse_sql('SELECT 1')
stmt = stmts[0].stmt
clone = ast.SelectStmt(stmt())
assert clone is not stmt
assert clone == stmt
assert repr(clone) == repr(stmt)
assert clone() == stmt()
def test_split():
sql = 'select 1; select 2; select "€€€€ ·"; select 4'
expected = ('select 1', 'select 2', 'select "€€€€ ·"', 'select 4')
assert split(sql) == expected
assert tuple(sql[s] for s in split(sql, only_slices=True)) == expected
def test_scan():
sql = 'select /* something here */ 1'
result = scan(sql)
assert result == [( 0, 5, 'SELECT', 'RESERVED_KEYWORD'), # noqa E201
( 7, 26, 'C_COMMENT', 'NO_KEYWORD'), # noqa E201
(28, 28, 'ICONST', 'NO_KEYWORD')]
assert sql[result[1].start:result[1].end+1] == '/* something here */'
sql = 'select 0.01 as "€" -- one €-cent'
assert [sql[t.start:t.end+1] for t in scan(sql)] == [
'select', '0.01', 'as', '"€"', '-- one €-cent']
# Combining character
sql = 'SELECT 1 AS "\u0101\u0301" -- etc'
assert [sql[t.start:t.end+1] for t in scan(sql)] == [
'SELECT', '1', 'AS', '"ā́"', '-- etc']
# Invalid input, see https://github.com/pganalyze/libpg_query/issues/108
sql = 'SELECT \\s 1'
result = scan(sql)
assert result == [( 0, 5, 'SELECT', 'RESERVED_KEYWORD'), # noqa E201
( 7, 7, 'ASCII_92', 'NO_KEYWORD'), # noqa E201
( 8, 8, 'IDENT', 'NO_KEYWORD'), # noqa E201
(10, 10, 'ICONST', 'NO_KEYWORD')]
assert sql[result[1].start] == '\\'
def test_deparse_protobuf():
assert deparse_protobuf(parse_sql_protobuf('select 1')) == 'SELECT 1'
|
