diff options
Diffstat (limited to 'test/dialect/postgresql/test_query.py')
| -rw-r--r-- | test/dialect/postgresql/test_query.py | 628 |
1 files changed, 312 insertions, 316 deletions
diff --git a/test/dialect/postgresql/test_query.py b/test/dialect/postgresql/test_query.py index a1e9c4657..04bce4e22 100644 --- a/test/dialect/postgresql/test_query.py +++ b/test/dialect/postgresql/test_query.py @@ -45,27 +45,25 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults): __only_on__ = "postgresql" __backend__ = True - def setup_test(self): - self.metadata = MetaData() + @testing.combinations((False,), (True,), argnames="implicit_returning") + def test_foreignkey_missing_insert( + self, metadata, connection, implicit_returning + ): - def teardown_test(self): - with testing.db.begin() as conn: - self.metadata.drop_all(conn) - - @testing.combinations((False,), (True,)) - def test_foreignkey_missing_insert(self, implicit_returning): - engine = engines.testing_engine( - options={"implicit_returning": implicit_returning} + Table( + "t1", + metadata, + Column("id", Integer, primary_key=True), + implicit_returning=implicit_returning, ) - - Table("t1", self.metadata, Column("id", Integer, primary_key=True)) t2 = Table( "t2", - self.metadata, + metadata, Column("id", Integer, ForeignKey("t1.id"), primary_key=True), + implicit_returning=implicit_returning, ) - self.metadata.create_all(engine) + metadata.create_all(connection) # want to ensure that "null value in column "id" violates not- # null constraint" is raised (IntegrityError on psycoopg2, but @@ -75,54 +73,36 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults): # the case here due to the foreign key. with expect_warnings(".*has no Python-side or server-side default.*"): - with engine.begin() as conn: - assert_raises( - (exc.IntegrityError, exc.ProgrammingError), - conn.execute, - t2.insert(), - ) - - def test_sequence_insert(self): - table = Table( - "testtable", - self.metadata, - Column("id", Integer, Sequence("my_seq"), primary_key=True), - Column("data", String(30)), - ) - self.metadata.create_all(testing.db) - self._assert_data_with_sequence(table, "my_seq") + assert_raises( + (exc.IntegrityError, exc.ProgrammingError), + connection.execute, + t2.insert(), + ) - @testing.requires.returning - def test_sequence_returning_insert(self): + @testing.combinations(True, False, argnames="implicit_returning") + def test_sequence_insert(self, metadata, connection, implicit_returning): table = Table( "testtable", - self.metadata, + metadata, Column("id", Integer, Sequence("my_seq"), primary_key=True), Column("data", String(30)), + implicit_returning=implicit_returning, ) - self.metadata.create_all(testing.db) - self._assert_data_with_sequence_returning(table, "my_seq") - - def test_opt_sequence_insert(self): - table = Table( - "testtable", - self.metadata, - Column( - "id", - Integer, - Sequence("my_seq", optional=True), - primary_key=True, - ), - Column("data", String(30)), - ) - self.metadata.create_all(testing.db) - self._assert_data_autoincrement(table) + metadata.create_all(connection) + if implicit_returning: + self._assert_data_with_sequence_returning( + connection, table, "my_seq" + ) + else: + self._assert_data_with_sequence(connection, table, "my_seq") - @testing.requires.returning - def test_opt_sequence_returning_insert(self): + @testing.combinations(True, False, argnames="implicit_returning") + def test_opt_sequence_insert( + self, metadata, connection, implicit_returning + ): table = Table( "testtable", - self.metadata, + metadata, Column( "id", Integer, @@ -130,78 +110,85 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults): primary_key=True, ), Column("data", String(30)), + implicit_returning=implicit_returning, ) - self.metadata.create_all(testing.db) - self._assert_data_autoincrement_returning(table) - - def test_autoincrement_insert(self): - table = Table( - "testtable", - self.metadata, - Column("id", Integer, primary_key=True), - Column("data", String(30)), - ) - self.metadata.create_all(testing.db) - self._assert_data_autoincrement(table) + metadata.create_all(connection) + if implicit_returning: + self._assert_data_autoincrement_returning(connection, table) + else: + self._assert_data_autoincrement(connection, table) - @testing.requires.returning - def test_autoincrement_returning_insert(self): + @testing.combinations(True, False, argnames="implicit_returning") + def test_autoincrement_insert( + self, metadata, connection, implicit_returning + ): table = Table( "testtable", - self.metadata, + metadata, Column("id", Integer, primary_key=True), Column("data", String(30)), + implicit_returning=implicit_returning, ) - self.metadata.create_all(testing.db) - self._assert_data_autoincrement_returning(table) + metadata.create_all(connection) + if implicit_returning: + self._assert_data_autoincrement_returning(connection, table) + else: + self._assert_data_autoincrement(connection, table) - def test_noautoincrement_insert(self): + @testing.combinations(True, False, argnames="implicit_returning") + def test_noautoincrement_insert( + self, metadata, connection, implicit_returning + ): table = Table( "testtable", - self.metadata, + metadata, Column("id", Integer, primary_key=True, autoincrement=False), Column("data", String(30)), + implicit_returning=implicit_returning, ) - self.metadata.create_all(testing.db) - self._assert_data_noautoincrement(table) - - def _assert_data_autoincrement(self, table): - engine = engines.testing_engine(options={"implicit_returning": False}) + metadata.create_all(connection) + self._assert_data_noautoincrement(connection, table) - with self.sql_execution_asserter(engine) as asserter: + def _assert_data_autoincrement(self, connection, table): + """ + invoked by: + * test_opt_sequence_insert + * test_autoincrement_insert + """ - with engine.begin() as conn: - # execute with explicit id + with self.sql_execution_asserter(connection) as asserter: + conn = connection + # execute with explicit id - r = conn.execute(table.insert(), {"id": 30, "data": "d1"}) - eq_(r.inserted_primary_key, (30,)) + r = conn.execute(table.insert(), {"id": 30, "data": "d1"}) + eq_(r.inserted_primary_key, (30,)) - # execute with prefetch id + # execute with prefetch id - r = conn.execute(table.insert(), {"data": "d2"}) - eq_(r.inserted_primary_key, (1,)) + r = conn.execute(table.insert(), {"data": "d2"}) + eq_(r.inserted_primary_key, (1,)) - # executemany with explicit ids + # executemany with explicit ids - conn.execute( - table.insert(), - [ - {"id": 31, "data": "d3"}, - {"id": 32, "data": "d4"}, - ], - ) + conn.execute( + table.insert(), + [ + {"id": 31, "data": "d3"}, + {"id": 32, "data": "d4"}, + ], + ) - # executemany, uses SERIAL + # executemany, uses SERIAL - conn.execute(table.insert(), [{"data": "d5"}, {"data": "d6"}]) + conn.execute(table.insert(), [{"data": "d5"}, {"data": "d6"}]) - # single execute, explicit id, inline + # single execute, explicit id, inline - conn.execute(table.insert().inline(), {"id": 33, "data": "d7"}) + conn.execute(table.insert().inline(), {"id": 33, "data": "d7"}) - # single execute, inline, uses SERIAL + # single execute, inline, uses SERIAL - conn.execute(table.insert().inline(), {"data": "d8"}) + conn.execute(table.insert().inline(), {"data": "d8"}) asserter.assert_( DialectSQL( @@ -229,44 +216,44 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults): ), ) - with engine.begin() as conn: - eq_( - conn.execute(table.select()).fetchall(), - [ - (30, "d1"), - (1, "d2"), - (31, "d3"), - (32, "d4"), - (2, "d5"), - (3, "d6"), - (33, "d7"), - (4, "d8"), - ], - ) + eq_( + conn.execute(table.select()).fetchall(), + [ + (30, "d1"), + (1, "d2"), + (31, "d3"), + (32, "d4"), + (2, "d5"), + (3, "d6"), + (33, "d7"), + (4, "d8"), + ], + ) - conn.execute(table.delete()) + conn.execute(table.delete()) # test the same series of events using a reflected version of # the table m2 = MetaData() - table = Table(table.name, m2, autoload_with=engine) - - with self.sql_execution_asserter(engine) as asserter: - with engine.begin() as conn: - conn.execute(table.insert(), {"id": 30, "data": "d1"}) - r = conn.execute(table.insert(), {"data": "d2"}) - eq_(r.inserted_primary_key, (5,)) - conn.execute( - table.insert(), - [ - {"id": 31, "data": "d3"}, - {"id": 32, "data": "d4"}, - ], - ) - conn.execute(table.insert(), [{"data": "d5"}, {"data": "d6"}]) - conn.execute(table.insert().inline(), {"id": 33, "data": "d7"}) - conn.execute(table.insert().inline(), {"data": "d8"}) + table = Table( + table.name, m2, autoload_with=connection, implicit_returning=False + ) + + with self.sql_execution_asserter(connection) as asserter: + conn.execute(table.insert(), {"id": 30, "data": "d1"}) + r = conn.execute(table.insert(), {"data": "d2"}) + eq_(r.inserted_primary_key, (5,)) + conn.execute( + table.insert(), + [ + {"id": 31, "data": "d3"}, + {"id": 32, "data": "d4"}, + ], + ) + conn.execute(table.insert(), [{"data": "d5"}, {"data": "d6"}]) + conn.execute(table.insert().inline(), {"id": 33, "data": "d7"}) + conn.execute(table.insert().inline(), {"data": "d8"}) asserter.assert_( DialectSQL( @@ -293,59 +280,61 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults): "INSERT INTO testtable (data) VALUES (:data)", [{"data": "d8"}] ), ) - with engine.begin() as conn: - eq_( - conn.execute(table.select()).fetchall(), - [ - (30, "d1"), - (5, "d2"), - (31, "d3"), - (32, "d4"), - (6, "d5"), - (7, "d6"), - (33, "d7"), - (8, "d8"), - ], - ) - conn.execute(table.delete()) - def _assert_data_autoincrement_returning(self, table): - engine = engines.testing_engine(options={"implicit_returning": True}) + eq_( + conn.execute(table.select()).fetchall(), + [ + (30, "d1"), + (5, "d2"), + (31, "d3"), + (32, "d4"), + (6, "d5"), + (7, "d6"), + (33, "d7"), + (8, "d8"), + ], + ) - with self.sql_execution_asserter(engine) as asserter: - with engine.begin() as conn: + def _assert_data_autoincrement_returning(self, connection, table): + """ + invoked by: + * test_opt_sequence_returning_insert + * test_autoincrement_returning_insert + """ + with self.sql_execution_asserter(connection) as asserter: + conn = connection - # execute with explicit id + # execute with explicit id - r = conn.execute(table.insert(), {"id": 30, "data": "d1"}) - eq_(r.inserted_primary_key, (30,)) + r = conn.execute(table.insert(), {"id": 30, "data": "d1"}) + eq_(r.inserted_primary_key, (30,)) - # execute with prefetch id + # execute with prefetch id - r = conn.execute(table.insert(), {"data": "d2"}) - eq_(r.inserted_primary_key, (1,)) + r = conn.execute(table.insert(), {"data": "d2"}) + eq_(r.inserted_primary_key, (1,)) - # executemany with explicit ids + # executemany with explicit ids - conn.execute( - table.insert(), - [ - {"id": 31, "data": "d3"}, - {"id": 32, "data": "d4"}, - ], - ) + conn.execute( + table.insert(), + [ + {"id": 31, "data": "d3"}, + {"id": 32, "data": "d4"}, + ], + ) - # executemany, uses SERIAL + # executemany, uses SERIAL - conn.execute(table.insert(), [{"data": "d5"}, {"data": "d6"}]) + conn.execute(table.insert(), [{"data": "d5"}, {"data": "d6"}]) - # single execute, explicit id, inline + # single execute, explicit id, inline - conn.execute(table.insert().inline(), {"id": 33, "data": "d7"}) + conn.execute(table.insert().inline(), {"id": 33, "data": "d7"}) - # single execute, inline, uses SERIAL + # single execute, inline, uses SERIAL - conn.execute(table.insert().inline(), {"data": "d8"}) + conn.execute(table.insert().inline(), {"data": "d8"}) asserter.assert_( DialectSQL( @@ -374,43 +363,46 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults): ), ) - with engine.begin() as conn: - eq_( - conn.execute(table.select()).fetchall(), - [ - (30, "d1"), - (1, "d2"), - (31, "d3"), - (32, "d4"), - (2, "d5"), - (3, "d6"), - (33, "d7"), - (4, "d8"), - ], - ) - conn.execute(table.delete()) + eq_( + conn.execute(table.select()).fetchall(), + [ + (30, "d1"), + (1, "d2"), + (31, "d3"), + (32, "d4"), + (2, "d5"), + (3, "d6"), + (33, "d7"), + (4, "d8"), + ], + ) + conn.execute(table.delete()) # test the same series of events using a reflected version of # the table m2 = MetaData() - table = Table(table.name, m2, autoload_with=engine) - - with self.sql_execution_asserter(engine) as asserter: - with engine.begin() as conn: - conn.execute(table.insert(), {"id": 30, "data": "d1"}) - r = conn.execute(table.insert(), {"data": "d2"}) - eq_(r.inserted_primary_key, (5,)) - conn.execute( - table.insert(), - [ - {"id": 31, "data": "d3"}, - {"id": 32, "data": "d4"}, - ], - ) - conn.execute(table.insert(), [{"data": "d5"}, {"data": "d6"}]) - conn.execute(table.insert().inline(), {"id": 33, "data": "d7"}) - conn.execute(table.insert().inline(), {"data": "d8"}) + table = Table( + table.name, + m2, + autoload_with=connection, + implicit_returning=True, + ) + + with self.sql_execution_asserter(connection) as asserter: + conn.execute(table.insert(), {"id": 30, "data": "d1"}) + r = conn.execute(table.insert(), {"data": "d2"}) + eq_(r.inserted_primary_key, (5,)) + conn.execute( + table.insert(), + [ + {"id": 31, "data": "d3"}, + {"id": 32, "data": "d4"}, + ], + ) + conn.execute(table.insert(), [{"data": "d5"}, {"data": "d6"}]) + conn.execute(table.insert().inline(), {"id": 33, "data": "d7"}) + conn.execute(table.insert().inline(), {"data": "d8"}) asserter.assert_( DialectSQL( @@ -439,39 +431,40 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults): ), ) - with engine.begin() as conn: - eq_( - conn.execute(table.select()).fetchall(), + eq_( + conn.execute(table.select()).fetchall(), + [ + (30, "d1"), + (5, "d2"), + (31, "d3"), + (32, "d4"), + (6, "d5"), + (7, "d6"), + (33, "d7"), + (8, "d8"), + ], + ) + + def _assert_data_with_sequence(self, connection, table, seqname): + """ + invoked by: + * test_sequence_insert + """ + + with self.sql_execution_asserter(connection) as asserter: + conn = connection + conn.execute(table.insert(), {"id": 30, "data": "d1"}) + conn.execute(table.insert(), {"data": "d2"}) + conn.execute( + table.insert(), [ - (30, "d1"), - (5, "d2"), - (31, "d3"), - (32, "d4"), - (6, "d5"), - (7, "d6"), - (33, "d7"), - (8, "d8"), + {"id": 31, "data": "d3"}, + {"id": 32, "data": "d4"}, ], ) - conn.execute(table.delete()) - - def _assert_data_with_sequence(self, table, seqname): - engine = engines.testing_engine(options={"implicit_returning": False}) - - with self.sql_execution_asserter(engine) as asserter: - with engine.begin() as conn: - conn.execute(table.insert(), {"id": 30, "data": "d1"}) - conn.execute(table.insert(), {"data": "d2"}) - conn.execute( - table.insert(), - [ - {"id": 31, "data": "d3"}, - {"id": 32, "data": "d4"}, - ], - ) - conn.execute(table.insert(), [{"data": "d5"}, {"data": "d6"}]) - conn.execute(table.insert().inline(), {"id": 33, "data": "d7"}) - conn.execute(table.insert().inline(), {"data": "d8"}) + conn.execute(table.insert(), [{"data": "d5"}, {"data": "d6"}]) + conn.execute(table.insert().inline(), {"id": 33, "data": "d7"}) + conn.execute(table.insert().inline(), {"data": "d8"}) asserter.assert_( DialectSQL( @@ -502,41 +495,40 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults): [{"data": "d8"}], ), ) - with engine.begin() as conn: - eq_( - conn.execute(table.select()).fetchall(), + eq_( + conn.execute(table.select()).fetchall(), + [ + (30, "d1"), + (1, "d2"), + (31, "d3"), + (32, "d4"), + (2, "d5"), + (3, "d6"), + (33, "d7"), + (4, "d8"), + ], + ) + + def _assert_data_with_sequence_returning(self, connection, table, seqname): + """ + invoked by: + * test_sequence_returning_insert + """ + + with self.sql_execution_asserter(connection) as asserter: + conn = connection + conn.execute(table.insert(), {"id": 30, "data": "d1"}) + conn.execute(table.insert(), {"data": "d2"}) + conn.execute( + table.insert(), [ - (30, "d1"), - (1, "d2"), - (31, "d3"), - (32, "d4"), - (2, "d5"), - (3, "d6"), - (33, "d7"), - (4, "d8"), + {"id": 31, "data": "d3"}, + {"id": 32, "data": "d4"}, ], ) - - # cant test reflection here since the Sequence must be - # explicitly specified - - def _assert_data_with_sequence_returning(self, table, seqname): - engine = engines.testing_engine(options={"implicit_returning": True}) - - with self.sql_execution_asserter(engine) as asserter: - with engine.begin() as conn: - conn.execute(table.insert(), {"id": 30, "data": "d1"}) - conn.execute(table.insert(), {"data": "d2"}) - conn.execute( - table.insert(), - [ - {"id": 31, "data": "d3"}, - {"id": 32, "data": "d4"}, - ], - ) - conn.execute(table.insert(), [{"data": "d5"}, {"data": "d6"}]) - conn.execute(table.insert().inline(), {"id": 33, "data": "d7"}) - conn.execute(table.insert().inline(), {"data": "d8"}) + conn.execute(table.insert(), [{"data": "d5"}, {"data": "d6"}]) + conn.execute(table.insert().inline(), {"id": 33, "data": "d7"}) + conn.execute(table.insert().inline(), {"data": "d8"}) asserter.assert_( DialectSQL( @@ -568,35 +560,34 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults): ), ) - with engine.begin() as conn: - eq_( - conn.execute(table.select()).fetchall(), - [ - (30, "d1"), - (1, "d2"), - (31, "d3"), - (32, "d4"), - (2, "d5"), - (3, "d6"), - (33, "d7"), - (4, "d8"), - ], - ) - - # cant test reflection here since the Sequence must be - # explicitly specified + eq_( + connection.execute(table.select()).fetchall(), + [ + (30, "d1"), + (1, "d2"), + (31, "d3"), + (32, "d4"), + (2, "d5"), + (3, "d6"), + (33, "d7"), + (4, "d8"), + ], + ) - def _assert_data_noautoincrement(self, table): - engine = engines.testing_engine(options={"implicit_returning": False}) + def _assert_data_noautoincrement(self, connection, table): + """ + invoked by: + * test_noautoincrement_insert + """ # turning off the cache because we are checking for compile-time # warnings - engine = engine.execution_options(compiled_cache=None) + connection.execution_options(compiled_cache=None) - with engine.begin() as conn: - conn.execute(table.insert(), {"id": 30, "data": "d1"}) + conn = connection + conn.execute(table.insert(), {"id": 30, "data": "d1"}) - with engine.begin() as conn: + with conn.begin_nested() as nested: with expect_warnings( ".*has no Python-side or server-side default.*" ): @@ -606,8 +597,9 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults): table.insert(), {"data": "d2"}, ) + nested.rollback() - with engine.begin() as conn: + with conn.begin_nested() as nested: with expect_warnings( ".*has no Python-side or server-side default.*" ): @@ -617,8 +609,9 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults): table.insert(), [{"data": "d2"}, {"data": "d3"}], ) + nested.rollback() - with engine.begin() as conn: + with conn.begin_nested() as nested: with expect_warnings( ".*has no Python-side or server-side default.*" ): @@ -628,8 +621,9 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults): table.insert(), {"data": "d2"}, ) + nested.rollback() - with engine.begin() as conn: + with conn.begin_nested() as nested: with expect_warnings( ".*has no Python-side or server-side default.*" ): @@ -639,28 +633,29 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults): table.insert(), [{"data": "d2"}, {"data": "d3"}], ) + nested.rollback() - with engine.begin() as conn: - conn.execute( - table.insert(), - [{"id": 31, "data": "d2"}, {"id": 32, "data": "d3"}], - ) - conn.execute(table.insert().inline(), {"id": 33, "data": "d4"}) - eq_( - conn.execute(table.select()).fetchall(), - [(30, "d1"), (31, "d2"), (32, "d3"), (33, "d4")], - ) - conn.execute(table.delete()) + conn.execute( + table.insert(), + [{"id": 31, "data": "d2"}, {"id": 32, "data": "d3"}], + ) + conn.execute(table.insert().inline(), {"id": 33, "data": "d4"}) + eq_( + conn.execute(table.select()).fetchall(), + [(30, "d1"), (31, "d2"), (32, "d3"), (33, "d4")], + ) + conn.execute(table.delete()) # test the same series of events using a reflected version of # the table m2 = MetaData() - table = Table(table.name, m2, autoload_with=engine) - with engine.begin() as conn: - conn.execute(table.insert(), {"id": 30, "data": "d1"}) + table = Table(table.name, m2, autoload_with=connection) + conn = connection + + conn.execute(table.insert(), {"id": 30, "data": "d1"}) - with engine.begin() as conn: + with conn.begin_nested() as nested: with expect_warnings( ".*has no Python-side or server-side default.*" ): @@ -670,8 +665,9 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults): table.insert(), {"data": "d2"}, ) + nested.rollback() - with engine.begin() as conn: + with conn.begin_nested() as nested: with expect_warnings( ".*has no Python-side or server-side default.*" ): @@ -681,17 +677,17 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults): table.insert(), [{"data": "d2"}, {"data": "d3"}], ) + nested.rollback() - with engine.begin() as conn: - conn.execute( - table.insert(), - [{"id": 31, "data": "d2"}, {"id": 32, "data": "d3"}], - ) - conn.execute(table.insert().inline(), {"id": 33, "data": "d4"}) - eq_( - conn.execute(table.select()).fetchall(), - [(30, "d1"), (31, "d2"), (32, "d3"), (33, "d4")], - ) + conn.execute( + table.insert(), + [{"id": 31, "data": "d2"}, {"id": 32, "data": "d3"}], + ) + conn.execute(table.insert().inline(), {"id": 33, "data": "d4"}) + eq_( + conn.execute(table.select()).fetchall(), + [(30, "d1"), (31, "d2"), (32, "d3"), (33, "d4")], + ) class MatchTest(fixtures.TablesTest, AssertsCompiledSQL): |