← Back to branch summary

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

« back to all changes in this revision

Viewing changes to test/sql/test_labels.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_labels.py
1
 
from test.lib.testing import assert_raises, assert_raises_message, eq_
2
 
from test.lib.engines import testing_engine
3
 
from test.lib import fixtures, AssertsCompiledSQL, testing
4
 
from sqlalchemy import *
5
 
from sqlalchemy import exc as exceptions
 
1
from sqlalchemy import exc as exceptions, select, MetaData, Integer, or_
6
2
from sqlalchemy.engine import default
7
3
from sqlalchemy.sql import table, column
8
 
from test.lib.schema import Table, Column
 
4
from sqlalchemy.testing import AssertsCompiledSQL, assert_raises, engines,\
 
5
    fixtures
 
6
from sqlalchemy.testing.schema import Table, Column
9
7
 
10
8
IDENT_LENGTH = 29
11
9
 
12
 
class LabelTypeTest(fixtures.TestBase):
13
 
    def test_type(self):
14
 
        m = MetaData()
15
 
        t = Table('sometable', m,
16
 
            Column('col1', Integer),
17
 
            Column('col2', Float))
18
 
        assert isinstance(t.c.col1.label('hi').type, Integer)
19
 
        assert isinstance(select([t.c.col2]).as_scalar().label('lala').type,
20
 
                    Float)
21
 
 
22
 
class LongLabelsTest(fixtures.TablesTest, AssertsCompiledSQL):
23
 
    run_inserts = 'once'
24
 
    run_deletes = None
25
 
 
26
 
    @classmethod
27
 
    def define_tables(cls, metadata):
28
 
        table1 = Table("some_large_named_table", metadata,
29
 
            Column("this_is_the_primarykey_column", Integer,
30
 
                            primary_key=True,
31
 
                            test_needs_autoincrement=True),
32
 
            Column("this_is_the_data_column", String(30))
33
 
            )
34
 
 
35
 
        table2 = Table("table_with_exactly_29_characs", metadata,
36
 
            Column("this_is_the_primarykey_column", Integer,
37
 
                            primary_key=True,
38
 
                            test_needs_autoincrement=True),
39
 
            Column("this_is_the_data_column", String(30))
40
 
            )
41
 
        cls.tables.table1 = table1
42
 
        cls.tables.table2 = table2
43
 
 
44
 
    @classmethod
45
 
    def insert_data(cls):
46
 
        table1 = cls.tables.table1
47
 
        table2 = cls.tables.table2
48
 
        for data in [
49
 
            {"this_is_the_primarykey_column":1,
50
 
                        "this_is_the_data_column":"data1"},
51
 
            {"this_is_the_primarykey_column":2,
52
 
                        "this_is_the_data_column":"data2"},
53
 
            {"this_is_the_primarykey_column":3,
54
 
                        "this_is_the_data_column":"data3"},
55
 
            {"this_is_the_primarykey_column":4,
56
 
                        "this_is_the_data_column":"data4"}
57
 
        ]:
58
 
            testing.db.execute(
59
 
                table1.insert(),
60
 
                **data
61
 
            )
62
 
        testing.db.execute(
63
 
            table2.insert(),
64
 
            {"this_is_the_primary_key_column":1,
65
 
            "this_is_the_data_column":"data"}
66
 
        )
67
 
 
68
 
    @classmethod
69
 
    def setup_class(cls):
70
 
        super(LongLabelsTest, cls).setup_class()
71
 
        cls.maxlen = testing.db.dialect.max_identifier_length
72
 
        testing.db.dialect.max_identifier_length = IDENT_LENGTH
73
 
 
74
 
    @classmethod
75
 
    def teardown_class(cls):
76
 
        testing.db.dialect.max_identifier_length = cls.maxlen
77
 
        super(LongLabelsTest, cls).teardown_class()
 
10
 
 
11
class MaxIdentTest(fixtures.TestBase, AssertsCompiledSQL):
 
12
    __dialect__ = 'DefaultDialect'
 
13
 
 
14
    table1 = table('some_large_named_table',
 
15
            column('this_is_the_primarykey_column'),
 
16
            column('this_is_the_data_column')
 
17
        )
 
18
 
 
19
    table2 = table('table_with_exactly_29_characs',
 
20
            column('this_is_the_primarykey_column'),
 
21
            column('this_is_the_data_column')
 
22
        )
 
23
 
 
24
    def _length_fixture(self, length=IDENT_LENGTH, positional=False):
 
25
        dialect = default.DefaultDialect()
 
26
        dialect.max_identifier_length = length
 
27
        if positional:
 
28
            dialect.paramstyle = 'format'
 
29
            dialect.positional = True
 
30
        return dialect
 
31
 
 
32
    def _engine_fixture(self, length=IDENT_LENGTH):
 
33
        eng = engines.testing_engine()
 
34
        eng.dialect.max_identifier_length = length
 
35
        return eng
 
36
 
 
37
    def test_table_alias_1(self):
 
38
        self.assert_compile(
 
39
            self.table2.alias().select(),
 
40
            'SELECT '
 
41
                'table_with_exactly_29_c_1.'
 
42
                'this_is_the_primarykey_column, '
 
43
                'table_with_exactly_29_c_1.this_is_the_data_column '
 
44
            'FROM '
 
45
                'table_with_exactly_29_characs '
 
46
            'AS table_with_exactly_29_c_1',
 
47
            dialect=self._length_fixture()
 
48
        )
 
49
 
 
50
    def test_table_alias_2(self):
 
51
        table1 = self.table1
 
52
        table2 = self.table2
 
53
        ta = table2.alias()
 
54
        on = table1.c.this_is_the_data_column == ta.c.this_is_the_data_column
 
55
        self.assert_compile(
 
56
            select([table1, ta]).select_from(table1.join(ta, on)).
 
57
                where(ta.c.this_is_the_data_column == 'data3'),
 
58
            'SELECT '
 
59
                'some_large_named_table.this_is_the_primarykey_column, '
 
60
                'some_large_named_table.this_is_the_data_column, '
 
61
                'table_with_exactly_29_c_1.this_is_the_primarykey_column, '
 
62
                'table_with_exactly_29_c_1.this_is_the_data_column '
 
63
            'FROM '
 
64
                'some_large_named_table '
 
65
            'JOIN '
 
66
                'table_with_exactly_29_characs '
 
67
            'AS '
 
68
                'table_with_exactly_29_c_1 '
 
69
            'ON '
 
70
                'some_large_named_table.this_is_the_data_column = '
 
71
                    'table_with_exactly_29_c_1.this_is_the_data_column '
 
72
            'WHERE '
 
73
                'table_with_exactly_29_c_1.this_is_the_data_column = '
 
74
                    ':this_is_the_data_column_1',
 
75
            dialect=self._length_fixture()
 
76
        )
78
77
 
79
78
    def test_too_long_name_disallowed(self):
80
 
        m = MetaData(testing.db)
81
 
        t1 = Table("this_name_is_too_long_for_what_were_doing_in_this_test",
 
79
        m = MetaData()
 
80
        t = Table('this_name_is_too_long_for_what_were_doing_in_this_test',
82
81
                        m, Column('foo', Integer))
83
 
        assert_raises(exceptions.IdentifierError, m.create_all)
84
 
        assert_raises(exceptions.IdentifierError, m.drop_all)
85
 
        assert_raises(exceptions.IdentifierError, t1.create)
86
 
        assert_raises(exceptions.IdentifierError, t1.drop)
87
 
 
88
 
    def test_basic_result(self):
89
 
        table1 = self.tables.table1
90
 
        s = table1.select(use_labels=True,
91
 
                        order_by=[table1.c.this_is_the_primarykey_column])
92
 
 
93
 
        result = [
94
 
            (row[table1.c.this_is_the_primarykey_column],
95
 
            row[table1.c.this_is_the_data_column])
96
 
            for row in testing.db.execute(s)
97
 
        ]
98
 
        eq_(result, [
99
 
            (1, "data1"),
100
 
            (2, "data2"),
101
 
            (3, "data3"),
102
 
            (4, "data4"),
103
 
        ])
104
 
 
105
 
    def test_result_limit(self):
106
 
        table1 = self.tables.table1
107
 
        # some dialects such as oracle (and possibly ms-sql
108
 
        # in a future version)
109
 
        # generate a subquery for limits/offsets.
110
 
        # ensure that the generated result map corresponds
111
 
        # to the selected table, not
112
 
        # the select query
 
82
        eng = self._engine_fixture()
 
83
        methods = (t.create, t.drop, m.create_all, m.drop_all)
 
84
        for meth in methods:
 
85
            assert_raises(exceptions.IdentifierError, meth, eng)
 
86
 
 
87
    def _assert_labeled_table1_select(self, s):
 
88
        table1 = self.table1
 
89
        compiled = s.compile(dialect=self._length_fixture())
 
90
 
 
91
        assert set(compiled.result_map['some_large_named_table__2'][1]).\
 
92
                issuperset(
 
93
                    [
 
94
                        'some_large_named_table_this_is_the_data_column',
 
95
                        'some_large_named_table__2',
 
96
                        table1.c.this_is_the_data_column
 
97
                    ]
 
98
                )
 
99
 
 
100
        assert set(compiled.result_map['some_large_named_table__1'][1]).\
 
101
                issuperset(
 
102
                    [
 
103
                        'some_large_named_table_this_is_the_primarykey_column',
 
104
                        'some_large_named_table__1',
 
105
                        table1.c.this_is_the_primarykey_column
 
106
                    ]
 
107
                )
 
108
 
 
109
    def test_result_map_use_labels(self):
 
110
        table1 = self.table1
 
111
        s = table1.select().apply_labels().\
 
112
                order_by(table1.c.this_is_the_primarykey_column)
 
113
 
 
114
        self._assert_labeled_table1_select(s)
 
115
 
 
116
    def test_result_map_limit(self):
 
117
        table1 = self.table1
 
118
        # some dialects such as oracle (and possibly ms-sql in a future
 
119
        # version) generate a subquery for limits/offsets. ensure that the
 
120
        # generated result map corresponds to the selected table, not the
 
121
        # select query
113
122
        s = table1.select(use_labels=True,
114
123
                        order_by=[table1.c.this_is_the_primarykey_column]).\
115
124
                        limit(2)
116
 
 
117
 
        result = [
118
 
            (row[table1.c.this_is_the_primarykey_column],
119
 
            row[table1.c.this_is_the_data_column])
120
 
            for row in testing.db.execute(s)
121
 
        ]
122
 
        eq_(result, [
123
 
            (1, "data1"),
124
 
            (2, "data2"),
125
 
        ])
126
 
 
127
 
    @testing.requires.offset
128
 
    def test_result_limit_offset(self):
129
 
        table1 = self.tables.table1
130
 
        s = table1.select(use_labels=True,
131
 
                        order_by=[table1.c.this_is_the_primarykey_column]).\
132
 
                        limit(2).offset(1)
133
 
 
134
 
        result = [
135
 
            (row[table1.c.this_is_the_primarykey_column],
136
 
            row[table1.c.this_is_the_data_column])
137
 
            for row in testing.db.execute(s)
138
 
        ]
139
 
        eq_(result, [
140
 
            (2, "data2"),
141
 
            (3, "data3"),
142
 
        ])
143
 
 
144
 
    def test_table_alias_1(self):
145
 
        table2 = self.tables.table2
146
 
        if testing.against('oracle'):
147
 
            self.assert_compile(
148
 
                table2.alias().select(),
149
 
                "SELECT table_with_exactly_29_c_1."
150
 
                "this_is_the_primarykey_column, "
151
 
                "table_with_exactly_29_c_1.this_is_the_data_column "
152
 
                "FROM table_with_exactly_29_characs "
153
 
                "table_with_exactly_29_c_1"
154
 
            )
155
 
        else:
156
 
            self.assert_compile(
157
 
                table2.alias().select(),
158
 
                "SELECT table_with_exactly_29_c_1."
159
 
                "this_is_the_primarykey_column, "
160
 
                "table_with_exactly_29_c_1.this_is_the_data_column "
161
 
                "FROM table_with_exactly_29_characs AS "
162
 
                "table_with_exactly_29_c_1"
163
 
            )
164
 
 
165
 
    def test_table_alias_2(self):
166
 
        table1 = self.tables.table1
167
 
        table2 = self.tables.table2
168
 
        ta = table2.alias()
169
 
        dialect = default.DefaultDialect()
170
 
        dialect.max_identifier_length = IDENT_LENGTH
171
 
        self.assert_compile(
172
 
            select([table1, ta]).select_from(
173
 
                        table1.join(ta,
174
 
                            table1.c.this_is_the_data_column==
175
 
                            ta.c.this_is_the_data_column)).\
176
 
                        where(ta.c.this_is_the_data_column=='data3'),
177
 
 
178
 
            "SELECT some_large_named_table.this_is_the_primarykey_column, "
179
 
            "some_large_named_table.this_is_the_data_column, "
180
 
            "table_with_exactly_29_c_1.this_is_the_primarykey_column, "
181
 
            "table_with_exactly_29_c_1.this_is_the_data_column FROM "
182
 
            "some_large_named_table JOIN table_with_exactly_29_characs "
183
 
            "AS table_with_exactly_29_c_1 ON "
184
 
            "some_large_named_table.this_is_the_data_column = "
185
 
            "table_with_exactly_29_c_1.this_is_the_data_column "
186
 
            "WHERE table_with_exactly_29_c_1.this_is_the_data_column = "
187
 
            ":this_is_the_data_column_1",
188
 
            dialect=dialect
189
 
        )
190
 
 
191
 
    def test_table_alias_3(self):
192
 
        table2 = self.tables.table2
193
 
        eq_(
194
 
            testing.db.execute(table2.alias().select()).first(),
195
 
            (1, "data")
196
 
        )
197
 
 
198
 
    def test_colbinds(self):
199
 
        table1 = self.tables.table1
200
 
        r = table1.select(table1.c.this_is_the_primarykey_column == 4).\
201
 
                    execute()
202
 
        assert r.fetchall() == [(4, "data4")]
203
 
 
204
 
        r = table1.select(or_(
 
125
        self._assert_labeled_table1_select(s)
 
126
 
 
127
    def test_result_map_subquery(self):
 
128
        table1 = self.table1
 
129
        s = table1.select(
 
130
                    table1.c.this_is_the_primarykey_column == 4).\
 
131
                    alias('foo')
 
132
        s2 = select([s])
 
133
        compiled = s2.compile(dialect=self._length_fixture())
 
134
        assert \
 
135
            set(compiled.result_map['this_is_the_data_column'][1]).\
 
136
            issuperset(['this_is_the_data_column',
 
137
                    s.c.this_is_the_data_column])
 
138
        assert \
 
139
            set(compiled.result_map['this_is_the_primarykey_column'][1]).\
 
140
            issuperset(['this_is_the_primarykey_column',
 
141
                    s.c.this_is_the_primarykey_column])
 
142
 
 
143
    def test_result_map_anon_alias(self):
 
144
        table1 = self.table1
 
145
        dialect = self._length_fixture()
 
146
 
 
147
        q = table1.select(table1.c.this_is_the_primarykey_column == 4).alias()
 
148
        s = select([q]).apply_labels()
 
149
 
 
150
        self.assert_compile(s,
 
151
            'SELECT '
 
152
                'anon_1.this_is_the_primarykey_column '
 
153
                    'AS anon_1_this_is_the_prim_1, '
 
154
                'anon_1.this_is_the_data_column '
 
155
                    'AS anon_1_this_is_the_data_2 '
 
156
            'FROM ('
 
157
                'SELECT '
 
158
                    'some_large_named_table.'
 
159
                    'this_is_the_primarykey_column '
 
160
                        'AS this_is_the_primarykey_column, '
 
161
                    'some_large_named_table.this_is_the_data_column '
 
162
                        'AS this_is_the_data_column '
 
163
                'FROM '
 
164
                    'some_large_named_table '
 
165
                'WHERE '
 
166
                    'some_large_named_table.this_is_the_primarykey_column '
 
167
                        '= :this_is_the_primarykey__1'
 
168
                ') '
 
169
            'AS anon_1',
 
170
            dialect=dialect)
 
171
        compiled = s.compile(dialect=dialect)
 
172
        assert set(compiled.result_map['anon_1_this_is_the_data_2'][1]).\
 
173
                issuperset([
 
174
                        'anon_1_this_is_the_data_2',
 
175
                        q.corresponding_column(
 
176
                                table1.c.this_is_the_data_column)
 
177
                    ])
 
178
 
 
179
        assert set(compiled.result_map['anon_1_this_is_the_prim_1'][1]).\
 
180
                issuperset([
 
181
                        'anon_1_this_is_the_prim_1',
 
182
                        q.corresponding_column(
 
183
                                table1.c.this_is_the_primarykey_column)
 
184
                    ])
 
185
 
 
186
    def test_column_bind_labels_1(self):
 
187
        table1 = self.table1
 
188
 
 
189
        s = table1.select(table1.c.this_is_the_primarykey_column == 4)
 
190
        self.assert_compile(
 
191
            s,
 
192
            "SELECT some_large_named_table.this_is_the_primarykey_column, "
 
193
            "some_large_named_table.this_is_the_data_column "
 
194
            "FROM some_large_named_table WHERE "
 
195
            "some_large_named_table.this_is_the_primarykey_column = "
 
196
            ":this_is_the_primarykey__1",
 
197
            checkparams={'this_is_the_primarykey__1': 4},
 
198
            dialect=self._length_fixture()
 
199
        )
 
200
 
 
201
        self.assert_compile(
 
202
            s,
 
203
            "SELECT some_large_named_table.this_is_the_primarykey_column, "
 
204
            "some_large_named_table.this_is_the_data_column "
 
205
            "FROM some_large_named_table WHERE "
 
206
            "some_large_named_table.this_is_the_primarykey_column = "
 
207
            "%s",
 
208
            checkpositional=(4, ),
 
209
            checkparams={'this_is_the_primarykey__1': 4},
 
210
            dialect=self._length_fixture(positional=True)
 
211
        )
 
212
 
 
213
    def test_column_bind_labels_2(self):
 
214
        table1 = self.table1
 
215
 
 
216
        s = table1.select(or_(
205
217
            table1.c.this_is_the_primarykey_column == 4,
206
218
            table1.c.this_is_the_primarykey_column == 2
207
 
        )).execute()
208
 
        assert r.fetchall() == [(2, "data2"), (4, "data4")]
209
 
 
210
 
    @testing.provide_metadata
211
 
    def test_insert_no_pk(self):
212
 
        t = Table("some_other_large_named_table", self.metadata,
213
 
            Column("this_is_the_primarykey_column", Integer,
214
 
                            Sequence("this_is_some_large_seq"),
215
 
                            primary_key=True),
216
 
            Column("this_is_the_data_column", String(30))
217
 
            )
218
 
        t.create(testing.db, checkfirst=True)
219
 
        testing.db.execute(t.insert(),
220
 
                **{"this_is_the_data_column":"data1"})
221
 
 
222
 
    @testing.requires.subqueries
223
 
    def test_subquery(self):
224
 
        table1 = self.tables.table1
225
 
        q = table1.select(table1.c.this_is_the_primarykey_column == 4).\
226
 
                        alias('foo')
227
 
        eq_(
228
 
            list(testing.db.execute(select([q]))),
229
 
            [(4, u'data4')]
230
 
        )
231
 
 
232
 
    @testing.requires.subqueries
233
 
    def test_anon_alias(self):
234
 
        table1 = self.tables.table1
235
 
        compile_dialect = default.DefaultDialect()
236
 
        compile_dialect.max_identifier_length = IDENT_LENGTH
237
 
 
238
 
        q = table1.select(table1.c.this_is_the_primarykey_column == 4).alias()
239
 
        x = select([q], use_labels=True)
240
 
 
241
 
        self.assert_compile(x,
242
 
            "SELECT anon_1.this_is_the_primarykey_column AS "
243
 
            "anon_1_this_is_the_prim_1, anon_1.this_is_the_data_column "
244
 
            "AS anon_1_this_is_the_data_2 "
245
 
            "FROM (SELECT some_large_named_table."
246
 
            "this_is_the_primarykey_column AS "
247
 
            "this_is_the_primarykey_column, "
248
 
            "some_large_named_table.this_is_the_data_column "
249
 
            "AS this_is_the_data_column "
250
 
            "FROM some_large_named_table "
251
 
            "WHERE some_large_named_table.this_is_the_primarykey_column "
252
 
            "= :this_is_the_primarykey__1) AS anon_1",
253
 
            dialect=compile_dialect)
254
 
 
255
 
        eq_(
256
 
            list(testing.db.execute(x)),
257
 
            [(4, u'data4')]
258
 
        )
259
 
 
260
 
    def test_adjustable(self):
261
 
        table1 = self.tables.table1
262
 
 
263
 
        q = table1.select(table1.c.this_is_the_primarykey_column == 4).alias('foo')
264
 
        x = select([q])
265
 
 
266
 
        compile_dialect = default.DefaultDialect(label_length=10)
267
 
        self.assert_compile(x,
268
 
            "SELECT foo.this_1, foo.this_2 FROM "
269
 
            "(SELECT some_large_named_table."
270
 
            "this_is_the_primarykey_column AS this_1, "
271
 
            "some_large_named_table.this_is_the_data_column AS this_2 "
272
 
            "FROM some_large_named_table WHERE "
273
 
            "some_large_named_table.this_is_the_primarykey_column = :this_1) AS foo",
274
 
            dialect=compile_dialect)
 
219
        ))
 
220
        self.assert_compile(
 
221
            s,
 
222
            "SELECT some_large_named_table.this_is_the_primarykey_column, "
 
223
            "some_large_named_table.this_is_the_data_column "
 
224
            "FROM some_large_named_table WHERE "
 
225
            "some_large_named_table.this_is_the_primarykey_column = "
 
226
            ":this_is_the_primarykey__1 OR "
 
227
            "some_large_named_table.this_is_the_primarykey_column = "
 
228
            ":this_is_the_primarykey__2",
 
229
            checkparams={
 
230
                'this_is_the_primarykey__1': 4,
 
231
                'this_is_the_primarykey__2': 2
 
232
            },
 
233
            dialect=self._length_fixture()
 
234
        )
 
235
        self.assert_compile(
 
236
            s,
 
237
            "SELECT some_large_named_table.this_is_the_primarykey_column, "
 
238
            "some_large_named_table.this_is_the_data_column "
 
239
            "FROM some_large_named_table WHERE "
 
240
            "some_large_named_table.this_is_the_primarykey_column = "
 
241
            "%s OR "
 
242
            "some_large_named_table.this_is_the_primarykey_column = "
 
243
            "%s",
 
244
            checkparams={
 
245
                'this_is_the_primarykey__1': 4,
 
246
                'this_is_the_primarykey__2': 2
 
247
            },
 
248
            checkpositional=(4, 2),
 
249
            dialect=self._length_fixture(positional=True)
 
250
        )
 
251
 
 
252
 
 
253
class LabelLengthTest(fixtures.TestBase, AssertsCompiledSQL):
 
254
    __dialect__ = 'DefaultDialect'
 
255
 
 
256
    table1 = table('some_large_named_table',
 
257
            column('this_is_the_primarykey_column'),
 
258
            column('this_is_the_data_column')
 
259
        )
 
260
 
 
261
    table2 = table('table_with_exactly_29_characs',
 
262
            column('this_is_the_primarykey_column'),
 
263
            column('this_is_the_data_column')
 
264
        )
 
265
 
 
266
    def test_adjustable_1(self):
 
267
        table1 = self.table1
 
268
        q = table1.select(
 
269
            table1.c.this_is_the_primarykey_column == 4).alias('foo')
 
270
        x = select([q])
 
271
        compile_dialect = default.DefaultDialect(label_length=10)
 
272
        self.assert_compile(x,
 
273
            'SELECT '
 
274
                'foo.this_1, foo.this_2 '
 
275
            'FROM ('
 
276
                'SELECT '
 
277
                    'some_large_named_table.this_is_the_primarykey_column '
 
278
                        'AS this_1, '
 
279
                    'some_large_named_table.this_is_the_data_column '
 
280
                        'AS this_2 '
 
281
                'FROM '
 
282
                    'some_large_named_table '
 
283
                'WHERE '
 
284
                    'some_large_named_table.this_is_the_primarykey_column '
 
285
                        '= :this_1'
 
286
                ') '
 
287
            'AS foo',
 
288
            dialect=compile_dialect)
 
289
 
 
290
    def test_adjustable_2(self):
 
291
        table1 = self.table1
 
292
 
 
293
        q = table1.select(
 
294
            table1.c.this_is_the_primarykey_column == 4).alias('foo')
 
295
        x = select([q])
 
296
 
 
297
        compile_dialect = default.DefaultDialect(label_length=10)
 
298
        self.assert_compile(x,
 
299
            'SELECT '
 
300
                'foo.this_1, foo.this_2 '
 
301
            'FROM ('
 
302
                'SELECT '
 
303
                    'some_large_named_table.this_is_the_primarykey_column '
 
304
                        'AS this_1, '
 
305
                    'some_large_named_table.this_is_the_data_column '
 
306
                        'AS this_2 '
 
307
                'FROM '
 
308
                    'some_large_named_table '
 
309
                'WHERE '
 
310
                    'some_large_named_table.this_is_the_primarykey_column '
 
311
                        '= :this_1'
 
312
                ') '
 
313
            'AS foo',
 
314
            dialect=compile_dialect)
 
315
 
 
316
    def test_adjustable_3(self):
 
317
        table1 = self.table1
275
318
 
276
319
        compile_dialect = default.DefaultDialect(label_length=4)
277
 
        self.assert_compile(x, "SELECT foo._1, foo._2 FROM "
278
 
            "(SELECT some_large_named_table.this_is_the_primarykey_column "
279
 
            "AS _1, some_large_named_table.this_is_the_data_column AS _2 "
280
 
            "FROM some_large_named_table WHERE "
281
 
            "some_large_named_table.this_is_the_primarykey_column = :_1) AS foo",
 
320
        q = table1.select(
 
321
            table1.c.this_is_the_primarykey_column == 4).alias('foo')
 
322
        x = select([q])
 
323
 
 
324
        self.assert_compile(x,
 
325
            'SELECT '
 
326
                'foo._1, foo._2 '
 
327
            'FROM ('
 
328
                'SELECT '
 
329
                    'some_large_named_table.this_is_the_primarykey_column '
 
330
                        'AS _1, '
 
331
                    'some_large_named_table.this_is_the_data_column '
 
332
                        'AS _2 '
 
333
                'FROM '
 
334
                    'some_large_named_table '
 
335
                'WHERE '
 
336
                    'some_large_named_table.this_is_the_primarykey_column '
 
337
                        '= :_1'
 
338
                ') '
 
339
            'AS foo',
282
340
        dialect=compile_dialect)
283
341
 
 
342
    def test_adjustable_4(self):
 
343
        table1 = self.table1
 
344
 
284
345
        q = table1.select(table1.c.this_is_the_primarykey_column == 4).alias()
285
346
        x = select([q], use_labels=True)
286
347
 
287
348
        compile_dialect = default.DefaultDialect(label_length=10)
288
349
        self.assert_compile(x,
289
 
            "SELECT anon_1.this_2 AS anon_1, anon_1.this_4 AS anon_3 FROM "
290
 
            "(SELECT some_large_named_table.this_is_the_primarykey_column "
291
 
            "AS this_2, some_large_named_table.this_is_the_data_column AS this_4 "
292
 
            "FROM some_large_named_table WHERE "
293
 
            "some_large_named_table.this_is_the_primarykey_column = :this_1) AS anon_1",
 
350
            'SELECT '
 
351
                'anon_1.this_2 AS anon_1, '
 
352
                    'anon_1.this_4 AS anon_3 '
 
353
            'FROM ('
 
354
                'SELECT '
 
355
                    'some_large_named_table.this_is_the_primarykey_column '
 
356
                        'AS this_2, '
 
357
                    'some_large_named_table.this_is_the_data_column '
 
358
                        'AS this_4 '
 
359
                'FROM '
 
360
                    'some_large_named_table '
 
361
                'WHERE '
 
362
                    'some_large_named_table.this_is_the_primarykey_column '
 
363
                        '= :this_1'
 
364
                ') '
 
365
            'AS anon_1',
294
366
            dialect=compile_dialect)
295
367
 
 
368
    def test_adjustable_5(self):
 
369
        table1 = self.table1
 
370
        q = table1.select(table1.c.this_is_the_primarykey_column == 4).alias()
 
371
        x = select([q], use_labels=True)
 
372
 
296
373
        compile_dialect = default.DefaultDialect(label_length=4)
297
 
        self.assert_compile(x, "SELECT _1._2 AS _1, _1._4 AS _3 FROM "
298
 
            "(SELECT some_large_named_table.this_is_the_primarykey_column "
299
 
            "AS _2, some_large_named_table.this_is_the_data_column AS _4 "
300
 
            "FROM some_large_named_table WHERE "
301
 
            "some_large_named_table.this_is_the_primarykey_column = :_1) AS _1",
 
374
        self.assert_compile(x,
 
375
            'SELECT '
 
376
                '_1._2 AS _1, '
 
377
                '_1._4 AS _3 '
 
378
            'FROM ('
 
379
                'SELECT '
 
380
                    'some_large_named_table.this_is_the_primarykey_column '
 
381
                        'AS _2, '
 
382
                    'some_large_named_table.this_is_the_data_column '
 
383
                        'AS _4 '
 
384
                'FROM '
 
385
                    'some_large_named_table '
 
386
                'WHERE '
 
387
                    'some_large_named_table.this_is_the_primarykey_column '
 
388
                        '= :_1'
 
389
                ') '
 
390
            'AS _1',
302
391
            dialect=compile_dialect)
303
392
 
304
 
    def test_adjustable_result_schema_column(self):
305
 
        table1 = self.tables.table1
306
 
 
307
 
        q = table1.select(table1.c.this_is_the_primarykey_column == 4).alias('foo')
308
 
        x = select([q])
309
 
 
310
 
        e = testing_engine(options={"label_length":10})
311
 
        e.pool = testing.db.pool
312
 
        row = e.execute(x).first()
313
 
        eq_(row.this_is_the_primarykey_column, 4)
314
 
        eq_(row.this_1, 4)
315
 
        eq_(row['this_1'], 4)
316
 
 
317
 
        q = table1.select(table1.c.this_is_the_primarykey_column == 4).alias('foo')
318
 
        row = e.execute(x).first()
319
 
        eq_(row.this_is_the_primarykey_column, 4)
320
 
        eq_(row.this_1, 4)
321
 
 
322
 
    def test_adjustable_result_lightweight_column(self):
323
 
 
324
 
        table1 = table("some_large_named_table",
325
 
            column("this_is_the_primarykey_column"),
326
 
            column("this_is_the_data_column")
327
 
        )
328
 
 
329
 
        q = table1.select(table1.c.this_is_the_primarykey_column == 4).alias('foo')
330
 
        x = select([q])
331
 
 
332
 
        e = testing_engine(options={"label_length":10})
333
 
        e.pool = testing.db.pool
334
 
        row = e.execute(x).first()
335
 
        eq_(row.this_is_the_primarykey_column, 4)
336
 
        eq_(row.this_1, 4)
 
393
    def test_adjustable_result_schema_column_1(self):
 
394
        table1 = self.table1
 
395
 
 
396
        q = table1.select(
 
397
            table1.c.this_is_the_primarykey_column == 4).apply_labels().\
 
398
                alias('foo')
 
399
 
 
400
        dialect = default.DefaultDialect(label_length=10)
 
401
        compiled = q.compile(dialect=dialect)
 
402
 
 
403
        assert set(compiled.result_map['some_2'][1]).issuperset([
 
404
            table1.c.this_is_the_data_column,
 
405
            'some_large_named_table_this_is_the_data_column',
 
406
            'some_2'
 
407
        ])
 
408
 
 
409
        assert set(compiled.result_map['some_1'][1]).issuperset([
 
410
            table1.c.this_is_the_primarykey_column,
 
411
            'some_large_named_table_this_is_the_primarykey_column',
 
412
            'some_1'
 
413
        ])
 
414
 
 
415
    def test_adjustable_result_schema_column_2(self):
 
416
        table1 = self.table1
 
417
 
 
418
        q = table1.select(
 
419
            table1.c.this_is_the_primarykey_column == 4).alias('foo')
 
420
        x = select([q])
 
421
 
 
422
        dialect = default.DefaultDialect(label_length=10)
 
423
        compiled = x.compile(dialect=dialect)
 
424
 
 
425
        assert set(compiled.result_map['this_2'][1]).issuperset([
 
426
            q.corresponding_column(table1.c.this_is_the_data_column),
 
427
            'this_is_the_data_column',
 
428
            'this_2'])
 
429
 
 
430
        assert set(compiled.result_map['this_1'][1]).issuperset([
 
431
            q.corresponding_column(table1.c.this_is_the_primarykey_column),
 
432
            'this_is_the_primarykey_column',
 
433
            'this_1'])
337
434
 
338
435
    def test_table_plus_column_exceeds_length(self):
339
 
        """test that the truncation occurs if tablename / colname are only
340
 
        greater than the max when concatenated."""
 
436
        """test that the truncation only occurs when tablename + colname are
 
437
        concatenated, if they are individually under the label length.
 
438
 
 
439
        """
341
440
 
342
441
        compile_dialect = default.DefaultDialect(label_length=30)
343
 
        m = MetaData()
344
 
        a_table = Table(
 
442
        a_table = table(
345
443
            'thirty_characters_table_xxxxxx',
346
 
            m,
347
 
            Column('id', Integer, primary_key=True)
 
444
            column('id')
348
445
        )
349
446
 
350
 
        other_table = Table(
 
447
        other_table = table(
351
448
            'other_thirty_characters_table_',
352
 
            m,
353
 
            Column('id', Integer, primary_key=True),
354
 
            Column('thirty_characters_table_id',
355
 
                Integer,
356
 
                ForeignKey('thirty_characters_table_xxxxxx.id'),
357
 
                primary_key=True
358
 
            )
 
449
            column('id'),
 
450
            column('thirty_characters_table_id')
359
451
        )
360
452
 
361
453
        anon = a_table.alias()
 
454
 
 
455
        j1 = other_table.outerjoin(anon,
 
456
                anon.c.id == other_table.c.thirty_characters_table_id)
 
457
 
362
458
        self.assert_compile(
363
 
            select([other_table,anon]).
364
 
                            select_from(
365
 
                                other_table.outerjoin(anon)
366
 
                        ).apply_labels(),
367
 
            "SELECT other_thirty_characters_table_.id AS "
368
 
            "other_thirty_characters__1, "
369
 
            "other_thirty_characters_table_.thirty_characters_table_id "
370
 
            "AS other_thirty_characters__2, thirty_characters_table__1.id "
371
 
            "AS thirty_characters_table__3 "
372
 
            "FROM other_thirty_characters_table_ "
373
 
            "LEFT OUTER JOIN thirty_characters_table_xxxxxx "
374
 
            "AS thirty_characters_table__1 ON "
375
 
            "thirty_characters_table__1.id = "
376
 
            "other_thirty_characters_table_.thirty_characters_table_id",
 
459
            select([other_table, anon]).
 
460
                    select_from(j1).apply_labels(),
 
461
            'SELECT '
 
462
                'other_thirty_characters_table_.id '
 
463
                    'AS other_thirty_characters__1, '
 
464
                'other_thirty_characters_table_.thirty_characters_table_id '
 
465
                    'AS other_thirty_characters__2, '
 
466
                'thirty_characters_table__1.id '
 
467
                    'AS thirty_characters_table__3 '
 
468
            'FROM '
 
469
                'other_thirty_characters_table_ '
 
470
            'LEFT OUTER JOIN '
 
471
                'thirty_characters_table_xxxxxx AS thirty_characters_table__1 '
 
472
            'ON thirty_characters_table__1.id = '
 
473
                'other_thirty_characters_table_.thirty_characters_table_id',
377
474
            dialect=compile_dialect)
378
475
 
379
 
        self.assert_compile(
380
 
                select([other_table, anon]).
381
 
                    select_from(
382
 
                                other_table.outerjoin(anon)
383
 
                    ).apply_labels(),
384
 
            "SELECT other_thirty_characters_table_.id AS "
385
 
            "other_thirty_characters__1, "
386
 
            "other_thirty_characters_table_.thirty_characters_table_id "
387
 
            "AS other_thirty_characters__2, "
388
 
            "thirty_characters_table__1.id AS thirty_characters_table__3 "
389
 
            "FROM other_thirty_characters_table_ "
390
 
            "LEFT OUTER JOIN thirty_characters_table_xxxxxx "
391
 
            "AS thirty_characters_table__1 ON "
392
 
            "thirty_characters_table__1.id = "
393
 
            "other_thirty_characters_table_.thirty_characters_table_id",
394
 
            dialect=compile_dialect
395
 
        )
 
476
    def test_colnames_longer_than_labels_lowercase(self):
 
477
        t1 = table('a', column('abcde'))
 
478
        self._test_colnames_longer_than_labels(t1)
 
479
 
 
480
    def test_colnames_longer_than_labels_uppercase(self):
 
481
        m = MetaData()
 
482
        t1 = Table('a', m, Column('abcde', Integer))
 
483
        self._test_colnames_longer_than_labels(t1)
 
484
 
 
485
    def _test_colnames_longer_than_labels(self, t1):
 
486
        dialect = default.DefaultDialect(label_length=4)
 
487
        a1 = t1.alias(name='asdf')
 
488
 
 
489
        # 'abcde' is longer than 4, but rendered as itself
 
490
        # needs to have all characters
 
491
        s = select([a1])
 
492
        self.assert_compile(select([a1]),
 
493
            'SELECT asdf.abcde FROM a AS asdf',
 
494
            dialect=dialect)
 
495
        compiled = s.compile(dialect=dialect)
 
496
        assert set(compiled.result_map['abcde'][1]).issuperset([
 
497
            'abcde', a1.c.abcde, 'abcde'])
 
498
 
 
499
        # column still there, but short label
 
500
        s = select([a1]).apply_labels()
 
501
        self.assert_compile(s,
 
502
            'SELECT asdf.abcde AS _1 FROM a AS asdf',
 
503
            dialect=dialect)
 
504
        compiled = s.compile(dialect=dialect)
 
505
        assert set(compiled.result_map['_1'][1]).issuperset([
 
506
            'asdf_abcde', a1.c.abcde, '_1'])