From: Mike Bayer Date: Sat, 1 Dec 2018 19:28:57 +0000 (-0500) Subject: Move CRUDTest, InlineDefaultTest from test_compiler X-Git-Tag: rel_1_3_0b2~74^2 X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=87cdda008673e01e2c32049f103e9cdebd2a5d77;p=thirdparty%2Fsqlalchemy%2Fsqlalchemy.git Move CRUDTest, InlineDefaultTest from test_compiler test_compiler is mostly related to SELECT statements as well as smaller SQL elements. While it still has some DDL related tests, move out all the remaining insert/update tests into the already present test_insert.py, test_update.py Fixes: #2630 Change-Id: I4167618543fd1235d12d1717c8c629d2374b325a --- diff --git a/test/dialect/mssql/test_compiler.py b/test/dialect/mssql/test_compiler.py index 0a79b3e598..70b9a6c901 100644 --- a/test/dialect/mssql/test_compiler.py +++ b/test/dialect/mssql/test_compiler.py @@ -170,6 +170,35 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): ) self.assert_compile(sql.delete(a1), "DELETE FROM t1 AS a1") + def test_update_from(self): + metadata = MetaData() + table1 = Table( + 'mytable', metadata, + Column('myid', Integer), + Column('name', String(30)), + Column('description', String(50))) + table2 = Table( + 'myothertable', metadata, + Column('otherid', Integer), + Column('othername', String(30))) + + mt = table1.alias() + + u = table1.update().values(name='foo')\ + .where(table2.c.otherid == table1.c.myid) + + # testing mssql.base.MSSQLCompiler.update_from_clause + self.assert_compile(u, + "UPDATE mytable SET name=:name " + "FROM mytable, myothertable WHERE " + "myothertable.otherid = mytable.myid") + + self.assert_compile(u.where(table2.c.othername == mt.c.name), + "UPDATE mytable SET name=:name " + "FROM mytable, myothertable, mytable AS mytable_1 " + "WHERE myothertable.otherid = mytable.myid " + "AND myothertable.othername = mytable_1.name") + def test_update_from_hint(self): t = table('sometable', column('somecolumn')) t2 = table('othertable', column('somecolumn')) diff --git a/test/sql/test_compiler.py b/test/sql/test_compiler.py index 22b46cc46b..f543b86773 100644 --- a/test/sql/test_compiler.py +++ b/test/sql/test_compiler.py @@ -3037,221 +3037,6 @@ class ExecutionOptionsTest(fixtures.TestBase): eq_(compiled.execution_options, {"autocommit": False}) -class CRUDTest(fixtures.TestBase, AssertsCompiledSQL): - __dialect__ = 'default_enhanced' - - def test_insert_literal_binds(self): - stmt = table1.insert().values(myid=3, name='jack') - - self.assert_compile( - stmt, - "INSERT INTO mytable (myid, name) VALUES (3, 'jack')", - literal_binds=True) - - def test_insert_literal_binds_sequence_notimplemented(self): - table = Table('x', MetaData(), Column('y', Integer, Sequence('y_seq'))) - dialect = default.DefaultDialect() - dialect.supports_sequences = True - - stmt = table.insert().values(myid=3, name='jack') - - assert_raises( - NotImplementedError, - stmt.compile, - compile_kwargs=dict(literal_binds=True), dialect=dialect - ) - - def test_update_literal_binds(self): - stmt = table1.update().values(name='jack').\ - where(table1.c.name == 'jill') - - self.assert_compile( - stmt, - "UPDATE mytable SET name='jack' WHERE mytable.name = 'jill'", - literal_binds=True) - - def test_delete_literal_binds(self): - stmt = table1.delete().where(table1.c.name == 'jill') - - self.assert_compile( - stmt, - "DELETE FROM mytable WHERE mytable.name = 'jill'", - literal_binds=True) - - def test_correlated_update(self): - # test against a straight text subquery - u = update( - table1, - values={ - table1.c.name: - text("(select name from mytable where id=mytable.id)") - } - ) - self.assert_compile( - u, - "UPDATE mytable SET name=(select name from mytable " - "where id=mytable.id)") - - mt = table1.alias() - u = update(table1, values={ - table1.c.name: - select([mt.c.name], mt.c.myid == table1.c.myid) - }) - self.assert_compile( - u, "UPDATE mytable SET name=(SELECT mytable_1.name FROM " - "mytable AS mytable_1 WHERE " - "mytable_1.myid = mytable.myid)") - - # test against a regular constructed subquery - s = select([table2], table2.c.otherid == table1.c.myid) - u = update(table1, table1.c.name == 'jack', values={table1.c.name: s}) - self.assert_compile( - u, "UPDATE mytable SET name=(SELECT myothertable.otherid, " - "myothertable.othername FROM myothertable WHERE " - "myothertable.otherid = mytable.myid) " - "WHERE mytable.name = :name_1") - - # test a non-correlated WHERE clause - s = select([table2.c.othername], table2.c.otherid == 7) - u = update(table1, table1.c.name == s) - self.assert_compile(u, - "UPDATE mytable SET myid=:myid, name=:name, " - "description=:description WHERE mytable.name = " - "(SELECT myothertable.othername FROM myothertable " - "WHERE myothertable.otherid = :otherid_1)") - - # test one that is actually correlated... - s = select([table2.c.othername], table2.c.otherid == table1.c.myid) - u = table1.update(table1.c.name == s) - self.assert_compile(u, - "UPDATE mytable SET myid=:myid, name=:name, " - "description=:description WHERE mytable.name = " - "(SELECT myothertable.othername FROM myothertable " - "WHERE myothertable.otherid = mytable.myid)") - - # test correlated FROM implicit in WHERE and SET clauses - u = table1.update().values(name=table2.c.othername)\ - .where(table2.c.otherid == table1.c.myid) - self.assert_compile( - u, "UPDATE mytable SET name=myothertable.othername " - "FROM myothertable WHERE myothertable.otherid = mytable.myid") - u = table1.update().values(name='foo')\ - .where(table2.c.otherid == table1.c.myid) - self.assert_compile( - u, "UPDATE mytable SET name=:name " - "FROM myothertable WHERE myothertable.otherid = mytable.myid") - - self.assert_compile(u, - "UPDATE mytable SET name=:name " - "FROM mytable, myothertable WHERE " - "myothertable.otherid = mytable.myid", - dialect=mssql.dialect()) - - self.assert_compile(u.where(table2.c.othername == mt.c.name), - "UPDATE mytable SET name=:name " - "FROM mytable, myothertable, mytable AS mytable_1 " - "WHERE myothertable.otherid = mytable.myid " - "AND myothertable.othername = mytable_1.name", - dialect=mssql.dialect()) - - def test_binds_that_match_columns(self): - """test bind params named after column names - replace the normal SET/VALUES generation.""" - - t = table('foo', column('x'), column('y')) - - u = t.update().where(t.c.x == bindparam('x')) - - assert_raises(exc.CompileError, u.compile) - - self.assert_compile(u, "UPDATE foo SET WHERE foo.x = :x", params={}) - - assert_raises(exc.CompileError, u.values(x=7).compile) - - self.assert_compile(u.values(y=7), - "UPDATE foo SET y=:y WHERE foo.x = :x") - - assert_raises(exc.CompileError, - u.values(x=7).compile, column_keys=['x', 'y']) - assert_raises(exc.CompileError, u.compile, column_keys=['x', 'y']) - - self.assert_compile( - u.values( - x=3 + - bindparam('x')), - "UPDATE foo SET x=(:param_1 + :x) WHERE foo.x = :x") - - self.assert_compile( - u.values( - x=3 + - bindparam('x')), - "UPDATE foo SET x=(:param_1 + :x) WHERE foo.x = :x", - params={ - 'x': 1}) - - self.assert_compile( - u.values( - x=3 + - bindparam('x')), - "UPDATE foo SET x=(:param_1 + :x), y=:y WHERE foo.x = :x", - params={ - 'x': 1, - 'y': 2}) - - i = t.insert().values(x=3 + bindparam('x')) - self.assert_compile(i, - "INSERT INTO foo (x) VALUES ((:param_1 + :x))") - self.assert_compile( - i, - "INSERT INTO foo (x, y) VALUES ((:param_1 + :x), :y)", - params={ - 'x': 1, - 'y': 2}) - - i = t.insert().values(x=bindparam('y')) - self.assert_compile(i, "INSERT INTO foo (x) VALUES (:y)") - - i = t.insert().values(x=bindparam('y'), y=5) - assert_raises(exc.CompileError, i.compile) - - i = t.insert().values(x=3 + bindparam('y'), y=5) - assert_raises(exc.CompileError, i.compile) - - i = t.insert().values(x=3 + bindparam('x2')) - self.assert_compile(i, - "INSERT INTO foo (x) VALUES ((:param_1 + :x2))") - self.assert_compile( - i, - "INSERT INTO foo (x) VALUES ((:param_1 + :x2))", - params={}) - self.assert_compile( - i, - "INSERT INTO foo (x, y) VALUES ((:param_1 + :x2), :y)", - params={ - 'x': 1, - 'y': 2}) - self.assert_compile( - i, - "INSERT INTO foo (x, y) VALUES ((:param_1 + :x2), :y)", - params={ - 'x2': 1, - 'y': 2}) - - def test_labels_no_collision(self): - - t = table('foo', column('id'), column('foo_id')) - - self.assert_compile( - t.update().where(t.c.id == 5), - "UPDATE foo SET id=:id, foo_id=:foo_id WHERE foo.id = :id_1" - ) - - self.assert_compile( - t.update().where(t.c.id == bindparam(key=t.c.id._label)), - "UPDATE foo SET id=:id, foo_id=:foo_id WHERE foo.id = :foo_id_1" - ) - - class DDLTest(fixtures.TestBase, AssertsCompiledSQL): __dialect__ = 'default' @@ -3425,45 +3210,6 @@ class DDLTest(fixtures.TestBase, AssertsCompiledSQL): ) -class InlineDefaultTest(fixtures.TestBase, AssertsCompiledSQL): - __dialect__ = 'default' - - def test_insert(self): - m = MetaData() - foo = Table('foo', m, - Column('id', Integer)) - - t = Table('test', m, - Column('col1', Integer, default=func.foo(1)), - Column('col2', Integer, default=select( - [func.coalesce(func.max(foo.c.id))])), - ) - - self.assert_compile( - t.insert( - inline=True, values={}), - "INSERT INTO test (col1, col2) VALUES (foo(:foo_1), " - "(SELECT coalesce(max(foo.id)) AS coalesce_1 FROM " - "foo))") - - def test_update(self): - m = MetaData() - foo = Table('foo', m, - Column('id', Integer)) - - t = Table('test', m, - Column('col1', Integer, onupdate=func.foo(1)), - Column('col2', Integer, onupdate=select( - [func.coalesce(func.max(foo.c.id))])), - Column('col3', String(30)) - ) - - self.assert_compile(t.update(inline=True, values={'col3': 'foo'}), - "UPDATE test SET col1=foo(:foo_1), col2=(SELECT " - "coalesce(max(foo.id)) AS coalesce_1 FROM foo), " - "col3=:col3") - - class SchemaTest(fixtures.TestBase, AssertsCompiledSQL): __dialect__ = 'default' @@ -3981,7 +3727,6 @@ class CorrelateTest(fixtures.TestBase, AssertsCompiledSQL): # multilevel FROM clauses behaves like 0.8.1 (i.e. doesn't happen) t1 = table('t1', column('x')) t2 = table('t2', column('y')) - t3 = table('t3', column('z')) s = select([t1.c.x]).where(t1.c.x == t2.c.y) s2 = select([t2, s]) diff --git a/test/sql/test_delete.py b/test/sql/test_delete.py index 91f2c2cdc2..331a536018 100644 --- a/test/sql/test_delete.py +++ b/test/sql/test_delete.py @@ -27,6 +27,16 @@ class _DeleteTestBase(object): class DeleteTest(_DeleteTestBase, fixtures.TablesTest, AssertsCompiledSQL): __dialect__ = 'default' + def test_delete_literal_binds(self): + table1 = self.tables.mytable + + stmt = table1.delete().where(table1.c.name == 'jill') + + self.assert_compile( + stmt, + "DELETE FROM mytable WHERE mytable.name = 'jill'", + literal_binds=True) + def test_delete(self): table1 = self.tables.mytable diff --git a/test/sql/test_insert.py b/test/sql/test_insert.py index 6ea5b4f37c..729c420c00 100644 --- a/test/sql/test_insert.py +++ b/test/sql/test_insert.py @@ -6,7 +6,7 @@ from sqlalchemy import Column, Integer, MetaData, String, Table,\ from sqlalchemy.dialects import mysql, postgresql from sqlalchemy.engine import default from sqlalchemy.testing import AssertsCompiledSQL,\ - assert_raises_message, fixtures, eq_, expect_warnings + assert_raises_message, fixtures, eq_, expect_warnings, assert_raises from sqlalchemy.sql import crud @@ -31,6 +31,91 @@ class _InsertTestBase(object): class InsertTest(_InsertTestBase, fixtures.TablesTest, AssertsCompiledSQL): __dialect__ = 'default' + def test_binds_that_match_columns(self): + """test bind params named after column names + replace the normal SET/VALUES generation.""" + + t = table('foo', column('x'), column('y')) + + i = t.insert().values(x=3 + bindparam('x')) + self.assert_compile(i, + "INSERT INTO foo (x) VALUES ((:param_1 + :x))") + self.assert_compile( + i, + "INSERT INTO foo (x, y) VALUES ((:param_1 + :x), :y)", + params={ + 'x': 1, + 'y': 2}) + + i = t.insert().values(x=bindparam('y')) + self.assert_compile(i, "INSERT INTO foo (x) VALUES (:y)") + + i = t.insert().values(x=bindparam('y'), y=5) + assert_raises(exc.CompileError, i.compile) + + i = t.insert().values(x=3 + bindparam('y'), y=5) + assert_raises(exc.CompileError, i.compile) + + i = t.insert().values(x=3 + bindparam('x2')) + self.assert_compile(i, + "INSERT INTO foo (x) VALUES ((:param_1 + :x2))") + self.assert_compile( + i, + "INSERT INTO foo (x) VALUES ((:param_1 + :x2))", + params={}) + self.assert_compile( + i, + "INSERT INTO foo (x, y) VALUES ((:param_1 + :x2), :y)", + params={ + 'x': 1, + 'y': 2}) + self.assert_compile( + i, + "INSERT INTO foo (x, y) VALUES ((:param_1 + :x2), :y)", + params={ + 'x2': 1, + 'y': 2}) + + def test_insert_literal_binds(self): + table1 = self.tables.mytable + stmt = table1.insert().values(myid=3, name='jack') + + self.assert_compile( + stmt, + "INSERT INTO mytable (myid, name) VALUES (3, 'jack')", + literal_binds=True) + + def test_insert_literal_binds_sequence_notimplemented(self): + table = Table('x', MetaData(), Column('y', Integer, Sequence('y_seq'))) + dialect = default.DefaultDialect() + dialect.supports_sequences = True + + stmt = table.insert().values(myid=3, name='jack') + + assert_raises( + NotImplementedError, + stmt.compile, + compile_kwargs=dict(literal_binds=True), dialect=dialect + ) + + def test_inline_defaults(self): + m = MetaData() + foo = Table('foo', m, + Column('id', Integer)) + + t = Table('test', m, + Column('col1', Integer, default=func.foo(1)), + Column('col2', Integer, default=select( + [func.coalesce(func.max(foo.c.id))])), + ) + + self.assert_compile( + t.insert( + inline=True, values={}), + "INSERT INTO test (col1, col2) VALUES (foo(:foo_1), " + "(SELECT coalesce(max(foo.id)) AS coalesce_1 FROM " + "foo))") + def test_generic_insert_bind_params_all_columns(self): table1 = self.tables.mytable diff --git a/test/sql/test_update.py b/test/sql/test_update.py index 138581061b..56d12d9272 100644 --- a/test/sql/test_update.py +++ b/test/sql/test_update.py @@ -1,10 +1,11 @@ from sqlalchemy import Integer, String, ForeignKey, and_, or_, func, \ - literal, update, table, bindparam, column, select, exc, exists + literal, update, table, bindparam, column, select, exc, exists, text, \ + MetaData from sqlalchemy import testing from sqlalchemy.dialects import mysql from sqlalchemy.engine import default from sqlalchemy.testing import AssertsCompiledSQL, eq_, fixtures, \ - assert_raises_message + assert_raises_message, assert_raises from sqlalchemy.testing.schema import Table, Column from sqlalchemy import util @@ -68,7 +69,188 @@ class _UpdateFromTestBase(object): class UpdateTest(_UpdateFromTestBase, fixtures.TablesTest, AssertsCompiledSQL): - __dialect__ = 'default' + __dialect__ = 'default_enhanced' + + def test_update_literal_binds(self): + table1 = self.tables.mytable + + table1 = self.tables.mytable + + stmt = table1.update().values(name='jack').\ + where(table1.c.name == 'jill') + + self.assert_compile( + stmt, + "UPDATE mytable SET name='jack' WHERE mytable.name = 'jill'", + literal_binds=True) + + def test_correlated_update_one(self): + table1 = self.tables.mytable + + # test against a straight text subquery + u = update( + table1, + values={ + table1.c.name: + text("(select name from mytable where id=mytable.id)") + } + ) + self.assert_compile( + u, + "UPDATE mytable SET name=(select name from mytable " + "where id=mytable.id)") + + def test_correlated_update_two(self): + table1 = self.tables.mytable + + mt = table1.alias() + u = update(table1, values={ + table1.c.name: + select([mt.c.name], mt.c.myid == table1.c.myid) + }) + self.assert_compile( + u, "UPDATE mytable SET name=(SELECT mytable_1.name FROM " + "mytable AS mytable_1 WHERE " + "mytable_1.myid = mytable.myid)") + + def test_correlated_update_three(self): + table1 = self.tables.mytable + table2 = self.tables.myothertable + + # test against a regular constructed subquery + s = select([table2], table2.c.otherid == table1.c.myid) + u = update(table1, table1.c.name == 'jack', values={table1.c.name: s}) + self.assert_compile( + u, "UPDATE mytable SET name=(SELECT myothertable.otherid, " + "myothertable.othername FROM myothertable WHERE " + "myothertable.otherid = mytable.myid) " + "WHERE mytable.name = :name_1") + + def test_correlated_update_four(self): + table1 = self.tables.mytable + table2 = self.tables.myothertable + + # test a non-correlated WHERE clause + s = select([table2.c.othername], table2.c.otherid == 7) + u = update(table1, table1.c.name == s) + self.assert_compile(u, + "UPDATE mytable SET myid=:myid, name=:name, " + "description=:description WHERE mytable.name = " + "(SELECT myothertable.othername FROM myothertable " + "WHERE myothertable.otherid = :otherid_1)") + + def test_correlated_update_five(self): + table1 = self.tables.mytable + table2 = self.tables.myothertable + + # test one that is actually correlated... + s = select([table2.c.othername], table2.c.otherid == table1.c.myid) + u = table1.update(table1.c.name == s) + self.assert_compile(u, + "UPDATE mytable SET myid=:myid, name=:name, " + "description=:description WHERE mytable.name = " + "(SELECT myothertable.othername FROM myothertable " + "WHERE myothertable.otherid = mytable.myid)") + + def test_correlated_update_six(self): + table1 = self.tables.mytable + table2 = self.tables.myothertable + + # test correlated FROM implicit in WHERE and SET clauses + u = table1.update().values(name=table2.c.othername)\ + .where(table2.c.otherid == table1.c.myid) + self.assert_compile( + u, "UPDATE mytable SET name=myothertable.othername " + "FROM myothertable WHERE myothertable.otherid = mytable.myid") + + def test_correlated_update_seven(self): + table1 = self.tables.mytable + table2 = self.tables.myothertable + + u = table1.update().values(name='foo')\ + .where(table2.c.otherid == table1.c.myid) + + # this is the "default_enhanced" compiler. there's no UPDATE FROM + # in the base compiler. + # See also test/dialect/mssql/test_compiler->test_update_from(). + self.assert_compile( + u, "UPDATE mytable SET name=:name " + "FROM myothertable WHERE myothertable.otherid = mytable.myid") + + def test_binds_that_match_columns(self): + """test bind params named after column names + replace the normal SET/VALUES generation.""" + + t = table('foo', column('x'), column('y')) + + u = t.update().where(t.c.x == bindparam('x')) + + assert_raises(exc.CompileError, u.compile) + + self.assert_compile(u, "UPDATE foo SET WHERE foo.x = :x", params={}) + + assert_raises(exc.CompileError, u.values(x=7).compile) + + self.assert_compile(u.values(y=7), + "UPDATE foo SET y=:y WHERE foo.x = :x") + + assert_raises(exc.CompileError, + u.values(x=7).compile, column_keys=['x', 'y']) + assert_raises(exc.CompileError, u.compile, column_keys=['x', 'y']) + + self.assert_compile( + u.values( + x=3 + + bindparam('x')), + "UPDATE foo SET x=(:param_1 + :x) WHERE foo.x = :x") + + self.assert_compile( + u.values( + x=3 + + bindparam('x')), + "UPDATE foo SET x=(:param_1 + :x) WHERE foo.x = :x", + params={ + 'x': 1}) + + self.assert_compile( + u.values( + x=3 + + bindparam('x')), + "UPDATE foo SET x=(:param_1 + :x), y=:y WHERE foo.x = :x", + params={ + 'x': 1, + 'y': 2}) + + def test_labels_no_collision(self): + + t = table('foo', column('id'), column('foo_id')) + + self.assert_compile( + t.update().where(t.c.id == 5), + "UPDATE foo SET id=:id, foo_id=:foo_id WHERE foo.id = :id_1" + ) + + self.assert_compile( + t.update().where(t.c.id == bindparam(key=t.c.id._label)), + "UPDATE foo SET id=:id, foo_id=:foo_id WHERE foo.id = :foo_id_1" + ) + + def test_inline_defaults(self): + m = MetaData() + foo = Table('foo', m, + Column('id', Integer)) + + t = Table('test', m, + Column('col1', Integer, onupdate=func.foo(1)), + Column('col2', Integer, onupdate=select( + [func.coalesce(func.max(foo.c.id))])), + Column('col3', String(30)) + ) + + self.assert_compile(t.update(inline=True, values={'col3': 'foo'}), + "UPDATE test SET col1=foo(:foo_1), col2=(SELECT " + "coalesce(max(foo.id)) AS coalesce_1 FROM foo), " + "col3=:col3") def test_update_1(self): table1 = self.tables.mytable