diff options
Diffstat (limited to 'test/sql/test_insert_exec.py')
| -rw-r--r-- | test/sql/test_insert_exec.py | 204 |
1 files changed, 118 insertions, 86 deletions
diff --git a/test/sql/test_insert_exec.py b/test/sql/test_insert_exec.py index e0e3b60a9..ef1a3be09 100644 --- a/test/sql/test_insert_exec.py +++ b/test/sql/test_insert_exec.py @@ -5,17 +5,16 @@ from sqlalchemy import func from sqlalchemy import INT from sqlalchemy import Integer from sqlalchemy import literal -from sqlalchemy import MetaData from sqlalchemy import Sequence from sqlalchemy import sql from sqlalchemy import String from sqlalchemy import testing from sqlalchemy import VARCHAR from sqlalchemy.testing import assert_raises_message -from sqlalchemy.testing import engines from sqlalchemy.testing import eq_ from sqlalchemy.testing import fixtures from sqlalchemy.testing import is_ +from sqlalchemy.testing import mock from sqlalchemy.testing.schema import Column from sqlalchemy.testing.schema import Table @@ -89,10 +88,10 @@ class InsertExecTest(fixtures.TablesTest): ], ) - def _test_lastrow_accessor(self, table_, values, assertvalues): + def _test_lastrow_accessor(self, connection, table_, values, assertvalues): """Tests the inserted_primary_key and lastrow_has_id() functions.""" - def insert_values(engine, table_, values): + def insert_values(table_, values): """ Inserts a row into a table, returns the full list of values INSERTed including defaults that fired off on the DB side and @@ -100,71 +99,62 @@ class InsertExecTest(fixtures.TablesTest): """ # verify implicit_returning is working - if engine.dialect.implicit_returning: + if ( + connection.dialect.implicit_returning + and table_.implicit_returning + ): ins = table_.insert() - comp = ins.compile(engine, column_keys=list(values)) + comp = ins.compile(connection, column_keys=list(values)) if not set(values).issuperset( c.key for c in table_.primary_key ): is_(bool(comp.returning), True) - with engine.begin() as connection: - result = connection.execute(table_.insert(), values) - ret = values.copy() - - ipk = result.inserted_primary_key - for col, id_ in zip(table_.primary_key, ipk): - ret[col.key] = id_ - - if result.lastrow_has_defaults(): - criterion = and_( - *[ - col == id_ - for col, id_ in zip( - table_.primary_key, result.inserted_primary_key - ) - ] - ) - row = connection.execute( - table_.select().where(criterion) - ).first() - for c in table_.c: - ret[c.key] = row._mapping[c] + result = connection.execute(table_.insert(), values) + ret = values.copy() + + ipk = result.inserted_primary_key + for col, id_ in zip(table_.primary_key, ipk): + ret[col.key] = id_ + + if result.lastrow_has_defaults(): + criterion = and_( + *[ + col == id_ + for col, id_ in zip( + table_.primary_key, result.inserted_primary_key + ) + ] + ) + row = connection.execute( + table_.select().where(criterion) + ).first() + for c in table_.c: + ret[c.key] = row._mapping[c] return ret, ipk - if testing.against("postgresql", "oracle", "mssql"): - assert testing.db.dialect.implicit_returning - - if testing.db.dialect.implicit_returning: - test_engines = [ - engines.testing_engine(options={"implicit_returning": False}), - engines.testing_engine(options={"implicit_returning": True}), - ] - else: - test_engines = [testing.db] - - for engine in test_engines: - try: - table_.create(bind=engine, checkfirst=True) - i, ipk = insert_values(engine, table_, values) - eq_(i, assertvalues) + table_.create(connection, checkfirst=True) + i, ipk = insert_values(table_, values) + eq_(i, assertvalues) - # named tuple tests - for col in table_.primary_key: - eq_(getattr(ipk, col.key), assertvalues[col.key]) - eq_(ipk._mapping[col.key], assertvalues[col.key]) + # named tuple tests + for col in table_.primary_key: + eq_(getattr(ipk, col.key), assertvalues[col.key]) + eq_(ipk._mapping[col.key], assertvalues[col.key]) - eq_( - ipk._fields, tuple([col.key for col in table_.primary_key]) - ) - - finally: - table_.drop(bind=engine) + eq_(ipk._fields, tuple([col.key for col in table_.primary_key])) - @testing.skip_if("sqlite") - def test_lastrow_accessor_one(self): - metadata = MetaData() + @testing.requires.supports_autoincrement_w_composite_pk + @testing.combinations( + (True, testing.requires.returning), + (False,), + argnames="implicit_returning", + ) + def test_lastrow_accessor_one( + self, metadata, connection, implicit_returning + ): self._test_lastrow_accessor( + connection, Table( "t1", metadata, @@ -175,15 +165,23 @@ class InsertExecTest(fixtures.TablesTest): test_needs_autoincrement=True, ), Column("foo", String(30), primary_key=True), + implicit_returning=implicit_returning, ), {"foo": "hi"}, {"id": 1, "foo": "hi"}, ) - @testing.skip_if("sqlite") - def test_lastrow_accessor_two(self): - metadata = MetaData() + @testing.requires.supports_autoincrement_w_composite_pk + @testing.combinations( + (True, testing.requires.returning), + (False,), + argnames="implicit_returning", + ) + def test_lastrow_accessor_two( + self, metadata, connection, implicit_returning + ): self._test_lastrow_accessor( + connection, Table( "t2", metadata, @@ -195,29 +193,45 @@ class InsertExecTest(fixtures.TablesTest): ), Column("foo", String(30), primary_key=True), Column("bar", String(30), server_default="hi"), + implicit_returning=implicit_returning, ), {"foo": "hi"}, {"id": 1, "foo": "hi", "bar": "hi"}, ) - def test_lastrow_accessor_three(self): - metadata = MetaData() + @testing.combinations( + (True, testing.requires.returning), + (False,), + argnames="implicit_returning", + ) + def test_lastrow_accessor_three( + self, metadata, connection, implicit_returning + ): self._test_lastrow_accessor( + connection, Table( "t3", metadata, Column("id", String(40), primary_key=True), Column("foo", String(30), primary_key=True), Column("bar", String(30)), + implicit_returning=implicit_returning, ), {"id": "hi", "foo": "thisisfoo", "bar": "thisisbar"}, {"id": "hi", "foo": "thisisfoo", "bar": "thisisbar"}, ) @testing.requires.sequences - def test_lastrow_accessor_four(self): - metadata = MetaData() + @testing.combinations( + (True, testing.requires.returning), + (False,), + argnames="implicit_returning", + ) + def test_lastrow_accessor_four( + self, metadata, connection, implicit_returning + ): self._test_lastrow_accessor( + connection, Table( "t4", metadata, @@ -229,15 +243,23 @@ class InsertExecTest(fixtures.TablesTest): ), Column("foo", String(30), primary_key=True), Column("bar", String(30), server_default="hi"), + implicit_returning=implicit_returning, ), {"foo": "hi", "id": 1}, {"id": 1, "foo": "hi", "bar": "hi"}, ) @testing.requires.sequences - def test_lastrow_accessor_four_a(self): - metadata = MetaData() + @testing.combinations( + (True, testing.requires.returning), + (False,), + argnames="implicit_returning", + ) + def test_lastrow_accessor_four_a( + self, metadata, connection, implicit_returning + ): self._test_lastrow_accessor( + connection, Table( "t4", metadata, @@ -248,28 +270,44 @@ class InsertExecTest(fixtures.TablesTest): primary_key=True, ), Column("foo", String(30)), + implicit_returning=implicit_returning, ), {"foo": "hi"}, {"id": 1, "foo": "hi"}, ) - def test_lastrow_accessor_five(self): - metadata = MetaData() + @testing.combinations( + (True, testing.requires.returning), + (False,), + argnames="implicit_returning", + ) + def test_lastrow_accessor_five( + self, metadata, connection, implicit_returning + ): self._test_lastrow_accessor( + connection, Table( "t5", metadata, Column("id", String(10), primary_key=True), Column("bar", String(30), server_default="hi"), + implicit_returning=implicit_returning, ), {"id": "id1"}, {"id": "id1", "bar": "hi"}, ) - @testing.skip_if("sqlite") - def test_lastrow_accessor_six(self): - metadata = MetaData() + @testing.requires.supports_autoincrement_w_composite_pk + @testing.combinations( + (True, testing.requires.returning), + (False,), + argnames="implicit_returning", + ) + def test_lastrow_accessor_six( + self, metadata, connection, implicit_returning + ): self._test_lastrow_accessor( + connection, Table( "t6", metadata, @@ -280,6 +318,7 @@ class InsertExecTest(fixtures.TablesTest): test_needs_autoincrement=True, ), Column("bar", Integer, primary_key=True), + implicit_returning=implicit_returning, ), {"bar": 0}, {"id": 1, "bar": 0}, @@ -287,35 +326,28 @@ class InsertExecTest(fixtures.TablesTest): # TODO: why not in the sqlite suite? @testing.only_on("sqlite+pysqlite") - @testing.provide_metadata - def test_lastrowid_zero(self): + def test_lastrowid_zero(self, metadata, connection): from sqlalchemy.dialects import sqlite - eng = engines.testing_engine() - class ExcCtx(sqlite.base.SQLiteExecutionContext): def get_lastrowid(self): return 0 - eng.dialect.execution_ctx_cls = ExcCtx t = Table( "t", self.metadata, Column("x", Integer, primary_key=True), Column("y", Integer), ) - with eng.begin() as conn: - t.create(conn) - r = conn.execute(t.insert().values(y=5)) + t.create(connection) + with mock.patch.object( + connection.dialect, "execution_ctx_cls", ExcCtx + ): + r = connection.execute(t.insert().values(y=5)) eq_(r.inserted_primary_key, (0,)) - @testing.fails_on( - "sqlite", "sqlite autoincrement doesn't work with composite pks" - ) - @testing.provide_metadata - def test_misordered_lastrow(self, connection): - metadata = self.metadata - + @testing.requires.supports_autoincrement_w_composite_pk + def test_misordered_lastrow(self, connection, metadata): related = Table( "related", metadata, |