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
|
import sqlalchemy as sa
from pytest import raises
from sqlalchemy.dialects import postgresql
from sqlalchemy_utils import Asterisk, row_to_json
from sqlalchemy_utils.expressions import explain, explain_analyze
from tests import TestCase
class ExpressionTestCase(TestCase):
dns = 'postgres://postgres@localhost/sqlalchemy_utils_test'
def create_models(self):
class Article(self.Base):
__tablename__ = 'article'
id = sa.Column(sa.Integer, primary_key=True)
name = sa.Column(sa.Unicode(255))
content = sa.Column(sa.UnicodeText)
self.Article = Article
def assert_startswith(self, query, query_part):
assert str(
query.compile(dialect=postgresql.dialect())
).startswith(query_part)
# Check that query executes properly
self.session.execute(query)
class TestExplain(ExpressionTestCase):
def test_render_explain(self):
self.assert_startswith(
explain(self.session.query(self.Article)),
'EXPLAIN SELECT'
)
def test_render_explain_with_analyze(self):
self.assert_startswith(
explain(self.session.query(self.Article), analyze=True),
'EXPLAIN (ANALYZE true) SELECT'
)
def test_with_string_as_stmt_param(self):
self.assert_startswith(
explain('SELECT 1 FROM article'),
'EXPLAIN SELECT'
)
def test_format(self):
self.assert_startswith(
explain('SELECT 1 FROM article', format='json'),
'EXPLAIN (FORMAT json) SELECT'
)
def test_timing(self):
self.assert_startswith(
explain('SELECT 1 FROM article', analyze=True, timing=False),
'EXPLAIN (ANALYZE true, TIMING false) SELECT'
)
def test_verbose(self):
self.assert_startswith(
explain('SELECT 1 FROM article', verbose=True),
'EXPLAIN (VERBOSE true) SELECT'
)
def test_buffers(self):
self.assert_startswith(
explain('SELECT 1 FROM article', analyze=True, buffers=True),
'EXPLAIN (ANALYZE true, BUFFERS true) SELECT'
)
def test_costs(self):
self.assert_startswith(
explain('SELECT 1 FROM article', costs=False),
'EXPLAIN (COSTS false) SELECT'
)
class TestExplainAnalyze(ExpressionTestCase):
def test_render_explain_analyze(self):
assert str(
explain_analyze(self.session.query(self.Article))
.compile(
dialect=postgresql.dialect()
)
).startswith('EXPLAIN (ANALYZE true) SELECT')
class TestAsterisk(object):
def test_with_table_object(self):
Base = sa.ext.declarative.declarative_base()
class Article(Base):
__tablename__ = 'article'
id = sa.Column(sa.Integer, primary_key=True)
assert str(Asterisk(Article.__table__)) == 'article.*'
def test_with_quoted_identifier(self):
Base = sa.ext.declarative.declarative_base()
class User(Base):
__tablename__ = 'user'
id = sa.Column(sa.Integer, primary_key=True)
assert str(Asterisk(User.__table__).compile(
dialect=postgresql.dialect()
)) == '"user".*'
class TestRowToJson(object):
def test_compiler_with_default_dialect(self):
with raises(sa.exc.CompileError):
str(row_to_json(sa.text('article.*')))
def test_compiler_with_postgresql(self):
assert str(row_to_json(sa.text('article.*')).compile(
dialect=postgresql.dialect()
)) == 'row_to_json(article.*)'
def test_type(self):
assert isinstance(
sa.func.row_to_json(sa.text('article.*')).type,
postgresql.JSON
)
class TestArrayAgg(object):
def test_compiler_with_default_dialect(self):
with raises(sa.exc.CompileError):
str(sa.func.array_agg(sa.text('u.name')))
def test_compiler_with_postgresql(self):
assert str(sa.func.array_agg(sa.text('u.name')).compile(
dialect=postgresql.dialect()
)) == "array_agg(u.name)"
def test_type(self):
assert isinstance(
sa.func.array_agg(sa.text('u.name')).type,
postgresql.ARRAY
)
def test_array_agg_with_default(self):
Base = sa.ext.declarative.declarative_base()
class Article(Base):
__tablename__ = 'article'
id = sa.Column(sa.Integer, primary_key=True)
assert str(sa.func.array_agg(Article.id, [1]).compile(
dialect=postgresql.dialect()
)) == (
'coalesce(array_agg(article.id), CAST(ARRAY[%(param_1)s]'
' AS INTEGER[]))'
)
|
