diff options
| author | Mike Bayer <mike_mp@zzzcomputing.com> | 2019-01-06 01:14:26 -0500 |
|---|---|---|
| committer | mike bayer <mike_mp@zzzcomputing.com> | 2019-01-06 17:34:50 +0000 |
| commit | 1e1a38e7801f410f244e4bbb44ec795ae152e04e (patch) | |
| tree | 28e725c5c8188bd0cfd133d1e268dbca9b524978 /test/ext/test_compiler.py | |
| parent | 404e69426b05a82d905cbb3ad33adafccddb00dd (diff) | |
| download | sqlalchemy-1e1a38e7801f410f244e4bbb44ec795ae152e04e.tar.gz | |
Run black -l 79 against all source files
This is a straight reformat run using black as is, with no edits
applied at all.
The black run will format code consistently, however in
some cases that are prevalent in SQLAlchemy code it produces
too-long lines. The too-long lines will be resolved in the
following commit that will resolve all remaining flake8 issues
including shadowed builtins, long lines, import order, unused
imports, duplicate imports, and docstring issues.
Change-Id: I7eda77fed3d8e73df84b3651fd6cfcfe858d4dc9
Diffstat (limited to 'test/ext/test_compiler.py')
| -rw-r--r-- | test/ext/test_compiler.py | 213 |
1 files changed, 98 insertions, 115 deletions
diff --git a/test/ext/test_compiler.py b/test/ext/test_compiler.py index c23d5f2ac..22ab1f163 100644 --- a/test/ext/test_compiler.py +++ b/test/ext/test_compiler.py @@ -1,8 +1,13 @@ from sqlalchemy import * from sqlalchemy.types import TypeEngine -from sqlalchemy.sql.expression import ClauseElement, ColumnClause,\ - FunctionElement, Select, \ - BindParameter, ColumnElement +from sqlalchemy.sql.expression import ( + ClauseElement, + ColumnClause, + FunctionElement, + Select, + BindParameter, + ColumnElement, +) from sqlalchemy.schema import DDLElement, CreateColumn, CreateTable from sqlalchemy.ext.compiler import compiles, deregister @@ -14,92 +19,87 @@ from sqlalchemy.testing import fixtures, AssertsCompiledSQL class UserDefinedTest(fixtures.TestBase, AssertsCompiledSQL): - __dialect__ = 'default' + __dialect__ = "default" def test_column(self): - class MyThingy(ColumnClause): def __init__(self, arg=None): - super(MyThingy, self).__init__(arg or 'MYTHINGY!') + super(MyThingy, self).__init__(arg or "MYTHINGY!") @compiles(MyThingy) def visit_thingy(thingy, compiler, **kw): return ">>%s<<" % thingy.name self.assert_compile( - select([column('foo'), MyThingy()]), - "SELECT foo, >>MYTHINGY!<<" + select([column("foo"), MyThingy()]), "SELECT foo, >>MYTHINGY!<<" ) self.assert_compile( - select([MyThingy('x'), MyThingy('y')]).where(MyThingy() == 5), - "SELECT >>x<<, >>y<< WHERE >>MYTHINGY!<< = :MYTHINGY!_1" + select([MyThingy("x"), MyThingy("y")]).where(MyThingy() == 5), + "SELECT >>x<<, >>y<< WHERE >>MYTHINGY!<< = :MYTHINGY!_1", ) def test_create_column_skip(self): @compiles(CreateColumn) def skip_xmin(element, compiler, **kw): - if element.element.name == 'xmin': + if element.element.name == "xmin": return None else: return compiler.visit_create_column(element, **kw) - t = Table('t', MetaData(), Column('a', Integer), - Column('xmin', Integer), - Column('c', Integer)) + t = Table( + "t", + MetaData(), + Column("a", Integer), + Column("xmin", Integer), + Column("c", Integer), + ) self.assert_compile( - CreateTable(t), - "CREATE TABLE t (a INTEGER, c INTEGER)" + CreateTable(t), "CREATE TABLE t (a INTEGER, c INTEGER)" ) def test_types(self): class MyType(TypeEngine): pass - @compiles(MyType, 'sqlite') + @compiles(MyType, "sqlite") def visit_type(type, compiler, **kw): return "SQLITE_FOO" - @compiles(MyType, 'postgresql') + @compiles(MyType, "postgresql") def visit_type(type, compiler, **kw): return "POSTGRES_FOO" from sqlalchemy.dialects.sqlite import base as sqlite from sqlalchemy.dialects.postgresql import base as postgresql - self.assert_compile( - MyType(), - "SQLITE_FOO", - dialect=sqlite.dialect() - ) + self.assert_compile(MyType(), "SQLITE_FOO", dialect=sqlite.dialect()) self.assert_compile( - MyType(), - "POSTGRES_FOO", - dialect=postgresql.dialect() + MyType(), "POSTGRES_FOO", dialect=postgresql.dialect() ) def test_stateful(self): class MyThingy(ColumnClause): def __init__(self): - super(MyThingy, self).__init__('MYTHINGY!') + super(MyThingy, self).__init__("MYTHINGY!") @compiles(MyThingy) def visit_thingy(thingy, compiler, **kw): - if not hasattr(compiler, 'counter'): + if not hasattr(compiler, "counter"): compiler.counter = 0 compiler.counter += 1 return str(compiler.counter) self.assert_compile( - select([column('foo'), MyThingy()]).order_by(desc(MyThingy())), - "SELECT foo, 1 ORDER BY 2 DESC" + select([column("foo"), MyThingy()]).order_by(desc(MyThingy())), + "SELECT foo, 1 ORDER BY 2 DESC", ) self.assert_compile( select([MyThingy(), MyThingy()]).where(MyThingy() == 5), - "SELECT 1, 2 WHERE 3 = :MYTHINGY!_1" + "SELECT 1, 2 WHERE 3 = :MYTHINGY!_1", ) def test_callout_to_compiler(self): @@ -112,34 +112,31 @@ class UserDefinedTest(fixtures.TestBase, AssertsCompiledSQL): def visit_insert_from_select(element, compiler, **kw): return "INSERT INTO %s (%s)" % ( compiler.process(element.table, asfrom=True), - compiler.process(element.select) + compiler.process(element.select), ) - t1 = table("mytable", column('x'), column('y'), column('z')) + t1 = table("mytable", column("x"), column("y"), column("z")) self.assert_compile( - InsertFromSelect( - t1, - select([t1]).where(t1.c.x > 5) - ), + InsertFromSelect(t1, select([t1]).where(t1.c.x > 5)), "INSERT INTO mytable (SELECT mytable.x, mytable.y, mytable.z " - "FROM mytable WHERE mytable.x > :x_1)" + "FROM mytable WHERE mytable.x > :x_1)", ) def test_no_default_but_has_a_visit(self): class MyThingy(ColumnClause): pass - @compiles(MyThingy, 'postgresql') + @compiles(MyThingy, "postgresql") def visit_thingy(thingy, compiler, **kw): return "mythingy" - eq_(str(MyThingy('x')), "x") + eq_(str(MyThingy("x")), "x") def test_no_default_has_no_visit(self): class MyThingy(TypeEngine): pass - @compiles(MyThingy, 'postgresql') + @compiles(MyThingy, "postgresql") def visit_thingy(thingy, compiler, **kw): return "mythingy" @@ -147,14 +144,15 @@ class UserDefinedTest(fixtures.TestBase, AssertsCompiledSQL): exc.CompileError, "<class 'test.ext.test_compiler..*MyThingy'> " "construct has no default compilation handler.", - str, MyThingy() + str, + MyThingy(), ) def test_no_default_message(self): class MyThingy(ClauseElement): pass - @compiles(MyThingy, 'postgresql') + @compiles(MyThingy, "postgresql") def visit_thingy(thingy, compiler, **kw): return "mythingy" @@ -162,7 +160,8 @@ class UserDefinedTest(fixtures.TestBase, AssertsCompiledSQL): exc.CompileError, "<class 'test.ext.test_compiler..*MyThingy'> " "construct has no default compilation handler.", - str, MyThingy() + str, + MyThingy(), ) def test_default_subclass(self): @@ -176,9 +175,7 @@ class UserDefinedTest(fixtures.TestBase, AssertsCompiledSQL): return "array" self.assert_compile( - MyArray(Integer), - "INTEGER[]", - dialect="postgresql" + MyArray(Integer), "INTEGER[]", dialect="postgresql" ) def test_annotations(self): @@ -187,35 +184,31 @@ class UserDefinedTest(fixtures.TestBase, AssertsCompiledSQL): """ - t1 = table('t1', column('c1'), column('c2')) + t1 = table("t1", column("c1"), column("c2")) dispatch = Select._compiler_dispatch try: + @compiles(Select) def compile(element, compiler, **kw): return "OVERRIDE" s1 = select([t1]) - self.assert_compile( - s1, "OVERRIDE" - ) - self.assert_compile( - s1._annotate({}), - "OVERRIDE" - ) + self.assert_compile(s1, "OVERRIDE") + self.assert_compile(s1._annotate({}), "OVERRIDE") finally: Select._compiler_dispatch = dispatch - if hasattr(Select, '_compiler_dispatcher'): + if hasattr(Select, "_compiler_dispatcher"): del Select._compiler_dispatcher def test_dialect_specific(self): class AddThingy(DDLElement): - __visit_name__ = 'add_thingy' + __visit_name__ = "add_thingy" class DropThingy(DDLElement): - __visit_name__ = 'drop_thingy' + __visit_name__ = "drop_thingy" - @compiles(AddThingy, 'sqlite') + @compiles(AddThingy, "sqlite") def visit_add_thingy(thingy, compiler, **kw): return "ADD SPECIAL SL THINGY" @@ -232,21 +225,22 @@ class UserDefinedTest(fixtures.TestBase, AssertsCompiledSQL): self.assert_compile(DropThingy(), "DROP THINGY") from sqlalchemy.dialects.sqlite import base - self.assert_compile(AddThingy(), - "ADD SPECIAL SL THINGY", - dialect=base.dialect()) - self.assert_compile(DropThingy(), - "DROP THINGY", - dialect=base.dialect()) + self.assert_compile( + AddThingy(), "ADD SPECIAL SL THINGY", dialect=base.dialect() + ) - @compiles(DropThingy, 'sqlite') + self.assert_compile( + DropThingy(), "DROP THINGY", dialect=base.dialect() + ) + + @compiles(DropThingy, "sqlite") def visit_drop_thingy(thingy, compiler, **kw): return "DROP SPECIAL SL THINGY" - self.assert_compile(DropThingy(), - "DROP SPECIAL SL THINGY", - dialect=base.dialect()) + self.assert_compile( + DropThingy(), "DROP SPECIAL SL THINGY", dialect=base.dialect() + ) self.assert_compile(DropThingy(), "DROP THINGY") @@ -260,19 +254,17 @@ class UserDefinedTest(fixtures.TestBase, AssertsCompiledSQL): def visit_myfunc(element, compiler, **kw): return "utcnow()" - @compiles(MyUtcFunction, 'postgresql') + @compiles(MyUtcFunction, "postgresql") def visit_myfunc(element, compiler, **kw): return "timezone('utc', current_timestamp)" self.assert_compile( - MyUtcFunction(), - "utcnow()", - use_default_dialect=True + MyUtcFunction(), "utcnow()", use_default_dialect=True ) self.assert_compile( MyUtcFunction(), "timezone('utc', current_timestamp)", - dialect=postgresql.dialect() + dialect=postgresql.dialect(), ) def test_function_calls_base(self): @@ -280,13 +272,13 @@ class UserDefinedTest(fixtures.TestBase, AssertsCompiledSQL): class greatest(FunctionElement): type = Numeric() - name = 'greatest' + name = "greatest" @compiles(greatest) def default_greatest(element, compiler, **kw): return compiler.visit_function(element) - @compiles(greatest, 'mssql') + @compiles(greatest, "mssql") def case_greatest(element, compiler, **kw): arg1, arg2 = list(element.clauses) return "CASE WHEN %s > %s THEN %s ELSE %s END" % ( @@ -297,26 +289,26 @@ class UserDefinedTest(fixtures.TestBase, AssertsCompiledSQL): ) self.assert_compile( - greatest('a', 'b'), - 'greatest(:greatest_1, :greatest_2)', - use_default_dialect=True + greatest("a", "b"), + "greatest(:greatest_1, :greatest_2)", + use_default_dialect=True, ) self.assert_compile( - greatest('a', 'b'), + greatest("a", "b"), "CASE WHEN :greatest_1 > :greatest_2 " "THEN :greatest_1 ELSE :greatest_2 END", - dialect=mssql.dialect() + dialect=mssql.dialect(), ) def test_subclasses_one(self): class Base(FunctionElement): - name = 'base' + name = "base" class Sub1(Base): - name = 'sub1' + name = "sub1" class Sub2(Base): - name = 'sub2' + name = "sub2" @compiles(Base) def visit_base(element, compiler, **kw): @@ -328,31 +320,31 @@ class UserDefinedTest(fixtures.TestBase, AssertsCompiledSQL): self.assert_compile( select([Sub1(), Sub2()]), - 'SELECT FOOsub1, sub2', - use_default_dialect=True + "SELECT FOOsub1, sub2", + use_default_dialect=True, ) def test_subclasses_two(self): class Base(FunctionElement): - name = 'base' + name = "base" class Sub1(Base): - name = 'sub1' + name = "sub1" @compiles(Base) def visit_base(element, compiler, **kw): return element.name class Sub2(Base): - name = 'sub2' + name = "sub2" class SubSub1(Sub1): - name = 'subsub1' + name = "subsub1" self.assert_compile( select([Sub1(), Sub2(), SubSub1()]), - 'SELECT sub1, sub2, subsub1', - use_default_dialect=True + "SELECT sub1, sub2, subsub1", + use_default_dialect=True, ) @compiles(Sub1) @@ -361,42 +353,36 @@ class UserDefinedTest(fixtures.TestBase, AssertsCompiledSQL): self.assert_compile( select([Sub1(), Sub2(), SubSub1()]), - 'SELECT FOOsub1, sub2, FOOsubsub1', - use_default_dialect=True + "SELECT FOOsub1, sub2, FOOsubsub1", + use_default_dialect=True, ) class DefaultOnExistingTest(fixtures.TestBase, AssertsCompiledSQL): """Test replacement of default compilation on existing constructs.""" - __dialect__ = 'default' + + __dialect__ = "default" def teardown(self): for cls in (Select, BindParameter): deregister(cls) def test_select(self): - t1 = table('t1', column('c1'), column('c2')) + t1 = table("t1", column("c1"), column("c2")) - @compiles(Select, 'sqlite') + @compiles(Select, "sqlite") def compile(element, compiler, **kw): return "OVERRIDE" s1 = select([t1]) - self.assert_compile( - s1, "SELECT t1.c1, t1.c2 FROM t1", - ) + self.assert_compile(s1, "SELECT t1.c1, t1.c2 FROM t1") from sqlalchemy.dialects.sqlite import base as sqlite - self.assert_compile( - s1, "OVERRIDE", - dialect=sqlite.dialect() - ) + + self.assert_compile(s1, "OVERRIDE", dialect=sqlite.dialect()) def test_binds_in_select(self): - t = table('t', - column('a'), - column('b'), - column('c')) + t = table("t", column("a"), column("b"), column("c")) @compiles(BindParameter) def gen_bind(element, compiler, **kw): @@ -405,14 +391,11 @@ class DefaultOnExistingTest(fixtures.TestBase, AssertsCompiledSQL): self.assert_compile( t.select().where(t.c.c == 5), "SELECT t.a, t.b, t.c FROM t WHERE t.c = BIND(:c_1)", - use_default_dialect=True + use_default_dialect=True, ) def test_binds_in_dml(self): - t = table('t', - column('a'), - column('b'), - column('c')) + t = table("t", column("a"), column("b"), column("c")) @compiles(BindParameter) def gen_bind(element, compiler, **kw): @@ -421,6 +404,6 @@ class DefaultOnExistingTest(fixtures.TestBase, AssertsCompiledSQL): self.assert_compile( t.insert(), "INSERT INTO t (a, b) VALUES (BIND(:a), BIND(:b))", - {'a': 1, 'b': 2}, - use_default_dialect=True + {"a": 1, "b": 2}, + use_default_dialect=True, ) |
