From f472405e598ad3088308db8bfef2879a57b26025 Mon Sep 17 00:00:00 2001 From: Mike Bayer Date: Wed, 21 Dec 2016 10:16:31 -0500 Subject: Don't select lastrowid for inline=True - Fixed bug where SQL Server dialects would attempt to select the last row identity for an INSERT from SELECT, failing in the case when the SELECT has no rows. For such a statement, the inline flag is set to True indicating no last primary key should be fetched. Change-Id: Ic40d56d9eadadc3024a4d71245f9eed4c420024a Fixes: #3876 --- lib/sqlalchemy/testing/suite/test_insert.py | 50 +++++++++++++++++++++++++++++ 1 file changed, 50 insertions(+) (limited to 'lib/sqlalchemy/testing') diff --git a/lib/sqlalchemy/testing/suite/test_insert.py b/lib/sqlalchemy/testing/suite/test_insert.py index 70e8a6b17..c0b6b18eb 100644 --- a/lib/sqlalchemy/testing/suite/test_insert.py +++ b/lib/sqlalchemy/testing/suite/test_insert.py @@ -140,6 +140,56 @@ class InsertBehaviorTest(fixtures.TablesTest): 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 -- cgit v1.2.1