diff options
Diffstat (limited to 'test/dialect/postgresql/test_compiler.py')
-rw-r--r-- | test/dialect/postgresql/test_compiler.py | 88 |
1 files changed, 75 insertions, 13 deletions
diff --git a/test/dialect/postgresql/test_compiler.py b/test/dialect/postgresql/test_compiler.py index 0fe5f7066..a4c6cc087 100644 --- a/test/dialect/postgresql/test_compiler.py +++ b/test/dialect/postgresql/test_compiler.py @@ -2265,41 +2265,103 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): ) -class InsertOnConflictTest(fixtures.TestBase, AssertsCompiledSQL): +class InsertOnConflictTest(fixtures.TablesTest, AssertsCompiledSQL): __dialect__ = postgresql.dialect() - def setup_test(self): - self.table1 = table1 = table( + run_create_tables = None + + @classmethod + def define_tables(cls, metadata): + cls.table1 = table1 = table( "mytable", column("myid", Integer), column("name", String(128)), column("description", String(128)), ) - md = MetaData() - self.table_with_metadata = Table( + cls.table_with_metadata = Table( "mytable", - md, + metadata, Column("myid", Integer, primary_key=True), Column("name", String(128)), Column("description", String(128)), ) - self.unique_constr = schema.UniqueConstraint( + cls.unique_constr = schema.UniqueConstraint( table1.c.name, name="uq_name" ) - self.excl_constr = ExcludeConstraint( + cls.excl_constr = ExcludeConstraint( (table1.c.name, "="), (table1.c.description, "&&"), name="excl_thing", ) - self.excl_constr_anon = ExcludeConstraint( - (self.table_with_metadata.c.name, "="), - (self.table_with_metadata.c.description, "&&"), - where=self.table_with_metadata.c.description != "foo", + cls.excl_constr_anon = ExcludeConstraint( + (cls.table_with_metadata.c.name, "="), + (cls.table_with_metadata.c.description, "&&"), + where=cls.table_with_metadata.c.description != "foo", ) - self.goofy_index = Index( + cls.goofy_index = Index( "goofy_index", table1.c.name, postgresql_where=table1.c.name > "m" ) + Table( + "users", + metadata, + Column("id", Integer, primary_key=True), + Column("name", String(50)), + ) + + Table( + "users_w_key", + metadata, + Column("id", Integer, primary_key=True), + Column("name", String(50), key="name_keyed"), + ) + + @testing.combinations("control", "excluded", "dict") + def test_set_excluded(self, scenario): + """test #8014, sending all of .excluded to set""" + + if scenario == "control": + users = self.tables.users + + stmt = insert(users) + self.assert_compile( + stmt.on_conflict_do_update( + constraint=users.primary_key, set_=stmt.excluded + ), + "INSERT INTO users (id, name) VALUES (%(id)s, %(name)s) ON " + "CONFLICT (id) DO UPDATE " + "SET id = excluded.id, name = excluded.name", + ) + else: + users_w_key = self.tables.users_w_key + + stmt = insert(users_w_key) + + if scenario == "excluded": + self.assert_compile( + stmt.on_conflict_do_update( + constraint=users_w_key.primary_key, set_=stmt.excluded + ), + "INSERT INTO users_w_key (id, name) " + "VALUES (%(id)s, %(name_keyed)s) ON " + "CONFLICT (id) DO UPDATE " + "SET id = excluded.id, name = excluded.name", + ) + else: + self.assert_compile( + stmt.on_conflict_do_update( + constraint=users_w_key.primary_key, + set_={ + "id": stmt.excluded.id, + "name_keyed": stmt.excluded.name_keyed, + }, + ), + "INSERT INTO users_w_key (id, name) " + "VALUES (%(id)s, %(name_keyed)s) ON " + "CONFLICT (id) DO UPDATE " + "SET id = excluded.id, name = excluded.name", + ) + def test_on_conflict_do_no_call_twice(self): users = self.table1 |