317 lines
9.4 KiB
Python
317 lines
9.4 KiB
Python
from .. import config
|
|
from .. import engines
|
|
from .. import fixtures
|
|
from ..assertions import eq_
|
|
from ..config import requirements
|
|
from ..schema import Column
|
|
from ..schema import Table
|
|
from ... import Integer
|
|
from ... import literal
|
|
from ... import literal_column
|
|
from ... import select
|
|
from ... import String
|
|
|
|
|
|
class LastrowidTest(fixtures.TablesTest):
|
|
run_deletes = "each"
|
|
|
|
__backend__ = True
|
|
|
|
__requires__ = "implements_get_lastrowid", "autoincrement_insert"
|
|
|
|
__engine_options__ = {"implicit_returning": False}
|
|
|
|
@classmethod
|
|
def define_tables(cls, metadata):
|
|
Table(
|
|
"autoinc_pk",
|
|
metadata,
|
|
Column(
|
|
"id", Integer, primary_key=True, test_needs_autoincrement=True
|
|
),
|
|
Column("data", String(50)),
|
|
)
|
|
|
|
Table(
|
|
"manual_pk",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True, autoincrement=False),
|
|
Column("data", String(50)),
|
|
)
|
|
|
|
def _assert_round_trip(self, table, conn):
|
|
row = conn.execute(table.select()).first()
|
|
eq_(row, (config.db.dialect.default_sequence_base, "some data"))
|
|
|
|
def test_autoincrement_on_insert(self):
|
|
|
|
config.db.execute(self.tables.autoinc_pk.insert(), data="some data")
|
|
self._assert_round_trip(self.tables.autoinc_pk, config.db)
|
|
|
|
def test_last_inserted_id(self):
|
|
|
|
r = config.db.execute(
|
|
self.tables.autoinc_pk.insert(), data="some data"
|
|
)
|
|
pk = config.db.scalar(select([self.tables.autoinc_pk.c.id]))
|
|
eq_(r.inserted_primary_key, [pk])
|
|
|
|
# failed on pypy1.9 but seems to be OK on pypy 2.1
|
|
# @exclusions.fails_if(lambda: util.pypy,
|
|
# "lastrowid not maintained after "
|
|
# "connection close")
|
|
@requirements.dbapi_lastrowid
|
|
def test_native_lastrowid_autoinc(self):
|
|
r = config.db.execute(
|
|
self.tables.autoinc_pk.insert(), data="some data"
|
|
)
|
|
lastrowid = r.lastrowid
|
|
pk = config.db.scalar(select([self.tables.autoinc_pk.c.id]))
|
|
eq_(lastrowid, pk)
|
|
|
|
|
|
class InsertBehaviorTest(fixtures.TablesTest):
|
|
run_deletes = "each"
|
|
__backend__ = True
|
|
|
|
@classmethod
|
|
def define_tables(cls, metadata):
|
|
Table(
|
|
"autoinc_pk",
|
|
metadata,
|
|
Column(
|
|
"id", Integer, primary_key=True, test_needs_autoincrement=True
|
|
),
|
|
Column("data", String(50)),
|
|
)
|
|
Table(
|
|
"manual_pk",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True, autoincrement=False),
|
|
Column("data", String(50)),
|
|
)
|
|
Table(
|
|
"includes_defaults",
|
|
metadata,
|
|
Column(
|
|
"id", Integer, primary_key=True, test_needs_autoincrement=True
|
|
),
|
|
Column("data", String(50)),
|
|
Column("x", Integer, default=5),
|
|
Column(
|
|
"y",
|
|
Integer,
|
|
default=literal_column("2", type_=Integer) + literal(2),
|
|
),
|
|
)
|
|
|
|
def test_autoclose_on_insert(self):
|
|
if requirements.returning.enabled:
|
|
engine = engines.testing_engine(
|
|
options={"implicit_returning": False}
|
|
)
|
|
else:
|
|
engine = config.db
|
|
|
|
r = engine.execute(self.tables.autoinc_pk.insert(), data="some data")
|
|
assert r._soft_closed
|
|
assert not r.closed
|
|
assert r.is_insert
|
|
assert not r.returns_rows
|
|
|
|
@requirements.returning
|
|
def test_autoclose_on_insert_implicit_returning(self):
|
|
r = config.db.execute(
|
|
self.tables.autoinc_pk.insert(), data="some data"
|
|
)
|
|
assert r._soft_closed
|
|
assert not r.closed
|
|
assert r.is_insert
|
|
assert not r.returns_rows
|
|
|
|
@requirements.empty_inserts
|
|
def test_empty_insert(self):
|
|
r = config.db.execute(self.tables.autoinc_pk.insert())
|
|
assert r._soft_closed
|
|
assert not r.closed
|
|
|
|
r = config.db.execute(
|
|
self.tables.autoinc_pk.select().where(
|
|
self.tables.autoinc_pk.c.id != None
|
|
)
|
|
)
|
|
|
|
assert len(r.fetchall())
|
|
|
|
@requirements.insert_from_select
|
|
def test_insert_from_select_autoinc(self):
|
|
src_table = self.tables.manual_pk
|
|
dest_table = self.tables.autoinc_pk
|
|
config.db.execute(
|
|
src_table.insert(),
|
|
[
|
|
dict(id=1, data="data1"),
|
|
dict(id=2, data="data2"),
|
|
dict(id=3, data="data3"),
|
|
],
|
|
)
|
|
|
|
result = config.db.execute(
|
|
dest_table.insert().from_select(
|
|
("data",),
|
|
select([src_table.c.data]).where(
|
|
src_table.c.data.in_(["data2", "data3"])
|
|
),
|
|
)
|
|
)
|
|
|
|
eq_(result.inserted_primary_key, [None])
|
|
|
|
result = config.db.execute(
|
|
select([dest_table.c.data]).order_by(dest_table.c.data)
|
|
)
|
|
eq_(result.fetchall(), [("data2",), ("data3",)])
|
|
|
|
@requirements.insert_from_select
|
|
def test_insert_from_select_autoinc_no_rows(self):
|
|
src_table = self.tables.manual_pk
|
|
dest_table = self.tables.autoinc_pk
|
|
|
|
result = config.db.execute(
|
|
dest_table.insert().from_select(
|
|
("data",),
|
|
select([src_table.c.data]).where(
|
|
src_table.c.data.in_(["data2", "data3"])
|
|
),
|
|
)
|
|
)
|
|
eq_(result.inserted_primary_key, [None])
|
|
|
|
result = config.db.execute(
|
|
select([dest_table.c.data]).order_by(dest_table.c.data)
|
|
)
|
|
|
|
eq_(result.fetchall(), [])
|
|
|
|
@requirements.insert_from_select
|
|
def test_insert_from_select(self):
|
|
table = self.tables.manual_pk
|
|
config.db.execute(
|
|
table.insert(),
|
|
[
|
|
dict(id=1, data="data1"),
|
|
dict(id=2, data="data2"),
|
|
dict(id=3, data="data3"),
|
|
],
|
|
)
|
|
|
|
config.db.execute(
|
|
table.insert(inline=True).from_select(
|
|
("id", "data"),
|
|
select([table.c.id + 5, table.c.data]).where(
|
|
table.c.data.in_(["data2", "data3"])
|
|
),
|
|
)
|
|
)
|
|
|
|
eq_(
|
|
config.db.execute(
|
|
select([table.c.data]).order_by(table.c.data)
|
|
).fetchall(),
|
|
[("data1",), ("data2",), ("data2",), ("data3",), ("data3",)],
|
|
)
|
|
|
|
@requirements.insert_from_select
|
|
def test_insert_from_select_with_defaults(self):
|
|
table = self.tables.includes_defaults
|
|
config.db.execute(
|
|
table.insert(),
|
|
[
|
|
dict(id=1, data="data1"),
|
|
dict(id=2, data="data2"),
|
|
dict(id=3, data="data3"),
|
|
],
|
|
)
|
|
|
|
config.db.execute(
|
|
table.insert(inline=True).from_select(
|
|
("id", "data"),
|
|
select([table.c.id + 5, table.c.data]).where(
|
|
table.c.data.in_(["data2", "data3"])
|
|
),
|
|
)
|
|
)
|
|
|
|
eq_(
|
|
config.db.execute(
|
|
select([table]).order_by(table.c.data, table.c.id)
|
|
).fetchall(),
|
|
[
|
|
(1, "data1", 5, 4),
|
|
(2, "data2", 5, 4),
|
|
(7, "data2", 5, 4),
|
|
(3, "data3", 5, 4),
|
|
(8, "data3", 5, 4),
|
|
],
|
|
)
|
|
|
|
|
|
class ReturningTest(fixtures.TablesTest):
|
|
run_create_tables = "each"
|
|
__requires__ = "returning", "autoincrement_insert"
|
|
__backend__ = True
|
|
|
|
__engine_options__ = {"implicit_returning": True}
|
|
|
|
def _assert_round_trip(self, table, conn):
|
|
row = conn.execute(table.select()).first()
|
|
eq_(row, (config.db.dialect.default_sequence_base, "some data"))
|
|
|
|
@classmethod
|
|
def define_tables(cls, metadata):
|
|
Table(
|
|
"autoinc_pk",
|
|
metadata,
|
|
Column(
|
|
"id", Integer, primary_key=True, test_needs_autoincrement=True
|
|
),
|
|
Column("data", String(50)),
|
|
)
|
|
|
|
@requirements.fetch_rows_post_commit
|
|
def test_explicit_returning_pk_autocommit(self):
|
|
engine = config.db
|
|
table = self.tables.autoinc_pk
|
|
r = engine.execute(
|
|
table.insert().returning(table.c.id), data="some data"
|
|
)
|
|
pk = r.first()[0]
|
|
fetched_pk = config.db.scalar(select([table.c.id]))
|
|
eq_(fetched_pk, pk)
|
|
|
|
def test_explicit_returning_pk_no_autocommit(self):
|
|
engine = config.db
|
|
table = self.tables.autoinc_pk
|
|
with engine.begin() as conn:
|
|
r = conn.execute(
|
|
table.insert().returning(table.c.id), data="some data"
|
|
)
|
|
pk = r.first()[0]
|
|
fetched_pk = config.db.scalar(select([table.c.id]))
|
|
eq_(fetched_pk, pk)
|
|
|
|
def test_autoincrement_on_insert_implicit_returning(self):
|
|
|
|
config.db.execute(self.tables.autoinc_pk.insert(), data="some data")
|
|
self._assert_round_trip(self.tables.autoinc_pk, config.db)
|
|
|
|
def test_last_inserted_id_implicit_returning(self):
|
|
|
|
r = config.db.execute(
|
|
self.tables.autoinc_pk.insert(), data="some data"
|
|
)
|
|
pk = config.db.scalar(select([self.tables.autoinc_pk.c.id]))
|
|
eq_(r.inserted_primary_key, [pk])
|
|
|
|
|
|
__all__ = ("LastrowidTest", "InsertBehaviorTest", "ReturningTest")
|