← Back to branch summary

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

« back to all changes in this revision

Viewing changes to test/sql/test_returning.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_returning.py
1
 
from test.lib.testing import eq_
 
1
from sqlalchemy.testing import eq_
2
2
from sqlalchemy import *
3
 
from test.lib import *
4
 
from test.lib.schema import Table, Column
 
3
from sqlalchemy import testing
 
4
from sqlalchemy.testing.schema import Table, Column
5
5
from sqlalchemy.types import TypeDecorator
6
 
from test.lib import fixtures
 
6
from sqlalchemy.testing import fixtures, AssertsExecutionResults, engines, \
 
7
        assert_raises_message
 
8
from sqlalchemy import exc as sa_exc
7
9
 
8
10
class ReturningTest(fixtures.TestBase, AssertsExecutionResults):
9
11
    __requires__ = 'returning',
36
38
    def teardown(self):
37
39
        table.drop()
38
40
 
39
 
    @testing.exclude('firebird', '<', (2, 0), '2.0+ feature')
40
 
    @testing.exclude('postgresql', '<', (8, 2), '8.2+ feature')
41
41
    def test_column_targeting(self):
42
42
        result = table.insert().returning(table.c.id, table.c.full).execute({'persons': 1, 'full': False})
43
43
 
55
55
        eq_(row['goofy'], "FOOsomegoofyBAR")
56
56
 
57
57
    @testing.fails_on('firebird', "fb can't handle returning x AS y")
58
 
    @testing.exclude('firebird', '<', (2, 0), '2.0+ feature')
59
 
    @testing.exclude('postgresql', '<', (8, 2), '8.2+ feature')
60
58
    def test_labeling(self):
61
59
        result = table.insert().values(persons=6).\
62
60
                            returning(table.c.persons.label('lala')).execute()
65
63
 
66
64
    @testing.fails_on('firebird', "fb/kintersbasdb can't handle the bind params")
67
65
    @testing.fails_on('oracle+zxjdbc', "JDBC driver bug")
68
 
    @testing.exclude('firebird', '<', (2, 0), '2.0+ feature')
69
 
    @testing.exclude('postgresql', '<', (8, 2), '8.2+ feature')
70
66
    def test_anon_expressions(self):
71
67
        result = table.insert().values(goofy="someOTHERgoofy").\
72
68
                            returning(func.lower(table.c.goofy, type_=GoofyType)).execute()
73
69
        row = result.first()
74
 
        assert row[0] == "foosomeothergoofyBAR"
 
70
        eq_(row[0], "foosomeothergoofyBAR")
75
71
 
76
72
        result = table.insert().values(persons=12).\
77
73
                            returning(table.c.persons + 18).execute()
78
74
        row = result.first()
79
 
        assert row[0] == 30
 
75
        eq_(row[0], 30)
80
76
 
81
 
    @testing.exclude('firebird', '<', (2, 1), '2.1+ feature')
82
 
    @testing.exclude('postgresql', '<', (8, 2), '8.2+ feature')
83
77
    def test_update_returning(self):
84
78
        table.insert().execute([{'persons': 5, 'full': False}, {'persons': 3, 'full': False}])
85
79
 
87
81
        eq_(result.fetchall(), [(1,)])
88
82
 
89
83
        result2 = select([table.c.id, table.c.full]).order_by(table.c.id).execute()
90
 
        eq_(result2.fetchall(), [(1,True),(2,False)])
 
84
        eq_(result2.fetchall(), [(1, True), (2, False)])
91
85
 
92
 
    @testing.exclude('firebird', '<', (2, 0), '2.0+ feature')
93
 
    @testing.exclude('postgresql', '<', (8, 2), '8.2+ feature')
94
86
    def test_insert_returning(self):
95
87
        result = table.insert().returning(table.c.id).execute({'persons': 1, 'full': False})
96
88
 
97
89
        eq_(result.fetchall(), [(1,)])
98
90
 
99
 
        @testing.fails_on('postgresql', '')
100
 
        @testing.fails_on('oracle+cx_oracle', '')
101
 
        @testing.crashes('mssql+mxodbc', 'Raises an error')
102
 
        def test_executemany():
103
 
            # return value is documented as failing with psycopg2/executemany
104
 
            result2 = table.insert().returning(table).execute(
105
 
                 [{'persons': 2, 'full': False}, {'persons': 3, 'full': True}])
106
 
 
107
 
            if testing.against('mssql+zxjdbc'):
108
 
                # jtds apparently returns only the first row
109
 
                eq_(result2.fetchall(), [(2, 2, False, None)])
110
 
            elif testing.against('firebird', 'mssql', 'oracle'):
111
 
                # Multiple inserts only return the last row
112
 
                eq_(result2.fetchall(), [(3, 3, True, None)])
113
 
            else:
114
 
                # nobody does this as far as we know (pg8000?)
115
 
                eq_(result2.fetchall(), [(2, 2, False, None), (3, 3, True, None)])
116
 
 
117
 
        test_executemany()
118
 
 
119
 
 
120
 
 
121
 
    @testing.exclude('firebird', '<', (2, 1), '2.1+ feature')
122
 
    @testing.exclude('postgresql', '<', (8, 2), '8.2+ feature')
 
91
    @testing.requires.multivalues_inserts
 
92
    def test_multirow_returning(self):
 
93
        ins = table.insert().returning(table.c.id, table.c.persons).values(
 
94
                            [
 
95
                                {'persons': 1, 'full': False},
 
96
                                {'persons': 2, 'full': True},
 
97
                                {'persons': 3, 'full': False},
 
98
                            ]
 
99
                        )
 
100
        result = testing.db.execute(ins)
 
101
        eq_(
 
102
                result.fetchall(),
 
103
                 [(1, 1), (2, 2), (3, 3)]
 
104
        )
 
105
 
 
106
    def test_no_ipk_on_returning(self):
 
107
        result = testing.db.execute(
 
108
                    table.insert().returning(table.c.id),
 
109
                    {'persons': 1, 'full': False}
 
110
                )
 
111
        assert_raises_message(
 
112
            sa_exc.InvalidRequestError,
 
113
            "Can't call inserted_primary_key when returning\(\) is used.",
 
114
            getattr, result, "inserted_primary_key"
 
115
        )
 
116
 
123
117
    @testing.fails_on_everything_except('postgresql', 'firebird')
124
118
    def test_literal_returning(self):
125
119
        if testing.against("postgresql"):
131
125
                                        'values (5, 10, %s) returning persons' % literal_true)
132
126
        eq_([dict(row) for row in result4], [{'persons': 10}])
133
127
 
134
 
    @testing.exclude('firebird', '<', (2, 1), '2.1+ feature')
135
 
    @testing.exclude('postgresql', '<', (8, 2), '8.2+ feature')
136
128
    def test_delete_returning(self):
137
129
        table.insert().execute([{'persons': 5, 'full': False}, {'persons': 3, 'full': False}])
138
130
 
140
132
        eq_(result.fetchall(), [(1,)])
141
133
 
142
134
        result2 = select([table.c.id, table.c.full]).order_by(table.c.id).execute()
143
 
        eq_(result2.fetchall(), [(2,False),])
 
135
        eq_(result2.fetchall(), [(2, False),])
144
136
 
145
137
class SequenceReturningTest(fixtures.TestBase):
146
 
    __requires__ = 'returning',
 
138
    __requires__ = 'returning', 'sequences'
147
139
 
148
140
    def setup(self):
149
141
        meta = MetaData(testing.db)