← Back to branch summary

~ubuntu-branches/debian/jessie/sqlalchemy/jessie

« back to all changes in this revision

Viewing changes to test/sql/test_type_expressions.py

  • Committer: Package Import Robot
  • Author(s): Piotr Ożarowski, Jakub Wilk, Piotr Ożarowski
  • Date: 2013-07-06 20:53:52 UTC
  • mfrom: (1.4.23) (16.1.17 experimental)
  • Revision ID: package-import@ubuntu.com-20130706205352-ryppl1eto3illd79
Tags: 0.8.2-1
[ Jakub Wilk ]
* Use canonical URIs for Vcs-* fields.

[ Piotr Ożarowski ]
* New upstream release
* Upload to unstable
* Build depend on python3-all instead of -dev, extensions are not built for
  Python 3.X 

Show diffs side-by-side

added added

removed removed

Lines of Context:
test/sql/test_type_expressions.py
 
1
from sqlalchemy import Table, Column, String, func, MetaData, select, TypeDecorator, cast
 
2
from sqlalchemy.testing import fixtures, AssertsCompiledSQL
 
3
from sqlalchemy import testing
 
4
from sqlalchemy.testing import eq_
 
5
 
 
6
 
 
7
class _ExprFixture(object):
 
8
    def _fixture(self):
 
9
        class MyString(String):
 
10
            def bind_expression(self, bindvalue):
 
11
                return func.lower(bindvalue)
 
12
 
 
13
            def column_expression(self, col):
 
14
                return func.lower(col)
 
15
 
 
16
        test_table = Table(
 
17
                'test_table',
 
18
                MetaData(), Column('x', String), Column('y', MyString)
 
19
        )
 
20
        return test_table
 
21
 
 
22
class SelectTest(_ExprFixture, fixtures.TestBase, AssertsCompiledSQL):
 
23
    __dialect__ = 'default'
 
24
 
 
25
    def test_select_cols(self):
 
26
        table = self._fixture()
 
27
 
 
28
        self.assert_compile(
 
29
            select([table]),
 
30
            "SELECT test_table.x, lower(test_table.y) AS y FROM test_table"
 
31
        )
 
32
 
 
33
    def test_anonymous_expr(self):
 
34
        table = self._fixture()
 
35
        self.assert_compile(
 
36
            select([cast(table.c.y, String)]),
 
37
            "SELECT CAST(test_table.y AS VARCHAR) AS anon_1 FROM test_table"
 
38
        )
 
39
 
 
40
    def test_select_cols_use_labels(self):
 
41
        table = self._fixture()
 
42
 
 
43
        self.assert_compile(
 
44
            select([table]).apply_labels(),
 
45
            "SELECT test_table.x AS test_table_x, "
 
46
            "lower(test_table.y) AS test_table_y FROM test_table"
 
47
        )
 
48
 
 
49
    def test_select_cols_use_labels_result_map_targeting(self):
 
50
        table = self._fixture()
 
51
 
 
52
        compiled = select([table]).apply_labels().compile()
 
53
        assert table.c.y in compiled.result_map['test_table_y'][1]
 
54
        assert table.c.x in compiled.result_map['test_table_x'][1]
 
55
 
 
56
        # the lower() function goes into the result_map, we don't really
 
57
        # need this but it's fine
 
58
        self.assert_compile(
 
59
            compiled.result_map['test_table_y'][1][2],
 
60
            "lower(test_table.y)"
 
61
        )
 
62
        # then the original column gets put in there as well.
 
63
        # it's not important that it's the last value.
 
64
        self.assert_compile(
 
65
            compiled.result_map['test_table_y'][1][-1],
 
66
            "test_table.y"
 
67
        )
 
68
 
 
69
    def test_insert_binds(self):
 
70
        table = self._fixture()
 
71
 
 
72
        self.assert_compile(
 
73
                table.insert(),
 
74
                "INSERT INTO test_table (x, y) VALUES (:x, lower(:y))"
 
75
        )
 
76
 
 
77
        self.assert_compile(
 
78
                table.insert().values(y="hi"),
 
79
                "INSERT INTO test_table (y) VALUES (lower(:y))"
 
80
        )
 
81
 
 
82
    def test_select_binds(self):
 
83
        table = self._fixture()
 
84
        self.assert_compile(
 
85
            select([table]).where(table.c.y == "hi"),
 
86
            "SELECT test_table.x, lower(test_table.y) AS y FROM "
 
87
            "test_table WHERE test_table.y = lower(:y_1)"
 
88
        )
 
89
 
 
90
class DerivedTest(_ExprFixture, fixtures.TestBase, AssertsCompiledSQL):
 
91
    __dialect__ = 'default'
 
92
 
 
93
    def test_select_from_select(self):
 
94
        table = self._fixture()
 
95
        self.assert_compile(
 
96
            table.select().select(),
 
97
            "SELECT x, lower(y) AS y FROM (SELECT test_table.x "
 
98
                "AS x, test_table.y AS y FROM test_table)"
 
99
        )
 
100
 
 
101
    def test_select_from_alias(self):
 
102
        table = self._fixture()
 
103
        self.assert_compile(
 
104
            table.select().alias().select(),
 
105
            "SELECT anon_1.x, lower(anon_1.y) AS y FROM (SELECT "
 
106
                "test_table.x AS x, test_table.y AS y "
 
107
                "FROM test_table) AS anon_1"
 
108
        )
 
109
 
 
110
    def test_select_from_aliased_join(self):
 
111
        table = self._fixture()
 
112
        s1 = table.select().alias()
 
113
        s2 = table.select().alias()
 
114
        j = s1.join(s2, s1.c.x == s2.c.x)
 
115
        s3 = j.select()
 
116
        self.assert_compile(s3,
 
117
            "SELECT anon_1.x, lower(anon_1.y) AS y, anon_2.x, "
 
118
            "lower(anon_2.y) AS y "
 
119
            "FROM (SELECT test_table.x AS x, test_table.y AS y "
 
120
            "FROM test_table) AS anon_1 JOIN (SELECT "
 
121
            "test_table.x AS x, test_table.y AS y "
 
122
            "FROM test_table) AS anon_2 ON anon_1.x = anon_2.x"
 
123
        )
 
124
 
 
125
class RoundTripTestBase(object):
 
126
    def test_round_trip(self):
 
127
        testing.db.execute(
 
128
            self.tables.test_table.insert(),
 
129
            {"x": "X1", "y": "Y1"},
 
130
            {"x": "X2", "y": "Y2"},
 
131
            {"x": "X3", "y": "Y3"},
 
132
        )
 
133
 
 
134
        # test insert coercion alone
 
135
        eq_(
 
136
            testing.db.execute(
 
137
                "select * from test_table order by y").fetchall(),
 
138
            [
 
139
                ("X1", "y1"),
 
140
                ("X2", "y2"),
 
141
                ("X3", "y3"),
 
142
            ]
 
143
        )
 
144
 
 
145
        # conversion back to upper
 
146
        eq_(
 
147
            testing.db.execute(
 
148
                select([self.tables.test_table]).\
 
149
                order_by(self.tables.test_table.c.y)
 
150
            ).fetchall(),
 
151
            [
 
152
                ("X1", "Y1"),
 
153
                ("X2", "Y2"),
 
154
                ("X3", "Y3"),
 
155
            ]
 
156
        )
 
157
 
 
158
    def test_targeting_no_labels(self):
 
159
        testing.db.execute(
 
160
            self.tables.test_table.insert(),
 
161
            {"x": "X1", "y": "Y1"},
 
162
        )
 
163
        row = testing.db.execute(select([self.tables.test_table])).first()
 
164
        eq_(
 
165
            row[self.tables.test_table.c.y],
 
166
            "Y1"
 
167
        )
 
168
 
 
169
    def test_targeting_by_string(self):
 
170
        testing.db.execute(
 
171
            self.tables.test_table.insert(),
 
172
            {"x": "X1", "y": "Y1"},
 
173
        )
 
174
        row = testing.db.execute(select([self.tables.test_table])).first()
 
175
        eq_(
 
176
            row["y"],
 
177
            "Y1"
 
178
        )
 
179
 
 
180
    def test_targeting_apply_labels(self):
 
181
        testing.db.execute(
 
182
            self.tables.test_table.insert(),
 
183
            {"x": "X1", "y": "Y1"},
 
184
        )
 
185
        row = testing.db.execute(select([self.tables.test_table]).
 
186
                        apply_labels()).first()
 
187
        eq_(
 
188
            row[self.tables.test_table.c.y],
 
189
            "Y1"
 
190
        )
 
191
 
 
192
    def test_targeting_individual_labels(self):
 
193
        testing.db.execute(
 
194
            self.tables.test_table.insert(),
 
195
            {"x": "X1", "y": "Y1"},
 
196
        )
 
197
        row = testing.db.execute(select([
 
198
                    self.tables.test_table.c.x.label('xbar'),
 
199
                    self.tables.test_table.c.y.label('ybar')
 
200
                ])).first()
 
201
        eq_(
 
202
            row[self.tables.test_table.c.y],
 
203
            "Y1"
 
204
        )
 
205
 
 
206
class StringRoundTripTest(fixtures.TablesTest, RoundTripTestBase):
 
207
    @classmethod
 
208
    def define_tables(cls, metadata):
 
209
        class MyString(String):
 
210
            def bind_expression(self, bindvalue):
 
211
                return func.lower(bindvalue)
 
212
 
 
213
            def column_expression(self, col):
 
214
                return func.upper(col)
 
215
 
 
216
        Table(
 
217
                'test_table',
 
218
                metadata,
 
219
                    Column('x', String(50)),
 
220
                    Column('y', MyString(50)
 
221
                )
 
222
        )
 
223
 
 
224
 
 
225
class TypeDecRoundTripTest(fixtures.TablesTest, RoundTripTestBase):
 
226
    @classmethod
 
227
    def define_tables(cls, metadata):
 
228
        class MyString(TypeDecorator):
 
229
            impl = String
 
230
            def bind_expression(self, bindvalue):
 
231
                return func.lower(bindvalue)
 
232
 
 
233
            def column_expression(self, col):
 
234
                return func.upper(col)
 
235
 
 
236
        Table(
 
237
                'test_table',
 
238
                metadata,
 
239
                    Column('x', String(50)),
 
240
                    Column('y', MyString(50)
 
241
                )
 
242
        )
 
243
 
 
244
class ReturningTest(fixtures.TablesTest):
 
245
    __requires__ = 'returning',
 
246
 
 
247
    @classmethod
 
248
    def define_tables(cls, metadata):
 
249
        class MyString(String):
 
250
            def column_expression(self, col):
 
251
                return func.lower(col)
 
252
 
 
253
        Table(
 
254
                'test_table',
 
255
                metadata, Column('x', String(50)),
 
256
                    Column('y', MyString(50), server_default="YVALUE")
 
257
        )
 
258
 
 
259
    @testing.provide_metadata
 
260
    def test_insert_returning(self):
 
261
        table = self.tables.test_table
 
262
        result = testing.db.execute(
 
263
                table.insert().returning(table.c.y),
 
264
                {"x": "xvalue"}
 
265
        )
 
266
        eq_(
 
267
            result.first(),
 
268
            ("yvalue",)
 
269
        )
 
270
 
 
271