diff options
Diffstat (limited to 'test/sql')
| -rw-r--r-- | test/sql/test_defaults.py | 105 | ||||
| -rw-r--r-- | test/sql/test_insert_exec.py | 204 | ||||
| -rw-r--r-- | test/sql/test_returning.py | 33 | ||||
| -rw-r--r-- | test/sql/test_sequences.py | 308 |
4 files changed, 328 insertions, 322 deletions
diff --git a/test/sql/test_defaults.py b/test/sql/test_defaults.py index 31cc15155..aa3728a98 100644 --- a/test/sql/test_defaults.py +++ b/test/sql/test_defaults.py @@ -20,7 +20,6 @@ from sqlalchemy.sql import select from sqlalchemy.sql import text from sqlalchemy.testing import assert_raises_message from sqlalchemy.testing import AssertsCompiledSQL -from sqlalchemy.testing import engines from sqlalchemy.testing import eq_ from sqlalchemy.testing import expect_warnings from sqlalchemy.testing import fixtures @@ -921,68 +920,70 @@ class CTEDefaultTest(fixtures.TablesTest): eq_(conn.execute(select(q.c.x, q.c.y)).first(), expected) -class PKDefaultTest(fixtures.TablesTest): +class PKDefaultTest(fixtures.TestBase): __requires__ = ("subqueries",) __backend__ = True - @classmethod - def define_tables(cls, metadata): - t2 = Table("t2", metadata, Column("nextid", Integer)) + @testing.fixture + def table_fixture(self, metadata, connection): + def go(implicit_returning): + t2 = Table( + "t2", + metadata, + Column("nextid", Integer), + implicit_returning=implicit_returning, + ) - Table( - "t1", - metadata, - Column( - "id", - Integer, - primary_key=True, - default=sa.select(func.max(t2.c.nextid)).scalar_subquery(), - ), - Column("data", String(30)), - ) + t1 = Table( + "t1", + metadata, + Column( + "id", + Integer, + primary_key=True, + default=sa.select(func.max(t2.c.nextid)).scalar_subquery(), + ), + Column("data", String(30)), + implicit_returning=implicit_returning, + ) - Table( - "date_table", - metadata, - Column( - "date_id", - DateTime(timezone=True), - default=text("current_timestamp"), - primary_key=True, - ), - ) + date_table = Table( + "date_table", + metadata, + Column( + "date_id", + DateTime(timezone=True), + default=text("current_timestamp"), + primary_key=True, + ), + implicit_returning=implicit_returning, + ) - @testing.requires.returning - def test_with_implicit_returning(self): - self._test(True) + metadata.create_all(connection) + return t1, t2, date_table - def test_regular(self): - self._test(False) + return go - def _test(self, returning): - t2, t1, date_table = ( - self.tables.t2, - self.tables.t1, - self.tables.date_table, - ) + @testing.combinations( + (True, testing.requires.returning), + (False,), + argnames="implicit_returning", + ) + def test_pk_default(self, connection, table_fixture, implicit_returning): + t1, t2, date_table = table_fixture(implicit_returning) - if not returning and not testing.db.dialect.implicit_returning: - engine = testing.db - else: - engine = engines.testing_engine( - options={"implicit_returning": returning} - ) - with engine.begin() as conn: - conn.execute(t2.insert(), dict(nextid=1)) - r = conn.execute(t1.insert(), dict(data="hi")) - eq_((1,), r.inserted_primary_key) + conn = connection + + conn.execute(t2.insert(), dict(nextid=1)) + r = conn.execute(t1.insert(), dict(data="hi")) + eq_((1,), r.inserted_primary_key) - conn.execute(t2.insert(), dict(nextid=2)) - r = conn.execute(t1.insert(), dict(data="there")) - eq_((2,), r.inserted_primary_key) + conn.execute(t2.insert(), dict(nextid=2)) + r = conn.execute(t1.insert(), dict(data="there")) + eq_((2,), r.inserted_primary_key) - r = conn.execute(date_table.insert()) - assert isinstance(r.inserted_primary_key[0], datetime.datetime) + r = conn.execute(date_table.insert()) + assert isinstance(r.inserted_primary_key[0], datetime.datetime) class PKIncrementTest(fixtures.TablesTest): 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, diff --git a/test/sql/test_returning.py b/test/sql/test_returning.py index 4069416d4..138e7a4c6 100644 --- a/test/sql/test_returning.py +++ b/test/sql/test_returning.py @@ -16,7 +16,6 @@ from sqlalchemy import update from sqlalchemy.testing import assert_raises_message from sqlalchemy.testing import AssertsCompiledSQL from sqlalchemy.testing import AssertsExecutionResults -from sqlalchemy.testing import engines from sqlalchemy.testing import eq_ from sqlalchemy.testing import fixtures from sqlalchemy.testing.schema import Column @@ -767,35 +766,3 @@ class ReturnDefaultsTest(fixtures.TablesTest): result.inserted_primary_key_rows, [(10,), (11,), (12,), (13,), (14,), (15,)], ) - - -class ImplicitReturningFlag(fixtures.TestBase): - __backend__ = True - - def test_flag_turned_off(self): - e = engines.testing_engine(options={"implicit_returning": False}) - assert e.dialect.implicit_returning is False - c = e.connect() - c.close() - assert e.dialect.implicit_returning is False - - def test_flag_turned_on(self): - e = engines.testing_engine(options={"implicit_returning": True}) - assert e.dialect.implicit_returning is True - c = e.connect() - c.close() - assert e.dialect.implicit_returning is True - - def test_flag_turned_default(self): - supports = [False] - - def go(): - supports[0] = True - - testing.requires.returning(go)() - e = engines.testing_engine() - - # version detection on connect sets it - c = e.connect() - c.close() - assert e.dialect.implicit_returning is supports[0] diff --git a/test/sql/test_sequences.py b/test/sql/test_sequences.py index d6906f9e6..c5c76ad08 100644 --- a/test/sql/test_sequences.py +++ b/test/sql/test_sequences.py @@ -11,7 +11,6 @@ from sqlalchemy.schema import CreateSequence from sqlalchemy.schema import DropSequence from sqlalchemy.sql import select 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_false @@ -180,29 +179,30 @@ class SequenceExecTest(fixtures.TestBase): connection.execute(t1.insert().values(x=s.next_value())) self._assert_seq_result(connection.scalar(t1.select())) - @testing.provide_metadata - def test_inserted_pk_no_returning(self): + def test_inserted_pk_no_returning(self, metadata, connection): """test inserted_primary_key contains [None] when pk_col=next_value(), implicit returning is not used.""" # I'm not really sure what this test wants to accomlish. - metadata = self.metadata - t1 = Table("t", metadata, Column("x", Integer, primary_key=True)) + t1 = Table( + "t", + metadata, + Column("x", Integer, primary_key=True), + implicit_returning=False, + ) s = Sequence("my_sequence_here", metadata=metadata) - e = engines.testing_engine(options={"implicit_returning": False}) - with e.begin() as conn: - - t1.create(conn) - s.create(conn) + conn = connection + t1.create(conn) + s.create(conn) - r = conn.execute(t1.insert().values(x=s.next_value())) + r = conn.execute(t1.insert().values(x=s.next_value())) - if testing.requires.emulated_lastrowid_even_with_sequences.enabled: - eq_(r.inserted_primary_key, (1,)) - else: - eq_(r.inserted_primary_key, (None,)) + if testing.requires.emulated_lastrowid_even_with_sequences.enabled: + eq_(r.inserted_primary_key, (1,)) + else: + eq_(r.inserted_primary_key, (None,)) @testing.combinations( ("implicit_returning",), @@ -213,43 +213,40 @@ class SequenceExecTest(fixtures.TestBase): argnames="returning", ) @testing.requires.multivalues_inserts - def test_seq_multivalues_inline(self, metadata, testing_engine, returning): + def test_seq_multivalues_inline(self, metadata, connection, returning): + _implicit_returning = "no_implicit_returning" not in returning t1 = Table( "t", metadata, Column("x", Integer, Sequence("my_seq"), primary_key=True), Column("data", String(50)), + implicit_returning=_implicit_returning, ) - e = engines.testing_engine( - options={ - "implicit_returning": "no_implicit_returning" not in returning - } + metadata.create_all(connection) + conn = connection + + stmt = t1.insert().values( + [{"data": "d1"}, {"data": "d2"}, {"data": "d3"}] ) - metadata.create_all(e) - with e.begin() as conn: + if returning == "explicit_returning": + stmt = stmt.returning(t1.c.x) + elif "return_defaults" in returning: + stmt = stmt.return_defaults() - stmt = t1.insert().values( - [{"data": "d1"}, {"data": "d2"}, {"data": "d3"}] - ) - if returning == "explicit_returning": - stmt = stmt.returning(t1.c.x) - elif "return_defaults" in returning: - stmt = stmt.return_defaults() - - r = conn.execute(stmt) - if returning == "explicit_returning": - eq_(r.all(), [(1,), (2,), (3,)]) - elif "return_defaults" in returning: - eq_(r.returned_defaults_rows, None) + r = conn.execute(stmt) + if returning == "explicit_returning": + eq_(r.all(), [(1,), (2,), (3,)]) + elif "return_defaults" in returning: + eq_(r.returned_defaults_rows, None) - # TODO: not sure what this is - eq_(r.inserted_primary_key_rows, [(None,)]) + # TODO: not sure what this is + eq_(r.inserted_primary_key_rows, [(None,)]) - eq_( - conn.execute(t1.select().order_by(t1.c.x)).all(), - [(1, "d1"), (2, "d2"), (3, "d3")], - ) + eq_( + conn.execute(t1.select().order_by(t1.c.x)).all(), + [(1, "d1"), (2, "d2"), (3, "d3")], + ) @testing.combinations( ("implicit_returning",), @@ -272,54 +269,49 @@ class SequenceExecTest(fixtures.TestBase): argnames="returning", ) def test_seq_multivalues_executemany( - self, metadata, testing_engine, returning + self, connection, metadata, returning ): + _implicit_returning = "no_implicit_returning" not in returning t1 = Table( "t", metadata, Column("x", Integer, Sequence("my_seq"), primary_key=True), Column("data", String(50)), + implicit_returning=_implicit_returning, ) - e = engines.testing_engine( - options={ - "implicit_returning": "no_implicit_returning" not in returning - } - ) - metadata.create_all(e) - with e.begin() as conn: + metadata.create_all(connection) + conn = connection - stmt = t1.insert() - if returning == "explicit_returning": - stmt = stmt.returning(t1.c.x) - elif "return_defaults" in returning: - stmt = stmt.return_defaults() + stmt = t1.insert() + if returning == "explicit_returning": + stmt = stmt.returning(t1.c.x) + elif "return_defaults" in returning: + stmt = stmt.return_defaults() - r = conn.execute( - stmt, [{"data": "d1"}, {"data": "d2"}, {"data": "d3"}] - ) - if returning == "explicit_returning": - eq_(r.all(), [(1,), (2,), (3,)]) - elif "return_defaults" in returning: - if "no_implicit_returning" in returning: - eq_(r.returned_defaults_rows, None) - eq_(r.inserted_primary_key_rows, [(1,), (2,), (3,)]) - else: - eq_(r.returned_defaults_rows, [(1,), (2,), (3,)]) - eq_(r.inserted_primary_key_rows, [(1,), (2,), (3,)]) - - eq_( - conn.execute(t1.select().order_by(t1.c.x)).all(), - [(1, "d1"), (2, "d2"), (3, "d3")], - ) + r = conn.execute( + stmt, [{"data": "d1"}, {"data": "d2"}, {"data": "d3"}] + ) + if returning == "explicit_returning": + eq_(r.all(), [(1,), (2,), (3,)]) + elif "return_defaults" in returning: + if "no_implicit_returning" in returning: + eq_(r.returned_defaults_rows, None) + eq_(r.inserted_primary_key_rows, [(1,), (2,), (3,)]) + else: + eq_(r.returned_defaults_rows, [(1,), (2,), (3,)]) + eq_(r.inserted_primary_key_rows, [(1,), (2,), (3,)]) + + eq_( + conn.execute(t1.select().order_by(t1.c.x)).all(), + [(1, "d1"), (2, "d2"), (3, "d3")], + ) @testing.requires.returning - @testing.provide_metadata - def test_inserted_pk_implicit_returning(self): + def test_inserted_pk_implicit_returning(self, connection, metadata): """test inserted_primary_key contains the result when pk_col=next_value(), when implicit returning is used.""" - metadata = self.metadata s = Sequence("my_sequence") t1 = Table( "t", @@ -329,13 +321,12 @@ class SequenceExecTest(fixtures.TestBase): Integer, primary_key=True, ), + implicit_returning=True, ) - t1.create(testing.db) + t1.create(connection) - e = engines.testing_engine(options={"implicit_returning": True}) - with e.begin() as conn: - r = conn.execute(t1.insert().values(x=s.next_value())) - self._assert_seq_result(r.inserted_primary_key[0]) + r = connection.execute(t1.insert().values(x=s.next_value())) + self._assert_seq_result(r.inserted_primary_key[0]) class SequenceTest(fixtures.TestBase, testing.AssertsCompiledSQL): @@ -501,42 +492,56 @@ class TableBoundSequenceTest(fixtures.TablesTest): __requires__ = ("sequences",) __backend__ = True - @classmethod - def define_tables(cls, metadata): - Table( - "cartitems", - metadata, - Column( - "cart_id", - Integer, - Sequence("cart_id_seq"), - primary_key=True, - autoincrement=False, - ), - Column("description", String(40)), - Column("createdate", sa.DateTime()), - ) + @testing.fixture + def table_fixture(self, metadata, connection, implicit_returning): + def go(implicit_returning): + cartitems = Table( + "cartitems", + metadata, + Column( + "cart_id", + Integer, + Sequence("cart_id_seq"), + primary_key=True, + autoincrement=False, + ), + Column("description", String(40)), + Column("createdate", sa.DateTime()), + implicit_returning=implicit_returning, + ) - # a little bit of implicit case sensitive naming test going on here - Table( - "Manager", - metadata, - Column( - "obj_id", - Integer, - Sequence("obj_id_seq"), - ), - Column("name", String(128)), - Column( - "id", - Integer, - Sequence("Manager_id_seq", optional=True), - primary_key=True, - ), - ) + # a little bit of implicit case sensitive naming test going on here + Manager = Table( + "Manager", + metadata, + Column( + "obj_id", + Integer, + Sequence("obj_id_seq"), + ), + Column("name", String(128)), + Column( + "id", + Integer, + Sequence("Manager_id_seq", optional=True), + primary_key=True, + ), + implicit_returning=implicit_returning, + ) + metadata.create_all(connection) + return Manager, cartitems - def test_insert_via_seq(self, connection): - cartitems = self.tables.cartitems + return go + + @testing.combinations( + (True, testing.requires.returning), + (False,), + argnames="implicit_returning", + ) + def test_insert_via_seq( + self, table_fixture, connection, implicit_returning + ): + Manager, cartitems = table_fixture(implicit_returning) connection.execute(cartitems.insert(), dict(description="hi")) connection.execute(cartitems.insert(), dict(description="there")) @@ -554,52 +559,53 @@ class TableBoundSequenceTest(fixtures.TablesTest): expected, ) - def test_seq_nonpk(self): + @testing.combinations( + (True, testing.requires.returning), + (False,), + argnames="implicit_returning", + ) + def test_seq_nonpk(self, connection, table_fixture, implicit_returning): """test sequences fire off as defaults on non-pk columns""" - sometable = self.tables.Manager - - engine = engines.testing_engine(options={"implicit_returning": False}) + sometable, cartitems = table_fixture(implicit_returning) - with engine.begin() as conn: - result = conn.execute(sometable.insert(), dict(name="somename")) + conn = connection + result = conn.execute(sometable.insert(), dict(name="somename")) - eq_(result.postfetch_cols(), [sometable.c.obj_id]) + eq_(result.postfetch_cols(), [sometable.c.obj_id]) - result = conn.execute(sometable.insert(), dict(name="someother")) + result = conn.execute(sometable.insert(), dict(name="someother")) - conn.execute( - sometable.insert(), [{"name": "name3"}, {"name": "name4"}] - ) + conn.execute( + sometable.insert(), [{"name": "name3"}, {"name": "name4"}] + ) - dsb = testing.db.dialect.default_sequence_base - eq_( - list( - conn.execute(sometable.select().order_by(sometable.c.id)) + dsb = testing.db.dialect.default_sequence_base + eq_( + list(conn.execute(sometable.select().order_by(sometable.c.id))), + [ + ( + dsb, + "somename", + dsb, ), - [ - ( - dsb, - "somename", - dsb, - ), - ( - dsb + 1, - "someother", - dsb + 1, - ), - ( - dsb + 2, - "name3", - dsb + 2, - ), - ( - dsb + 3, - "name4", - dsb + 3, - ), - ], - ) + ( + dsb + 1, + "someother", + dsb + 1, + ), + ( + dsb + 2, + "name3", + dsb + 2, + ), + ( + dsb + 3, + "name4", + dsb + 3, + ), + ], + ) class SequenceAsServerDefaultTest( |