1
from .. import fixtures, config
2
from ..config import requirements
3
from .. import exclusions
4
from ..assertions import eq_
7
from sqlalchemy import Integer, String, select, util
9
from ..schema import Table, Column
12
class LastrowidTest(fixtures.TablesTest):
15
__requires__ = 'implements_get_lastrowid', 'autoincrement_insert'
17
__engine_options__ = {"implicit_returning": False}
20
def define_tables(cls, metadata):
21
Table('autoinc_pk', metadata,
22
Column('id', Integer, primary_key=True,
23
test_needs_autoincrement=True),
24
Column('data', String(50))
27
Table('manual_pk', metadata,
28
Column('id', Integer, primary_key=True, autoincrement=False),
29
Column('data', String(50))
32
def _assert_round_trip(self, table, conn):
33
row = conn.execute(table.select()).first()
36
(config.db.dialect.default_sequence_base, "some data")
39
def test_autoincrement_on_insert(self):
42
self.tables.autoinc_pk.insert(),
45
self._assert_round_trip(self.tables.autoinc_pk, config.db)
47
def test_last_inserted_id(self):
49
r = config.db.execute(
50
self.tables.autoinc_pk.insert(),
53
pk = config.db.scalar(select([self.tables.autoinc_pk.c.id]))
55
r.inserted_primary_key,
59
@exclusions.fails_if(lambda: util.pypy, "lastrowid not maintained after "
61
@requirements.dbapi_lastrowid
62
def test_native_lastrowid_autoinc(self):
63
r = config.db.execute(
64
self.tables.autoinc_pk.insert(),
67
lastrowid = r.lastrowid
68
pk = config.db.scalar(select([self.tables.autoinc_pk.c.id]))
74
class InsertBehaviorTest(fixtures.TablesTest):
78
def define_tables(cls, metadata):
79
Table('autoinc_pk', metadata,
80
Column('id', Integer, primary_key=True, \
81
test_needs_autoincrement=True),
82
Column('data', String(50))
85
def test_autoclose_on_insert(self):
86
if requirements.returning.enabled:
87
engine = engines.testing_engine(
88
options={'implicit_returning': False})
93
self.tables.autoinc_pk.insert(),
98
assert not r.returns_rows
100
@requirements.returning
101
def test_autoclose_on_insert_implicit_returning(self):
102
r = config.db.execute(
103
self.tables.autoinc_pk.insert(),
108
assert not r.returns_rows
110
@requirements.empty_inserts
111
def test_empty_insert(self):
112
r = config.db.execute(
113
self.tables.autoinc_pk.insert(),
117
r = config.db.execute(
118
self.tables.autoinc_pk.select().\
119
where(self.tables.autoinc_pk.c.id != None)
122
assert len(r.fetchall())
125
class ReturningTest(fixtures.TablesTest):
127
__requires__ = 'returning', 'autoincrement_insert'
129
__engine_options__ = {"implicit_returning": True}
131
def _assert_round_trip(self, table, conn):
132
row = conn.execute(table.select()).first()
135
(config.db.dialect.default_sequence_base, "some data")
139
def define_tables(cls, metadata):
140
Table('autoinc_pk', metadata,
141
Column('id', Integer, primary_key=True, \
142
test_needs_autoincrement=True),
143
Column('data', String(50))
146
def test_explicit_returning_pk(self):
148
table = self.tables.autoinc_pk
150
table.insert().returning(
155
fetched_pk = config.db.scalar(select([table.c.id]))
158
def test_autoincrement_on_insert_implcit_returning(self):
161
self.tables.autoinc_pk.insert(),
164
self._assert_round_trip(self.tables.autoinc_pk, config.db)
166
def test_last_inserted_id_implicit_returning(self):
168
r = config.db.execute(
169
self.tables.autoinc_pk.insert(),
172
pk = config.db.scalar(select([self.tables.autoinc_pk.c.id]))
174
r.inserted_primary_key,
179
__all__ = ('LastrowidTest', 'InsertBehaviorTest', 'ReturningTest')