diff options
| author | Khairi Hafsham <jumanjisama@gmail.com> | 2017-02-02 13:02:21 -0500 |
|---|---|---|
| committer | Mike Bayer <mike_mp@zzzcomputing.com> | 2017-02-07 11:21:56 -0500 |
| commit | 772374735da27df1ddb907f4a0f5085b46dbe82b (patch) | |
| tree | 41f88c835a74d4665c97853ae8676a5181d61db3 /test/dialect | |
| parent | d71f4b47186972c5248c94ee2d04364da95a0965 (diff) | |
| download | sqlalchemy-772374735da27df1ddb907f4a0f5085b46dbe82b.tar.gz | |
Make all tests to be PEP8 compliant
tested using pycodestyle version 2.2.0
Fixes: #3885
Change-Id: I5df43adc3aefe318f9eeab72a078247a548ec566
Pull-request: https://github.com/zzzeek/sqlalchemy/pull/343
Diffstat (limited to 'test/dialect')
22 files changed, 597 insertions, 584 deletions
diff --git a/test/dialect/mssql/test_compiler.py b/test/dialect/mssql/test_compiler.py index 0cf84c7d5..10a7d09ab 100644 --- a/test/dialect/mssql/test_compiler.py +++ b/test/dialect/mssql/test_compiler.py @@ -11,6 +11,7 @@ from sqlalchemy import Integer, String, Table, Column, select, MetaData,\ UniqueConstraint, Index, Sequence, literal from sqlalchemy import testing + class CompileTest(fixtures.TestBase, AssertsCompiledSQL): __dialect__ = mssql.dialect() diff --git a/test/dialect/mssql/test_engine.py b/test/dialect/mssql/test_engine.py index 39562e0db..1021f4e99 100644 --- a/test/dialect/mssql/test_engine.py +++ b/test/dialect/mssql/test_engine.py @@ -42,7 +42,9 @@ class ParseConnectTest(fixtures.TestBase): def test_pyodbc_hostname(self): dialect = pyodbc.dialect() - u = url.make_url('mssql://username:password@hostspec/database?driver=SQL+Server') + u = url.make_url( + 'mssql://username:password@hostspec/database?driver=SQL+Server' + ) connection = dialect.create_connect_args(u) eq_([['DRIVER={SQL Server};Server=hostspec;Database=database;UI' 'D=username;PWD=password'], {}], connection) @@ -88,9 +90,9 @@ class ParseConnectTest(fixtures.TestBase): eq_(connection[1], {}) eq_(connection[0][0] in ('DRIVER={SQL Server};Server=hostspec;Database=database;' - 'UID=username;PWD=password;foo=bar;LANGUAGE=us_english', - 'DRIVER={SQL Server};Server=hostspec;Database=database;UID=' - 'username;PWD=password;LANGUAGE=us_english;foo=bar'), True) + 'UID=username;PWD=password;foo=bar;LANGUAGE=us_english', + 'DRIVER={SQL Server};Server=hostspec;Database=database;UID=' + 'username;PWD=password;LANGUAGE=us_english;foo=bar'), True) def test_pyodbc_odbc_connect(self): dialect = pyodbc.dialect() @@ -174,7 +176,7 @@ class ParseConnectTest(fixtures.TestBase): connection = dialect.create_connect_args(u) eq_( [[], {'host': 'somehost', 'password': 'tiger', - 'user': 'scott', 'database': 'test'}], connection + 'user': 'scott', 'database': 'test'}], connection ) u = \ @@ -182,7 +184,7 @@ class ParseConnectTest(fixtures.TestBase): connection = dialect.create_connect_args(u) eq_( [[], {'host': 'somehost:5000', 'password': 'tiger', - 'user': 'scott', 'database': 'test'}], connection + 'user': 'scott', 'database': 'test'}], connection ) def test_pymssql_disconnect(self): @@ -195,7 +197,7 @@ class ParseConnectTest(fixtures.TestBase): 'Error 10054', 'Not connected to any MS SQL server', 'Connection is closed' - ]: + ]: eq_(dialect.is_disconnect(error, None, None), True) eq_(dialect.is_disconnect("not an error", None, None), False) @@ -239,4 +241,4 @@ class VersionDetectionTest(fixtures.TestBase): eq_( dialect._get_server_version_info(conn), (11, 0, 9216, 62) - )
\ No newline at end of file + ) diff --git a/test/dialect/mssql/test_query.py b/test/dialect/mssql/test_query.py index 81fa2fb1e..1164270e9 100644 --- a/test/dialect/mssql/test_query.py +++ b/test/dialect/mssql/test_query.py @@ -477,7 +477,7 @@ def full_text_search_missing(): connection.execute('CREATE FULLTEXT CATALOG Catalog AS ' 'DEFAULT') return False - except: + except Exception: return True finally: connection.close() diff --git a/test/dialect/mssql/test_reflection.py b/test/dialect/mssql/test_reflection.py index ccde93ba1..5e17709b4 100644 --- a/test/dialect/mssql/test_reflection.py +++ b/test/dialect/mssql/test_reflection.py @@ -3,11 +3,14 @@ from sqlalchemy.testing import eq_, is_, in_ from sqlalchemy import * from sqlalchemy import types, schema, event from sqlalchemy.databases import mssql -from sqlalchemy.testing import fixtures, AssertsCompiledSQL, \ - ComparesTables +from sqlalchemy.testing import (fixtures, + AssertsCompiledSQL, + ComparesTables) from sqlalchemy import testing from sqlalchemy.engine.reflection import Inspector from sqlalchemy import util +from sqlalchemy.dialects.mssql.information_schema import CoerceUnicode, tables +from sqlalchemy.dialects.mssql import base class ReflectionTest(fixtures.TestBase, ComparesTables): @@ -37,7 +40,7 @@ class ReflectionTest(fixtures.TestBase, ComparesTables): server_default='5'), Column('test9', types.BINARY(100)), Column('test_numeric', types.Numeric()), - ) + ) addresses = Table( 'engine_email_addresses', @@ -54,7 +57,9 @@ class ReflectionTest(fixtures.TestBase, ComparesTables): autoload=True, autoload_with=testing.db) reflected_addresses = Table('engine_email_addresses', - meta2, autoload=True, autoload_with=testing.db) + meta2, + autoload=True, + autoload_with=testing.db) self.assert_tables_equal(users, reflected_users) self.assert_tables_equal(addresses, reflected_addresses) @@ -70,7 +75,7 @@ class ReflectionTest(fixtures.TestBase, ComparesTables): meta2 = MetaData(testing.db) table2 = Table('identity_test', meta2, autoload=True) sequence = isinstance(table2.c['col1'].default, schema.Sequence) \ - and table2.c['col1'].default + and table2.c['col1'].default assert sequence.start == 2 assert sequence.increment == 3 @@ -85,15 +90,14 @@ class ReflectionTest(fixtures.TestBase, ComparesTables): assert isinstance(t1.c.id.type, Integer) assert isinstance(t1.c.data.type, types.NullType) - @testing.provide_metadata def test_db_qualified_items(self): metadata = self.metadata Table('foo', metadata, Column('id', Integer, primary_key=True)) - Table('bar', metadata, - Column('id', Integer, primary_key=True), - Column('foo_id', Integer, ForeignKey('foo.id', name="fkfoo")) - ) + Table('bar', + metadata, + Column('id', Integer, primary_key=True), + Column('foo_id', Integer, ForeignKey('foo.id', name="fkfoo"))) metadata.create_all() dbname = testing.db.scalar("select db_name()") @@ -101,7 +105,7 @@ class ReflectionTest(fixtures.TestBase, ComparesTables): inspector = inspect(testing.db) bar_via_db = inspector.get_foreign_keys( - "bar", schema="%s.%s" % (dbname, owner)) + "bar", schema="%s.%s" % (dbname, owner)) eq_( bar_via_db, [{ @@ -119,7 +123,6 @@ class ReflectionTest(fixtures.TestBase, ComparesTables): autoload_with=testing.db) eq_(m2.tables["test.dbo.foo"].schema, "test.dbo") - @testing.provide_metadata def test_indexes_cols(self): metadata = self.metadata @@ -140,10 +143,10 @@ class ReflectionTest(fixtures.TestBase, ComparesTables): def test_indexes_cols_with_commas(self): metadata = self.metadata - t1 = Table('t', metadata, - Column('x, col', Integer, key='x'), - Column('y', Integer) - ) + t1 = Table('t', + metadata, + Column('x, col', Integer, key='x'), + Column('y', Integer)) Index('foo', t1.c.x, t1.c.y) metadata.create_all() @@ -159,8 +162,10 @@ class ReflectionTest(fixtures.TestBase, ComparesTables): def test_indexes_cols_with_spaces(self): metadata = self.metadata - t1 = Table('t', metadata, Column('x col', Integer, key='x'), - Column('y', Integer)) + t1 = Table('t', + metadata, + Column('x col', Integer, key='x'), + Column('y', Integer)) Index('foo', t1.c.x, t1.c.y) metadata.create_all() @@ -198,9 +203,6 @@ class ReflectionTest(fixtures.TestBase, ComparesTables): in_('max', str(col['type'].compile(dialect=testing.db.dialect))) -from sqlalchemy.dialects.mssql.information_schema import CoerceUnicode, tables -from sqlalchemy.dialects.mssql import base - class InfoCoerceUnicodeTest(fixtures.TestBase, AssertsCompiledSQL): def test_info_unicode_coercion(self): @@ -224,10 +226,12 @@ class InfoCoerceUnicodeTest(fixtures.TestBase, AssertsCompiledSQL): stmt = tables.c.table_name == 'somename' self.assert_compile( stmt, - "[INFORMATION_SCHEMA].[TABLES].[TABLE_NAME] = CAST(:table_name_1 AS NVARCHAR(max))", + "[INFORMATION_SCHEMA].[TABLES].[TABLE_NAME] = " + "CAST(:table_name_1 AS NVARCHAR(max))", dialect=dialect ) + class ReflectHugeViewTest(fixtures.TestBase): __only_on__ = 'mssql' __backend__ = True @@ -242,20 +246,19 @@ class ReflectHugeViewTest(fixtures.TestBase): self.metadata = MetaData(testing.db) t = Table('base_table', self.metadata, - *[ - Column("long_named_column_number_%d" % i, Integer) + *[Column("long_named_column_number_%d" % i, Integer) for i in range(self.col_num) - ] - ) + ] + ) self.view_str = view_str = \ "CREATE VIEW huge_named_view AS SELECT %s FROM base_table" % ( - ",".join("long_named_column_number_%d" % i - for i in range(self.col_num)) + ",".join("long_named_column_number_%d" % i + for i in range(self.col_num)) ) assert len(view_str) > 4000 - event.listen(t, 'after_create', DDL(view_str) ) - event.listen(t, 'before_drop', DDL("DROP VIEW huge_named_view") ) + event.listen(t, 'after_create', DDL(view_str)) + event.listen(t, 'before_drop', DDL("DROP VIEW huge_named_view")) self.metadata.create_all() @@ -266,4 +269,3 @@ class ReflectHugeViewTest(fixtures.TestBase): inspector = Inspector.from_engine(testing.db) view_def = inspector.get_view_definition("huge_named_view") eq_(view_def, self.view_str) - diff --git a/test/dialect/mssql/test_types.py b/test/dialect/mssql/test_types.py index f13e26c67..841624303 100644 --- a/test/dialect/mssql/test_types.py +++ b/test/dialect/mssql/test_types.py @@ -427,6 +427,7 @@ class TypeDDLTest(fixtures.TestBase): % (col.name, columns[index][3])) self.assert_(repr(col)) + metadata = None @@ -826,6 +827,7 @@ class MonkeyPatchedBinaryTest(fixtures.TestBase): result = module.Binary(input) eq_(result, expected_result) + binary_table = None MyPickleType = None @@ -930,17 +932,17 @@ class BinaryTest(fixtures.TestBase, AssertsExecutionResults): mypickle=MyPickleType), bind=testing.db): with engine.connect() as conn: - l = conn.execute(stmt).fetchall() - eq_(list(stream1), list(l[0]['data'])) + result = conn.execute(stmt).fetchall() + eq_(list(stream1), list(result[0]['data'])) paddedstream = list(stream1[0:100]) paddedstream.extend(['\x00'] * (100 - len(paddedstream))) - eq_(paddedstream, list(l[0]['data_slice'])) - eq_(list(stream2), list(l[1]['data'])) - eq_(list(stream2), list(l[1]['data_image'])) - eq_(testobj1, l[0]['pickled']) - eq_(testobj2, l[1]['pickled']) - eq_(testobj3.moredata, l[0]['mypickle'].moredata) - eq_(l[0]['mypickle'].stuff, 'this is the right stuff') + eq_(paddedstream, list(result[0]['data_slice'])) + eq_(list(stream2), list(result[1]['data'])) + eq_(list(stream2), list(result[1]['data_image'])) + eq_(testobj1, result[0]['pickled']) + eq_(testobj2, result[1]['pickled']) + eq_(testobj3.moredata, result[0]['mypickle'].moredata) + eq_(result[0]['mypickle'].stuff, 'this is the right stuff') def _test_binary_none(self, deprecate_large_types): engine = engines.testing_engine( diff --git a/test/dialect/mysql/test_compiler.py b/test/dialect/mysql/test_compiler.py index b72507104..35ff603a8 100644 --- a/test/dialect/mysql/test_compiler.py +++ b/test/dialect/mysql/test_compiler.py @@ -17,6 +17,7 @@ from sqlalchemy import testing from sqlalchemy.sql import table, column import re + class CompileTest(fixtures.TestBase, AssertsCompiledSQL): __dialect__ = mysql.dialect() @@ -43,10 +44,12 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): def test_create_index_with_prefix(self): m = MetaData() tbl = Table('testtbl', m, Column('data', String(255))) - idx = Index('test_idx1', tbl.c.data, mysql_length=10, mysql_prefix='FULLTEXT') + idx = Index('test_idx1', tbl.c.data, mysql_length=10, + mysql_prefix='FULLTEXT') self.assert_compile(schema.CreateIndex(idx), - 'CREATE FULLTEXT INDEX test_idx1 ON testtbl (data(10))') + 'CREATE FULLTEXT INDEX test_idx1 ' + 'ON testtbl (data(10))') def test_create_index_with_length(self): m = MetaData() @@ -645,4 +648,4 @@ class SQLTest(fixtures.TestBase, AssertsCompiledSQL): self.assert_compile( t1.outerjoin(t2, t1.c.x == t2.c.y, full=True), "t1 FULL OUTER JOIN t2 ON t1.x = t2.y" - )
\ No newline at end of file + ) diff --git a/test/dialect/mysql/test_dialect.py b/test/dialect/mysql/test_dialect.py index ab719b368..37c613e72 100644 --- a/test/dialect/mysql/test_dialect.py +++ b/test/dialect/mysql/test_dialect.py @@ -26,7 +26,7 @@ class DialectTest(fixtures.TestBase): def _test_ssl_arguments(self, dialect): kwarg = dialect.create_connect_args( make_url("mysql://scott:tiger@localhost:3306/test" - "?ssl_ca=/ca.pem&ssl_cert=/cert.pem&ssl_key=/key.pem") + "?ssl_ca=/ca.pem&ssl_cert=/cert.pem&ssl_key=/key.pem") )[1] # args that differ among mysqldb and oursql for k in ('use_unicode', 'found_rows', 'client_flag'): @@ -64,16 +64,19 @@ class DialectTest(fixtures.TestBase): from sqlalchemy.dialects.mysql import mysqlconnector dialect = mysqlconnector.dialect() kw = dialect.create_connect_args( - make_url("mysql+mysqlconnector://u:p@host/db?raise_on_warnings=true") - )[1] + make_url( + "mysql+mysqlconnector://u:p@host/db?raise_on_warnings=true" + ) + )[1] eq_(kw['raise_on_warnings'], True) kw = dialect.create_connect_args( - make_url("mysql+mysqlconnector://u:p@host/db?raise_on_warnings=false") - )[1] + make_url( + "mysql+mysqlconnector://u:p@host/db?raise_on_warnings=false" + ) + )[1] eq_(kw['raise_on_warnings'], False) - kw = dialect.create_connect_args( make_url("mysql+mysqlconnector://u:p@host/db") )[1] @@ -121,6 +124,7 @@ class DialectTest(fixtures.TestBase): ) assert c.execute('SELECT @@tx_isolation;').scalar() == mysql_value + class SQLModeDetectionTest(fixtures.TestBase): __only_on__ = 'mysql' __backend__ = True @@ -130,7 +134,7 @@ class SQLModeDetectionTest(fixtures.TestBase): cursor = con.cursor() cursor.execute("set sql_mode='%s'" % (",".join(modes))) e = engines.testing_engine(options={ - 'pool_events':[ + 'pool_events': [ (connect, 'first_connect'), (connect, 'connect') ] @@ -186,4 +190,3 @@ class ExecutionTest(fixtures.TestBase): def test_sysdate(self): d = testing.db.scalar(func.sysdate()) assert isinstance(d, datetime.datetime) - diff --git a/test/dialect/mysql/test_query.py b/test/dialect/mysql/test_query.py index c6b7a1036..04f3ca67d 100644 --- a/test/dialect/mysql/test_query.py +++ b/test/dialect/mysql/test_query.py @@ -38,16 +38,16 @@ class MatchTest(fixtures.TestBase): metadata = MetaData(testing.db) cattable = Table('cattable', metadata, - Column('id', Integer, primary_key=True), - Column('description', String(50)), - mysql_engine='MyISAM' - ) + Column('id', Integer, primary_key=True), + Column('description', String(50)), + mysql_engine='MyISAM') matchtable = Table('matchtable', metadata, - Column('id', Integer, primary_key=True), - Column('title', String(200)), - Column('category_id', Integer, ForeignKey('cattable.id')), - mysql_engine='MyISAM' - ) + Column('id', Integer, primary_key=True), + Column('title', String(200)), + Column('category_id', + Integer, + ForeignKey('cattable.id')), + mysql_engine='MyISAM') metadata.create_all() cattable.insert().execute([ @@ -103,7 +103,8 @@ class MatchTest(fixtures.TestBase): # test [ticket:3263] result = testing.db.execute( select([ - matchtable.c.title.match('Agile Ruby Programming').label('ruby'), + matchtable.c.title.match('Agile Ruby Programming') + .label('ruby'), matchtable.c.title.match('Dive Python').label('python'), matchtable.c.title ]).order_by(matchtable.c.id) @@ -149,7 +150,7 @@ class MatchTest(fixtures.TestBase): def test_match_across_joins(self): results = (matchtable.select(). - where(and_(cattable.c.id==matchtable.c.category_id, + where(and_(cattable.c.id == matchtable.c.category_id, or_(cattable.c.description.match('Ruby'), matchtable.c.title.match('nutshell')))). order_by(matchtable.c.id). @@ -211,4 +212,3 @@ class AnyAllTest(fixtures.TablesTest): is_( testing.db.execute(stmt).scalar(), True ) - diff --git a/test/dialect/mysql/test_reflection.py b/test/dialect/mysql/test_reflection.py index 2ccb299bc..9e122e680 100644 --- a/test/dialect/mysql/test_reflection.py +++ b/test/dialect/mysql/test_reflection.py @@ -184,7 +184,7 @@ class TypeReflectionTest(fixtures.TestBase): def test_legacy_enum_types(self): specs = [ - (mysql.ENUM("''","'fleem'"), mysql.ENUM("''","'fleem'")), # noqa + (mysql.ENUM("''", "'fleem'"), mysql.ENUM("''", "'fleem'")), ] self._run_test(specs, ['enums']) @@ -399,7 +399,10 @@ class ReflectionTest(fixtures.TestBase, AssertsExecutionResults): @testing.provide_metadata def test_view_reflection(self): - Table('x', self.metadata, Column('a', Integer), Column('b', String(50))) + Table('x', + self.metadata, + Column('a', Integer), + Column('b', String(50))) self.metadata.create_all() with testing.db.connect() as conn: @@ -427,7 +430,6 @@ class ReflectionTest(fixtures.TestBase, AssertsExecutionResults): [('a', mysql.INTEGER), ('b', mysql.VARCHAR)] ) - @testing.exclude('mysql', '<', (5, 0, 0), 'no information_schema support') def test_system_views(self): dialect = testing.db.dialect @@ -573,14 +575,11 @@ class RawReflectionTest(fixtures.TestBase): 'REFERENCES `users` (`id`) ' 'ON DELETE CASCADE ON UPDATE CASCADE') eq_(m.groups(), ('addresses_user_id_fkey', '`user_id`', - '`users`', '`id`', None, 'CASCADE', 'CASCADE')) - + '`users`', '`id`', None, 'CASCADE', 'CASCADE')) m = regex.match(' CONSTRAINT `addresses_user_id_fkey` ' 'FOREIGN KEY (`user_id`) ' 'REFERENCES `users` (`id`) ' 'ON DELETE CASCADE ON UPDATE SET NULL') eq_(m.groups(), ('addresses_user_id_fkey', '`user_id`', - '`users`', '`id`', None, 'CASCADE', 'SET NULL')) - - + '`users`', '`id`', None, 'CASCADE', 'SET NULL')) diff --git a/test/dialect/mysql/test_types.py b/test/dialect/mysql/test_types.py index 48663da25..1a7b7df3d 100644 --- a/test/dialect/mysql/test_types.py +++ b/test/dialect/mysql/test_types.py @@ -6,14 +6,18 @@ from sqlalchemy import sql, exc, schema from sqlalchemy.util import u from sqlalchemy import util from sqlalchemy.dialects.mysql import base as mysql -from sqlalchemy.testing import fixtures, AssertsCompiledSQL, AssertsExecutionResults +from sqlalchemy.testing import (fixtures, + AssertsCompiledSQL, + AssertsExecutionResults) from sqlalchemy import testing import datetime import decimal from sqlalchemy import types as sqltypes -class TypesTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL): +class TypesTest(fixtures.TestBase, + AssertsExecutionResults, + AssertsCompiledSQL): "Test MySQL column types" __dialect__ = mysql.dialect() @@ -33,11 +37,11 @@ class TypesTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL): 'NUMERIC'), (mysql.MSNumeric, [12], {}, 'NUMERIC(12)'), - (mysql.MSNumeric, [12, 4], {'unsigned':True}, + (mysql.MSNumeric, [12, 4], {'unsigned': True}, 'NUMERIC(12, 4) UNSIGNED'), - (mysql.MSNumeric, [12, 4], {'zerofill':True}, + (mysql.MSNumeric, [12, 4], {'zerofill': True}, 'NUMERIC(12, 4) ZEROFILL'), - (mysql.MSNumeric, [12, 4], {'zerofill':True, 'unsigned':True}, + (mysql.MSNumeric, [12, 4], {'zerofill': True, 'unsigned': True}, 'NUMERIC(12, 4) UNSIGNED ZEROFILL'), (mysql.MSDecimal, [], {}, @@ -48,29 +52,29 @@ class TypesTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL): 'DECIMAL(12)'), (mysql.MSDecimal, [12, None], {}, 'DECIMAL(12)'), - (mysql.MSDecimal, [12, 4], {'unsigned':True}, + (mysql.MSDecimal, [12, 4], {'unsigned': True}, 'DECIMAL(12, 4) UNSIGNED'), - (mysql.MSDecimal, [12, 4], {'zerofill':True}, + (mysql.MSDecimal, [12, 4], {'zerofill': True}, 'DECIMAL(12, 4) ZEROFILL'), - (mysql.MSDecimal, [12, 4], {'zerofill':True, 'unsigned':True}, + (mysql.MSDecimal, [12, 4], {'zerofill': True, 'unsigned': True}, 'DECIMAL(12, 4) UNSIGNED ZEROFILL'), (mysql.MSDouble, [None, None], {}, 'DOUBLE'), - (mysql.MSDouble, [12, 4], {'unsigned':True}, + (mysql.MSDouble, [12, 4], {'unsigned': True}, 'DOUBLE(12, 4) UNSIGNED'), - (mysql.MSDouble, [12, 4], {'zerofill':True}, + (mysql.MSDouble, [12, 4], {'zerofill': True}, 'DOUBLE(12, 4) ZEROFILL'), - (mysql.MSDouble, [12, 4], {'zerofill':True, 'unsigned':True}, + (mysql.MSDouble, [12, 4], {'zerofill': True, 'unsigned': True}, 'DOUBLE(12, 4) UNSIGNED ZEROFILL'), (mysql.MSReal, [None, None], {}, 'REAL'), - (mysql.MSReal, [12, 4], {'unsigned':True}, + (mysql.MSReal, [12, 4], {'unsigned': True}, 'REAL(12, 4) UNSIGNED'), - (mysql.MSReal, [12, 4], {'zerofill':True}, + (mysql.MSReal, [12, 4], {'zerofill': True}, 'REAL(12, 4) ZEROFILL'), - (mysql.MSReal, [12, 4], {'zerofill':True, 'unsigned':True}, + (mysql.MSReal, [12, 4], {'zerofill': True, 'unsigned': True}, 'REAL(12, 4) UNSIGNED ZEROFILL'), (mysql.MSFloat, [], {}, @@ -81,68 +85,68 @@ class TypesTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL): 'FLOAT(12)'), (mysql.MSFloat, [12, 4], {}, 'FLOAT(12, 4)'), - (mysql.MSFloat, [12, 4], {'unsigned':True}, + (mysql.MSFloat, [12, 4], {'unsigned': True}, 'FLOAT(12, 4) UNSIGNED'), - (mysql.MSFloat, [12, 4], {'zerofill':True}, + (mysql.MSFloat, [12, 4], {'zerofill': True}, 'FLOAT(12, 4) ZEROFILL'), - (mysql.MSFloat, [12, 4], {'zerofill':True, 'unsigned':True}, + (mysql.MSFloat, [12, 4], {'zerofill': True, 'unsigned': True}, 'FLOAT(12, 4) UNSIGNED ZEROFILL'), (mysql.MSInteger, [], {}, 'INTEGER'), (mysql.MSInteger, [4], {}, 'INTEGER(4)'), - (mysql.MSInteger, [4], {'unsigned':True}, + (mysql.MSInteger, [4], {'unsigned': True}, 'INTEGER(4) UNSIGNED'), - (mysql.MSInteger, [4], {'zerofill':True}, + (mysql.MSInteger, [4], {'zerofill': True}, 'INTEGER(4) ZEROFILL'), - (mysql.MSInteger, [4], {'zerofill':True, 'unsigned':True}, + (mysql.MSInteger, [4], {'zerofill': True, 'unsigned': True}, 'INTEGER(4) UNSIGNED ZEROFILL'), (mysql.MSBigInteger, [], {}, 'BIGINT'), (mysql.MSBigInteger, [4], {}, 'BIGINT(4)'), - (mysql.MSBigInteger, [4], {'unsigned':True}, + (mysql.MSBigInteger, [4], {'unsigned': True}, 'BIGINT(4) UNSIGNED'), - (mysql.MSBigInteger, [4], {'zerofill':True}, + (mysql.MSBigInteger, [4], {'zerofill': True}, 'BIGINT(4) ZEROFILL'), - (mysql.MSBigInteger, [4], {'zerofill':True, 'unsigned':True}, + (mysql.MSBigInteger, [4], {'zerofill': True, 'unsigned': True}, 'BIGINT(4) UNSIGNED ZEROFILL'), - (mysql.MSMediumInteger, [], {}, - 'MEDIUMINT'), - (mysql.MSMediumInteger, [4], {}, - 'MEDIUMINT(4)'), - (mysql.MSMediumInteger, [4], {'unsigned':True}, - 'MEDIUMINT(4) UNSIGNED'), - (mysql.MSMediumInteger, [4], {'zerofill':True}, - 'MEDIUMINT(4) ZEROFILL'), - (mysql.MSMediumInteger, [4], {'zerofill':True, 'unsigned':True}, - 'MEDIUMINT(4) UNSIGNED ZEROFILL'), + (mysql.MSMediumInteger, [], {}, + 'MEDIUMINT'), + (mysql.MSMediumInteger, [4], {}, + 'MEDIUMINT(4)'), + (mysql.MSMediumInteger, [4], {'unsigned': True}, + 'MEDIUMINT(4) UNSIGNED'), + (mysql.MSMediumInteger, [4], {'zerofill': True}, + 'MEDIUMINT(4) ZEROFILL'), + (mysql.MSMediumInteger, [4], {'zerofill': True, 'unsigned': True}, + 'MEDIUMINT(4) UNSIGNED ZEROFILL'), (mysql.MSTinyInteger, [], {}, 'TINYINT'), (mysql.MSTinyInteger, [1], {}, 'TINYINT(1)'), - (mysql.MSTinyInteger, [1], {'unsigned':True}, + (mysql.MSTinyInteger, [1], {'unsigned': True}, 'TINYINT(1) UNSIGNED'), - (mysql.MSTinyInteger, [1], {'zerofill':True}, + (mysql.MSTinyInteger, [1], {'zerofill': True}, 'TINYINT(1) ZEROFILL'), - (mysql.MSTinyInteger, [1], {'zerofill':True, 'unsigned':True}, + (mysql.MSTinyInteger, [1], {'zerofill': True, 'unsigned': True}, 'TINYINT(1) UNSIGNED ZEROFILL'), (mysql.MSSmallInteger, [], {}, 'SMALLINT'), (mysql.MSSmallInteger, [4], {}, 'SMALLINT(4)'), - (mysql.MSSmallInteger, [4], {'unsigned':True}, + (mysql.MSSmallInteger, [4], {'unsigned': True}, 'SMALLINT(4) UNSIGNED'), - (mysql.MSSmallInteger, [4], {'zerofill':True}, + (mysql.MSSmallInteger, [4], {'zerofill': True}, 'SMALLINT(4) ZEROFILL'), - (mysql.MSSmallInteger, [4], {'zerofill':True, 'unsigned':True}, + (mysql.MSSmallInteger, [4], {'zerofill': True, 'unsigned': True}, 'SMALLINT(4) UNSIGNED ZEROFILL'), - ] + ] for type_, args, kw, res in columns: type_inst = type_(*args, **kw) @@ -161,11 +165,10 @@ class TypesTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL): @testing.provide_metadata def test_precision_float_roundtrip(self): t = Table('t', self.metadata, - Column('scale_value', mysql.DOUBLE( - precision=15, scale=12, asdecimal=True)), - Column('unscale_value', mysql.DOUBLE( - decimal_return_scale=12, asdecimal=True)) - ) + Column('scale_value', mysql.DOUBLE( + precision=15, scale=12, asdecimal=True)), + Column('unscale_value', mysql.DOUBLE( + decimal_return_scale=12, asdecimal=True))) t.create(testing.db) testing.db.execute( t.insert(), scale_value=45.768392065789, @@ -184,25 +187,25 @@ class TypesTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL): columns = [ (mysql.MSChar, [1], {}, 'CHAR(1)'), - (mysql.NCHAR, [1], {}, - 'NATIONAL CHAR(1)'), - (mysql.MSChar, [1], {'binary':True}, + (mysql.NCHAR, [1], {}, + 'NATIONAL CHAR(1)'), + (mysql.MSChar, [1], {'binary': True}, 'CHAR(1) BINARY'), - (mysql.MSChar, [1], {'ascii':True}, + (mysql.MSChar, [1], {'ascii': True}, 'CHAR(1) ASCII'), - (mysql.MSChar, [1], {'unicode':True}, + (mysql.MSChar, [1], {'unicode': True}, 'CHAR(1) UNICODE'), - (mysql.MSChar, [1], {'ascii':True, 'binary':True}, + (mysql.MSChar, [1], {'ascii': True, 'binary': True}, 'CHAR(1) ASCII BINARY'), - (mysql.MSChar, [1], {'unicode':True, 'binary':True}, + (mysql.MSChar, [1], {'unicode': True, 'binary': True}, 'CHAR(1) UNICODE BINARY'), - (mysql.MSChar, [1], {'charset':'utf8'}, + (mysql.MSChar, [1], {'charset': 'utf8'}, 'CHAR(1) CHARACTER SET utf8'), - (mysql.MSChar, [1], {'charset':'utf8', 'binary':True}, + (mysql.MSChar, [1], {'charset': 'utf8', 'binary': True}, 'CHAR(1) CHARACTER SET utf8 BINARY'), - (mysql.MSChar, [1], {'charset':'utf8', 'unicode':True}, + (mysql.MSChar, [1], {'charset': 'utf8', 'unicode': True}, 'CHAR(1) CHARACTER SET utf8'), - (mysql.MSChar, [1], {'charset':'utf8', 'ascii':True}, + (mysql.MSChar, [1], {'charset': 'utf8', 'ascii': True}, 'CHAR(1) CHARACTER SET utf8'), (mysql.MSChar, [1], {'collation': 'utf8_bin'}, 'CHAR(1) COLLATE utf8_bin'), @@ -211,42 +214,42 @@ class TypesTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL): (mysql.MSChar, [1], {'charset': 'utf8', 'binary': True}, 'CHAR(1) CHARACTER SET utf8 BINARY'), (mysql.MSChar, [1], {'charset': 'utf8', 'collation': 'utf8_bin', - 'binary': True}, + 'binary': True}, 'CHAR(1) CHARACTER SET utf8 COLLATE utf8_bin'), - (mysql.MSChar, [1], {'national':True}, + (mysql.MSChar, [1], {'national': True}, 'NATIONAL CHAR(1)'), - (mysql.MSChar, [1], {'national':True, 'charset':'utf8'}, + (mysql.MSChar, [1], {'national': True, 'charset': 'utf8'}, 'NATIONAL CHAR(1)'), - (mysql.MSChar, [1], {'national':True, 'charset':'utf8', - 'binary':True}, + (mysql.MSChar, [1], {'national': True, 'charset': 'utf8', + 'binary': True}, 'NATIONAL CHAR(1) BINARY'), - (mysql.MSChar, [1], {'national':True, 'binary':True, - 'unicode':True}, + (mysql.MSChar, [1], {'national': True, 'binary': True, + 'unicode': True}, 'NATIONAL CHAR(1) BINARY'), - (mysql.MSChar, [1], {'national':True, 'collation':'utf8_bin'}, + (mysql.MSChar, [1], {'national': True, 'collation': 'utf8_bin'}, 'NATIONAL CHAR(1) COLLATE utf8_bin'), - (mysql.MSString, [1], {'charset':'utf8', 'collation':'utf8_bin'}, + (mysql.MSString, [1], {'charset': 'utf8', 'collation': 'utf8_bin'}, 'VARCHAR(1) CHARACTER SET utf8 COLLATE utf8_bin'), - (mysql.MSString, [1], {'national':True, 'collation':'utf8_bin'}, + (mysql.MSString, [1], {'national': True, 'collation': 'utf8_bin'}, 'NATIONAL VARCHAR(1) COLLATE utf8_bin'), - (mysql.MSTinyText, [], {'charset':'utf8', 'collation':'utf8_bin'}, + (mysql.MSTinyText, + [], + {'charset': 'utf8', 'collation': 'utf8_bin'}, 'TINYTEXT CHARACTER SET utf8 COLLATE utf8_bin'), - (mysql.MSMediumText, [], {'charset':'utf8', 'binary':True}, + (mysql.MSMediumText, [], {'charset': 'utf8', 'binary': True}, 'MEDIUMTEXT CHARACTER SET utf8 BINARY'), - (mysql.MSLongText, [], {'ascii':True}, + (mysql.MSLongText, [], {'ascii': True}, 'LONGTEXT ASCII'), - (mysql.ENUM, ["foo", "bar"], {'unicode':True}, + (mysql.ENUM, ["foo", "bar"], {'unicode': True}, '''ENUM('foo','bar') UNICODE'''), (String, [20], {"collation": "utf8"}, 'VARCHAR(20) COLLATE utf8') - - - ] + ] for type_, args, kw, res in columns: type_inst = type_(*args, **kw) @@ -257,8 +260,8 @@ class TypesTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL): # test that repr() copies out all arguments self.assert_compile( eval("mysql.%r" % type_inst) - if type_ is not String - else eval("%r" % type_inst), + if type_ is not String + else eval("%r" % type_inst), res ) @@ -268,11 +271,10 @@ class TypesTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL): @testing.provide_metadata def test_charset_collate_table(self): t = Table('foo', self.metadata, - Column('id', Integer), - Column('data', UnicodeText), - mysql_default_charset='utf8', - mysql_collate='utf8_bin' - ) + Column('id', Integer), + Column('data', UnicodeText), + mysql_default_charset='utf8', + mysql_collate='utf8_bin') t.create() m2 = MetaData(testing.db) t2 = Table('foo', m2, autoload=True) @@ -284,7 +286,8 @@ class TypesTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL): # MySQLdb 1.2.3 and also need to pass either use_unicode=1 # or charset=utf8 to the URL. t.insert().execute(id=1, data=u('some text')) - assert isinstance(testing.db.scalar(select([t.c.data])), util.text_type) + assert isinstance(testing.db.scalar(select([t.c.data])), + util.text_type) def test_bit_50(self): """Exercise BIT types on 5.0+ (not valid for all engine types)""" @@ -321,7 +324,7 @@ class TypesTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL): row = table.select().execute().first() try: self.assert_(list(row) == expected) - except: + except Exception: print("Storing %s" % store) print("Expected %s" % expected) print("Found %s" % list(row)) @@ -360,8 +363,7 @@ class TypesTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL): Column('b2', Boolean), Column('b3', mysql.MSTinyInteger(1)), Column('b4', mysql.MSTinyInteger(1, unsigned=True)), - Column('b5', mysql.MSTinyInteger), - ) + Column('b5', mysql.MSTinyInteger)) self.metadata.create_all() table = bool_table @@ -394,8 +396,7 @@ class TypesTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL): Column('b2', Boolean), Column('b3', BOOLEAN), Column('b4', BOOLEAN), - autoload=True, - ) + autoload=True) eq_(colspec(table.c.b3), 'b3 BOOL') eq_(colspec(table.c.b4), 'b4 BOOL') roundtrip([None, None, None, None, None]) @@ -410,8 +411,8 @@ class TypesTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL): def test_timestamp_fsp(self): self.assert_compile( - mysql.TIMESTAMP(fsp=5), - "TIMESTAMP(5)" + mysql.TIMESTAMP(fsp=5), + "TIMESTAMP(5)" ) def test_timestamp_defaults(self): @@ -532,40 +533,37 @@ class TypesTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL): ] ) - def test_datetime_generic(self): self.assert_compile( - mysql.DATETIME(), - "DATETIME" + mysql.DATETIME(), + "DATETIME" ) def test_datetime_fsp(self): self.assert_compile( - mysql.DATETIME(fsp=4), - "DATETIME(4)" + mysql.DATETIME(fsp=4), + "DATETIME(4)" ) - def test_time_generic(self): """"Exercise TIME.""" self.assert_compile( - mysql.TIME(), - "TIME" + mysql.TIME(), + "TIME" ) def test_time_fsp(self): self.assert_compile( - mysql.TIME(fsp=5), - "TIME(5)" + mysql.TIME(fsp=5), + "TIME(5)" ) def test_time_result_processor(self): eq_( mysql.TIME().result_processor(None, None)( - datetime.timedelta(seconds=35, minutes=517, - microseconds=450 - )), + datetime.timedelta(seconds=35, minutes=517, + microseconds=450)), datetime.time(8, 37, 35, 450) ) @@ -573,8 +571,7 @@ class TypesTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL): @testing.provide_metadata def test_time_roundtrip(self): t = Table('mysql_time', self.metadata, - Column('t1', mysql.TIME()) - ) + Column('t1', mysql.TIME())) t.create() t.insert().values(t1=datetime.time(8, 37, 35)).execute() eq_(select([t.c.t1]).scalar(), datetime.time(8, 37, 35)) @@ -667,9 +664,8 @@ class EnumSetTest( 'mysql_enum', self.metadata, Column('e1', e1), Column('e2', e2, nullable=False), - Column( - 'e2generic', - Enum("a", "b", validate_strings=True), nullable=False), + Column('e2generic', + Enum("a", "b", validate_strings=True), nullable=False), Column('e3', e3), Column('e4', e4, nullable=False), @@ -1075,7 +1071,8 @@ class EnumSetTest( t.create() with testing.db.connect() as conn: - conn.execute(t.insert(), {"e1": "nonexistent", "e2": "nonexistent"}) + conn.execute(t.insert(), + {"e1": "nonexistent", "e2": "nonexistent"}) conn.execute(t.insert(), {"e1": "", "e2": ""}) conn.execute(t.insert(), {"e1": "two", "e2": "two"}) conn.execute(t.insert(), {"e1": None, "e2": None}) @@ -1091,4 +1088,3 @@ class EnumSetTest( def colspec(c): return testing.db.dialect.ddl_compiler( testing.db.dialect, None).get_column_specification(c) - diff --git a/test/dialect/postgresql/test_compiler.py b/test/dialect/postgresql/test_compiler.py index 160641855..eddf20877 100644 --- a/test/dialect/postgresql/test_compiler.py +++ b/test/dialect/postgresql/test_compiler.py @@ -17,6 +17,7 @@ from sqlalchemy.sql import util as sql_util from sqlalchemy.util import u, OrderedDict from sqlalchemy.dialects.postgresql import aggregate_order_by, insert + class SequenceTest(fixtures.TestBase, AssertsCompiledSQL): __prefer__ = 'postgresql' @@ -414,9 +415,14 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): m = MetaData() tbl = Table('testtbl', m, Column('data', String)) - idx1 = Index('test_idx1', tbl.c.data) - idx2 = Index('test_idx2', tbl.c.data, postgresql_tablespace='sometablespace') - idx3 = Index('test_idx3', tbl.c.data, postgresql_tablespace='another table space') + idx1 = Index('test_idx1', + tbl.c.data) + idx2 = Index('test_idx2', + tbl.c.data, + postgresql_tablespace='sometablespace') + idx3 = Index('test_idx3', + tbl.c.data, + postgresql_tablespace='another table space') self.assert_compile(schema.CreateIndex(idx1), 'CREATE INDEX test_idx1 ON testtbl ' @@ -437,13 +443,12 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): m = MetaData() tbl = Table('testtbl', m, Column('data', String)) - idx1 = Index( - 'test_idx1', - tbl.c.data, - postgresql_using='btree', - postgresql_tablespace='atablespace', - postgresql_with={"fillfactor": 60}, - postgresql_where=and_(tbl.c.data > 5, tbl.c.data < 10)) + idx1 = Index('test_idx1', + tbl.c.data, + postgresql_using='btree', + postgresql_tablespace='atablespace', + postgresql_with={"fillfactor": 60}, + postgresql_where=and_(tbl.c.data > 5, tbl.c.data < 10)) self.assert_compile(schema.CreateIndex(idx1), 'CREATE INDEX test_idx1 ON testtbl ' @@ -928,8 +933,11 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): postgresql.array([1, 2]) == [3, 4, 5], "ARRAY[%(param_1)s, %(param_2)s] = " "ARRAY[%(param_3)s, %(param_4)s, %(param_5)s]", - checkparams={'param_5': 5, 'param_4': 4, 'param_1': 1, - 'param_3': 3, 'param_2': 2} + checkparams={'param_5': 5, + 'param_4': 4, + 'param_1': 1, + 'param_3': 3, + 'param_2': 2} ) @@ -1053,7 +1061,8 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): stmt2 = sql_util.ClauseAdapter(a1).traverse(stmt) self.assert_compile( stmt2, - "SELECT array_agg(foo.a ORDER BY foo.b DESC) AS array_agg_1 FROM table1 AS foo" + "SELECT array_agg(foo.a ORDER BY foo.b DESC) AS array_agg_1 " + "FROM table1 AS foo" ) diff --git a/test/dialect/postgresql/test_dialect.py b/test/dialect/postgresql/test_dialect.py index 9f7af8638..ba68faf16 100644 --- a/test/dialect/postgresql/test_dialect.py +++ b/test/dialect/postgresql/test_dialect.py @@ -236,8 +236,8 @@ class MiscTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL): r = t.insert().execute(user_name='user', user_password='lala') assert r.inserted_primary_key == [1] - l = t.select().execute().fetchall() - assert l == [(1, 'user', 'lala')] + result = t.select().execute().fetchall() + assert result == [(1, 'user', 'lala')] finally: testing.db.execute('drop table speedy_users') diff --git a/test/dialect/postgresql/test_on_conflict.py b/test/dialect/postgresql/test_on_conflict.py index 0e1dea06a..7c83f2826 100644 --- a/test/dialect/postgresql/test_on_conflict.py +++ b/test/dialect/postgresql/test_on_conflict.py @@ -511,4 +511,3 @@ class OnConflictTest(fixtures.TablesTest): conn.scalar(sql.select([bind_targets.c.data])), "new updated data processed" ) - diff --git a/test/dialect/postgresql/test_reflection.py b/test/dialect/postgresql/test_reflection.py index 5f9e6df9a..1d5648e57 100644 --- a/test/dialect/postgresql/test_reflection.py +++ b/test/dialect/postgresql/test_reflection.py @@ -706,8 +706,8 @@ class ReflectionTest(fixtures.TestBase): tmp.sort() r1, r2 = [idx[1] for idx in tmp] assert r1.name == 'idx2' - assert r1.unique == True - assert r2.unique == False + assert r1.unique is True + assert r2.unique is False assert [t2.c.id] == r1.columns assert [t2.c.name] == r2.columns diff --git a/test/dialect/postgresql/test_types.py b/test/dialect/postgresql/test_types.py index c4106c68e..ddc121a8b 100644 --- a/test/dialect/postgresql/test_types.py +++ b/test/dialect/postgresql/test_types.py @@ -548,7 +548,7 @@ class NumericInterpretationTest(fixtures.TestBase): psycopg2, psycopg2cffi, base dialects = (pg8000.dialect(), pygresql.dialect(), - psycopg2.dialect(), psycopg2cffi.dialect()) + psycopg2.dialect(), psycopg2cffi.dialect()) for dialect in dialects: typ = Numeric().dialect_impl(dialect) for code in base._INT_TYPES + base._FLOAT_TYPES + \ @@ -1089,10 +1089,11 @@ class ArrayRoundTripTest(fixtures.TablesTest, AssertsExecutionResults): def test_array_comparison(self): arrtable = self.tables.arrtable arrtable.insert().execute(id=5, intarr=[1, 2, 3], - strarr=[util.u('abc'), util.u('def')]) - results = select([arrtable.c.id]).\ - where(arrtable.c.intarr < [4, 5, 6]).execute()\ - .fetchall() + strarr=[util.u('abc'), util.u('def')]) + results = select([arrtable.c.id])\ + .where(arrtable.c.intarr < [4, 5, 6])\ + .execute()\ + .fetchall() eq_(len(results), 1) eq_(results[0][0], 5) @@ -1819,7 +1820,7 @@ class HStoreTest(AssertsCompiledSQL, fixtures.TestBase): def test_where_getitem(self): self._test_where( - self.hashcol['bar'] == None, + self.hashcol['bar'] == None, # noqa "(test_table.hash -> %(hash_1)s) IS NULL" ) @@ -1908,7 +1909,7 @@ class HStoreTest(AssertsCompiledSQL, fixtures.TestBase): def test_cols_against_is(self): self._test_cols( - self.hashcol['foo'] != None, + self.hashcol['foo'] != None, # noqa "(test_table.hash -> %(hash_1)s) IS NOT NULL AS anon_1" ) @@ -2445,13 +2446,13 @@ class JSONTest(AssertsCompiledSQL, fixtures.TestBase): # do anything def test_where_getitem(self): self._test_where( - self.jsoncol['bar'] == None, + self.jsoncol['bar'] == None, # noqa "(test_table.test_column -> %(test_column_1)s) IS NULL" ) def test_where_path(self): self._test_where( - self.jsoncol[("foo", 1)] == None, + self.jsoncol[("foo", 1)] == None, # noqa "(test_table.test_column #> %(test_column_1)s) IS NULL" ) @@ -2490,7 +2491,7 @@ class JSONTest(AssertsCompiledSQL, fixtures.TestBase): def test_where_getitem_as_text(self): self._test_where( - self.jsoncol['bar'].astext == None, + self.jsoncol['bar'].astext == None, # noqa "(test_table.test_column ->> %(test_column_1)s) IS NULL" ) @@ -2510,7 +2511,7 @@ class JSONTest(AssertsCompiledSQL, fixtures.TestBase): def test_where_path_as_text(self): self._test_where( - self.jsoncol[("foo", 1)].astext == None, + self.jsoncol[("foo", 1)].astext == None, # noqa "(test_table.test_column #>> %(test_column_1)s) IS NULL" ) diff --git a/test/dialect/test_firebird.py b/test/dialect/test_firebird.py index 86464c8cb..ff8c16eb1 100644 --- a/test/dialect/test_firebird.py +++ b/test/dialect/test_firebird.py @@ -4,13 +4,17 @@ from sqlalchemy.databases import firebird from sqlalchemy.exc import ProgrammingError from sqlalchemy.sql import table, column from sqlalchemy import types as sqltypes -from sqlalchemy.testing import fixtures, AssertsExecutionResults, AssertsCompiledSQL +from sqlalchemy.testing import (fixtures, + AssertsExecutionResults, + AssertsCompiledSQL) from sqlalchemy import testing from sqlalchemy.testing import engines from sqlalchemy import String, VARCHAR, NVARCHAR, Unicode, Integer,\ func, insert, update, MetaData, select, Table, Column, text,\ Sequence, Float from sqlalchemy import schema +from sqlalchemy.testing.mock import Mock, call + class DomainReflectionTest(fixtures.TestBase, AssertsExecutionResults): "Test Firebird domains" @@ -29,7 +33,7 @@ class DomainReflectionTest(fixtures.TestBase, AssertsExecutionResults): con.execute('CREATE DOMAIN img_domain AS BLOB SUB_TYPE ' 'BINARY') except ProgrammingError as e: - if not 'attempt to store duplicate value' in str(e): + if 'attempt to store duplicate value' not in str(e): raise e con.execute('''CREATE GENERATOR gen_testtable_id''') con.execute('''CREATE TABLE testtable (question int_domain, @@ -75,8 +79,8 @@ class DomainReflectionTest(fixtures.TestBase, AssertsExecutionResults): 'dt', 'redundant', ]), - "Columns of reflected table didn't equal expected " - "columns") + "Columns of reflected table didn't equal expected " + "columns") eq_(table.c.question.primary_key, True) # disabled per http://www.sqlalchemy.org/trac/ticket/1660 @@ -106,7 +110,8 @@ class BuggyDomainReflectionTest(fixtures.TestBase, AssertsExecutionResults): __only_on__ = 'firebird' # NB: spacing and newlines are *significant* here! - # PS: this test is superfluous on recent FB, where the issue 356 is probably fixed... + # PS: this test is superfluous on recent FB, where the issue 356 is + # probably fixed... AUTOINC_DM = """\ CREATE DOMAIN AUTOINC_DM @@ -268,11 +273,10 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): exc.CompileError, "VARCHAR requires a length on dialect firebird", type_.compile, - dialect=firebird.dialect()) + dialect=firebird.dialect()) t1 = Table('sometable', MetaData(), - Column('somecolumn', type_) - ) + Column('somecolumn', type_)) assert_raises_message( exc.CompileError, r"\(in table 'sometable', column 'somecolumn'\)\: " @@ -286,8 +290,10 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): self.assert_compile(func.current_time(), 'CURRENT_TIME') self.assert_compile(func.foo(), 'foo') m = MetaData() - t = Table('sometable', m, Column('col1', Integer), Column('col2' - , Integer)) + t = Table('sometable', + m, + Column('col1', Integer), + Column('col2', Integer)) self.assert_compile(select([func.max(t.c.col1)]), 'SELECT max(sometable.col1) AS max_1 FROM ' 'sometable') @@ -300,11 +306,12 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): 'SUBSTRING(:substring_1 FROM :substring_2)') def test_update_returning(self): - table1 = table('mytable', column('myid', Integer), column('name' - , String(128)), column('description', - String(128))) - u = update(table1, values=dict(name='foo' - )).returning(table1.c.myid, table1.c.name) + table1 = table('mytable', + column('myid', Integer), + column('name', String(128)), + column('description', String(128))) + u = update(table1, values=dict(name='foo'))\ + .returning(table1.c.myid, table1.c.name) self.assert_compile(u, 'UPDATE mytable SET name=:name RETURNING ' 'mytable.myid, mytable.name') @@ -313,18 +320,19 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): 'UPDATE mytable SET name=:name RETURNING ' 'mytable.myid, mytable.name, ' 'mytable.description') - u = update(table1, values=dict(name='foo' - )).returning(func.length(table1.c.name)) + u = update(table1, values=dict(name='foo')) \ + .returning(func.length(table1.c.name)) self.assert_compile(u, 'UPDATE mytable SET name=:name RETURNING ' 'char_length(mytable.name) AS length_1') def test_insert_returning(self): - table1 = table('mytable', column('myid', Integer), column('name' - , String(128)), column('description', - String(128))) - i = insert(table1, values=dict(name='foo' - )).returning(table1.c.myid, table1.c.name) + table1 = table('mytable', + column('myid', Integer), + column('name', String(128)), + column('description', String(128))) + i = insert(table1, values=dict(name='foo'))\ + .returning(table1.c.myid, table1.c.name) self.assert_compile(i, 'INSERT INTO mytable (name) VALUES (:name) ' 'RETURNING mytable.myid, mytable.name') @@ -333,8 +341,8 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): 'INSERT INTO mytable (name) VALUES (:name) ' 'RETURNING mytable.myid, mytable.name, ' 'mytable.description') - i = insert(table1, values=dict(name='foo' - )).returning(func.length(table1.c.name)) + i = insert(table1, values=dict(name='foo'))\ + .returning(func.length(table1.c.name)) self.assert_compile(i, 'INSERT INTO mytable (name) VALUES (:name) ' 'RETURNING char_length(mytable.name) AS ' @@ -361,6 +369,8 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): column("$somecol"), '"$somecol"' ) + + class TypesTest(fixtures.TestBase): __only_on__ = 'firebird' @@ -368,13 +378,12 @@ class TypesTest(fixtures.TestBase): def test_infinite_float(self): metadata = self.metadata t = Table('t', metadata, - Column('data', Float) - ) + Column('data', Float)) metadata.create_all() t.insert().execute(data=float('inf')) eq_(t.select().execute().fetchall(), - [(float('inf'),)] - ) + [(float('inf'),)]) + class MiscTest(fixtures.TestBase): @@ -390,9 +399,10 @@ class MiscTest(fixtures.TestBase): # string length the UDF was declared to accept). This test # checks that at least it works ok in other cases. - t = Table('t1', metadata, Column('id', Integer, - Sequence('t1idseq'), primary_key=True), Column('name' - , String(10))) + t = Table('t1', + metadata, + Column('id', Integer, Sequence('t1idseq'), primary_key=True), + Column('name', String(10))) metadata.create_all() t.insert(values=dict(name='dante')).execute() t.insert(values=dict(name='alighieri')).execute() @@ -421,10 +431,8 @@ class MiscTest(fixtures.TestBase): metadata.bind = engine t = Table('t1', metadata, Column('data', String(10))) metadata.create_all() - r = t.insert().execute({'data': 'd1'}, {'data': 'd2'}, {'data' - : 'd3'}) - r = t.update().where(t.c.data == 'd2').values(data='d3' - ).execute() + r = t.insert().execute({'data': 'd1'}, {'data': 'd2'}, {'data': 'd3'}) + r = t.update().where(t.c.data == 'd2').values(data='d3').execute() eq_(r.rowcount, 1) r = t.delete().where(t.c.data == 'd3').execute() eq_(r.rowcount, 2) @@ -432,14 +440,11 @@ class MiscTest(fixtures.TestBase): t.delete().execution_options(enable_rowcount=False).execute() eq_(r.rowcount, -1) engine.dispose() - engine = engines.testing_engine(options={'enable_rowcount' - : False}) + engine = engines.testing_engine(options={'enable_rowcount': False}) assert not engine.dialect.supports_sane_rowcount metadata.bind = engine - r = t.insert().execute({'data': 'd1'}, {'data': 'd2'}, {'data' - : 'd3'}) - r = t.update().where(t.c.data == 'd2').values(data='d3' - ).execute() + r = t.insert().execute({'data': 'd1'}, {'data': 'd2'}, {'data': 'd3'}) + r = t.update().where(t.c.data == 'd2').values(data='d3').execute() eq_(r.rowcount, -1) r = t.delete().where(t.c.data == 'd3').execute() eq_(r.rowcount, -1) @@ -457,8 +462,6 @@ class MiscTest(fixtures.TestBase): 'hello % world'): eq_(testing.db.scalar(expr), result) -from sqlalchemy.testing.mock import Mock, call - class ArgumentTest(fixtures.TestBase): def _dbapi(self): @@ -480,8 +483,7 @@ class ArgumentTest(fixtures.TestBase): _initialize=False ) ) - engine = engines.testing_engine("firebird+%s://" % type_, - options=kw) + engine = engines.testing_engine("firebird+%s://" % type_, options=kw) return engine def test_retaining_flag_default_kinterbasdb(self): @@ -508,7 +510,6 @@ class ArgumentTest(fixtures.TestBase): engine = self._engine("fdb", retaining=False) self._assert_retaining(engine, False) - def _assert_retaining(self, engine, flag): conn = engine.connect() trans = conn.begin() @@ -524,8 +525,3 @@ class ArgumentTest(fixtures.TestBase): engine.dialect.dbapi.connect.return_value.rollback.mock_calls, [call(flag)] ) - - - - - diff --git a/test/dialect/test_mxodbc.py b/test/dialect/test_mxodbc.py index e46de9149..e5f9f330f 100644 --- a/test/dialect/test_mxodbc.py +++ b/test/dialect/test_mxodbc.py @@ -4,18 +4,14 @@ from sqlalchemy.testing import engines from sqlalchemy.testing import fixtures from sqlalchemy.testing.mock import Mock + def mock_dbapi(): return Mock(paramstyle='qmark', connect=Mock( - return_value=Mock( - cursor=Mock( - return_value=Mock( - description=None, - rowcount=None) - ) - ) - ) - ) + return_value=Mock( + cursor=Mock(return_value=Mock(description=None, + rowcount=None))))) + class MxODBCTest(fixtures.TestBase): @@ -24,7 +20,8 @@ class MxODBCTest(fixtures.TestBase): dbapi = mock_dbapi() engine = engines.testing_engine('mssql+mxodbc://localhost', - options={'module': dbapi, '_initialize': False}) + options={'module': dbapi, + '_initialize': False}) conn = engine.connect() # crud: uses execute @@ -36,14 +33,11 @@ class MxODBCTest(fixtures.TestBase): conn.execute(t1.select()) # manual flagging - conn.execution_options(native_odbc_execute=True).\ - execute(t1.select()) - conn.execution_options(native_odbc_execute=False).\ - execute(t1.insert().values(c1='foo')) + conn.execution_options(native_odbc_execute=True).execute(t1.select()) + conn.execution_options(native_odbc_execute=False)\ + .execute(t1.insert().values(c1='foo')) - eq_( - [c[2] for c in + eq_([c[2] for c in dbapi.connect.return_value.cursor.return_value.execute.mock_calls], [{'direct': True}, {'direct': True}, {'direct': True}, - {'direct': True}, {'direct': False}, {'direct': True}] - ) + {'direct': True}, {'direct': False}, {'direct': True}]) diff --git a/test/dialect/test_oracle.py b/test/dialect/test_oracle.py index 79d37237c..bb3f56568 100644 --- a/test/dialect/test_oracle.py +++ b/test/dialect/test_oracle.py @@ -6,7 +6,9 @@ from sqlalchemy import * from sqlalchemy import types as sqltypes, exc, schema from sqlalchemy.sql import table, column from sqlalchemy.sql.elements import quoted_name -from sqlalchemy.testing import fixtures, AssertsExecutionResults, AssertsCompiledSQL +from sqlalchemy.testing import (fixtures, + AssertsExecutionResults, + AssertsCompiledSQL) from sqlalchemy import testing from sqlalchemy.util import u, b from sqlalchemy import util @@ -21,6 +23,7 @@ import os from sqlalchemy import sql from sqlalchemy.testing.mock import Mock + class OutParamTest(fixtures.TestBase, AssertsExecutionResults): __only_on__ = 'oracle+cx_oracle' __backend__ = True @@ -41,11 +44,12 @@ class OutParamTest(fixtures.TestBase, AssertsExecutionResults): def test_out_params(self): result = testing.db.execute(text('begin foo(:x_in, :x_out, :y_out, ' - ':z_out); end;', - bindparams=[bindparam('x_in', Float), - outparam('x_out', Integer), - outparam('y_out', Float), - outparam('z_out', String)]), x_in=5) + ':z_out); end;', + bindparams=[bindparam('x_in', Float), + outparam('x_out', Integer), + outparam('y_out', Float), + outparam('z_out', String)]), + x_in=5) eq_(result.out_parameters, {'x_out': 10, 'y_out': 75, 'z_out': None}) assert isinstance(result.out_parameters['x_out'], int) @@ -54,6 +58,7 @@ class OutParamTest(fixtures.TestBase, AssertsExecutionResults): def teardown_class(cls): testing.db.execute("DROP PROCEDURE foo") + class CXOracleArgsTest(fixtures.TestBase): __only_on__ = 'oracle+cx_oracle' __backend__ = True @@ -72,9 +77,10 @@ class CXOracleArgsTest(fixtures.TestBase): def test_exclude_inputsizes_custom(self): import cx_Oracle dialect = cx_oracle.dialect(dbapi=cx_Oracle, - exclude_setinputsizes=('NCLOB',)) + exclude_setinputsizes=('NCLOB',)) eq_(dialect.exclude_setinputsizes, set([cx_Oracle.NCLOB])) + class QuotedBindRoundTripTest(fixtures.TestBase): __only_on__ = 'oracle' @@ -86,13 +92,12 @@ class QuotedBindRoundTripTest(fixtures.TestBase): metadata = self.metadata table = Table("t1", metadata, - Column("option", Integer), - Column("plain", Integer, quote=True), - # test that quote works for a reserved word - # that the dialect isn't aware of when quote - # is set - Column("union", Integer, quote=True) - ) + Column("option", Integer), + Column("plain", Integer, quote=True), + # test that quote works for a reserved word + # that the dialect isn't aware of when quote + # is set + Column("union", Integer, quote=True)) metadata.create_all() table.insert().execute( @@ -131,8 +136,9 @@ class QuotedBindRoundTripTest(fixtures.TestBase): testing.db.scalar(t.select()), 10 ) + class CompileTest(fixtures.TestBase, AssertsCompiledSQL): - __dialect__ = "oracle" #oracle.dialect() + __dialect__ = "oracle" # oracle.dialect() def test_true_false(self): self.assert_compile( @@ -161,8 +167,8 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): s = select([s.c.col1, s.c.col2]) self.assert_compile(s, "SELECT col1, col2 FROM (SELECT " - "sometable.col1 AS col1, sometable.col2 " - "AS col2 FROM sometable)") + "sometable.col1 AS col1, sometable.col2 " + "AS col2 FROM sometable)") def test_bindparam_quote(self): """test that bound parameters take on quoting for reserved words, @@ -178,7 +184,8 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): ) t = Table("s", MetaData(), Column('plain', Integer, quote=True)) self.assert_compile( - t.insert().values(plain=5), 'INSERT INTO s ("plain") VALUES (:"plain")' + t.insert().values(plain=5), + 'INSERT INTO s ("plain") VALUES (:"plain")' ) self.assert_compile( t.update().values(plain=5), 'UPDATE s SET "plain"=:"plain"' @@ -253,8 +260,8 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): 'FROM (SELECT col1, col2, ROWNUM AS ora_rn ' 'FROM (SELECT sometable.col1 AS col1, ' 'sometable.col2 AS col2 FROM sometable) ' - 'WHERE ROWNUM <= :param_1 + :param_2) WHERE ora_rn > ' - ':param_2)', + 'WHERE ROWNUM <= :param_1 + :param_2) ' + 'WHERE ora_rn > :param_2)', checkparams={'param_1': 10, 'param_2': 20}) self.assert_compile(s2, @@ -262,8 +269,8 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): 'FROM (SELECT col1, col2, ROWNUM AS ora_rn ' 'FROM (SELECT sometable.col1 AS col1, ' 'sometable.col2 AS col2 FROM sometable) ' - 'WHERE ROWNUM <= :param_1 + :param_2) WHERE ora_rn > ' - ':param_2)') + 'WHERE ROWNUM <= :param_1 + :param_2) ' + 'WHERE ora_rn > :param_2)') c = s2.compile(dialect=oracle.OracleDialect()) eq_(len(c._result_columns), 2) assert s.c.col1 in set(c._create_result_map()['col1'][1]) @@ -303,7 +310,7 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): def test_for_update(self): table1 = table('mytable', - column('myid'), column('name'), column('description')) + column('myid'), column('name'), column('description')) self.assert_compile( table1.select(table1.c.myid == 7).with_for_update(), @@ -311,9 +318,12 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): "FROM mytable WHERE mytable.myid = :myid_1 FOR UPDATE") self.assert_compile( - table1.select(table1.c.myid == 7).with_for_update(of=table1.c.myid), + table1 + .select(table1.c.myid == 7) + .with_for_update(of=table1.c.myid), "SELECT mytable.myid, mytable.name, mytable.description " - "FROM mytable WHERE mytable.myid = :myid_1 FOR UPDATE OF mytable.myid") + "FROM mytable WHERE mytable.myid = :myid_1 " + "FOR UPDATE OF mytable.myid") self.assert_compile( table1.select(table1.c.myid == 7).with_for_update(nowait=True), @@ -321,44 +331,48 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): "FROM mytable WHERE mytable.myid = :myid_1 FOR UPDATE NOWAIT") self.assert_compile( - table1.select(table1.c.myid == 7). - with_for_update(nowait=True, of=table1.c.myid), + table1 + .select(table1.c.myid == 7) + .with_for_update(nowait=True, of=table1.c.myid), "SELECT mytable.myid, mytable.name, mytable.description " "FROM mytable WHERE mytable.myid = :myid_1 " "FOR UPDATE OF mytable.myid NOWAIT") self.assert_compile( - table1.select(table1.c.myid == 7). - with_for_update(nowait=True, of=[table1.c.myid, table1.c.name]), + table1 + .select(table1.c.myid == 7) + .with_for_update(nowait=True, of=[table1.c.myid, table1.c.name]), "SELECT mytable.myid, mytable.name, mytable.description " "FROM mytable WHERE mytable.myid = :myid_1 FOR UPDATE OF " "mytable.myid, mytable.name NOWAIT") self.assert_compile( - table1.select(table1.c.myid == 7). - with_for_update(skip_locked=True, of=[table1.c.myid, table1.c.name]), + table1.select(table1.c.myid == 7) + .with_for_update(skip_locked=True, + of=[table1.c.myid, table1.c.name]), "SELECT mytable.myid, mytable.name, mytable.description " "FROM mytable WHERE mytable.myid = :myid_1 FOR UPDATE OF " "mytable.myid, mytable.name SKIP LOCKED") # key_share has no effect self.assert_compile( - table1.select(table1.c.myid == 7). - with_for_update(key_share=True), + table1.select(table1.c.myid == 7).with_for_update(key_share=True), "SELECT mytable.myid, mytable.name, mytable.description " "FROM mytable WHERE mytable.myid = :myid_1 FOR UPDATE") # read has no effect self.assert_compile( - table1.select(table1.c.myid == 7). - with_for_update(read=True, key_share=True), + table1 + .select(table1.c.myid == 7) + .with_for_update(read=True, key_share=True), "SELECT mytable.myid, mytable.name, mytable.description " "FROM mytable WHERE mytable.myid = :myid_1 FOR UPDATE") ta = table1.alias() self.assert_compile( - ta.select(ta.c.myid == 7). - with_for_update(of=[ta.c.myid, ta.c.name]), + ta + .select(ta.c.myid == 7) + .with_for_update(of=[ta.c.myid, ta.c.name]), "SELECT mytable_1.myid, mytable_1.name, mytable_1.description " "FROM mytable mytable_1 " "WHERE mytable_1.myid = :myid_1 FOR UPDATE OF " @@ -451,46 +465,52 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): t = table('sometable', column('col1'), column('col2')) dialect = oracle.OracleDialect(use_binds_for_limits=False) - self.assert_compile(select([t]).limit(10), - "SELECT col1, col2 FROM (SELECT sometable.col1 AS col1, " - "sometable.col2 AS col2 FROM sometable) WHERE ROWNUM <= 10", - dialect=dialect) + self.assert_compile( + select([t]).limit(10), + "SELECT col1, col2 FROM (SELECT sometable.col1 AS col1, " + "sometable.col2 AS col2 FROM sometable) WHERE ROWNUM <= 10", + dialect=dialect) - self.assert_compile(select([t]).offset(10), - "SELECT col1, col2 FROM (SELECT col1, col2, ROWNUM AS ora_rn " - "FROM (SELECT sometable.col1 AS col1, sometable.col2 AS col2 " - "FROM sometable)) WHERE ora_rn > 10", - dialect=dialect) + self.assert_compile( + select([t]).offset(10), + "SELECT col1, col2 FROM (SELECT col1, col2, ROWNUM AS ora_rn " + "FROM (SELECT sometable.col1 AS col1, sometable.col2 AS col2 " + "FROM sometable)) WHERE ora_rn > 10", + dialect=dialect) - self.assert_compile(select([t]).limit(10).offset(10), - "SELECT col1, col2 FROM (SELECT col1, col2, ROWNUM AS ora_rn " - "FROM (SELECT sometable.col1 AS col1, sometable.col2 AS col2 " - "FROM sometable) WHERE ROWNUM <= 20) WHERE ora_rn > 10", - dialect=dialect) + self.assert_compile( + select([t]).limit(10).offset(10), + "SELECT col1, col2 FROM (SELECT col1, col2, ROWNUM AS ora_rn " + "FROM (SELECT sometable.col1 AS col1, sometable.col2 AS col2 " + "FROM sometable) WHERE ROWNUM <= 20) WHERE ora_rn > 10", + dialect=dialect) def test_use_binds_for_limits_enabled(self): t = table('sometable', column('col1'), column('col2')) dialect = oracle.OracleDialect(use_binds_for_limits=True) - self.assert_compile(select([t]).limit(10), - "SELECT col1, col2 FROM (SELECT sometable.col1 AS col1, " - "sometable.col2 AS col2 FROM sometable) WHERE ROWNUM " - "<= :param_1", - dialect=dialect) - - self.assert_compile(select([t]).offset(10), - "SELECT col1, col2 FROM (SELECT col1, col2, ROWNUM AS ora_rn " - "FROM (SELECT sometable.col1 AS col1, sometable.col2 AS col2 " - "FROM sometable)) WHERE ora_rn > :param_1", - dialect=dialect) - - self.assert_compile(select([t]).limit(10).offset(10), - "SELECT col1, col2 FROM (SELECT col1, col2, ROWNUM AS ora_rn " - "FROM (SELECT sometable.col1 AS col1, sometable.col2 AS col2 " - "FROM sometable) WHERE ROWNUM <= :param_1 + :param_2) " - "WHERE ora_rn > :param_2", - dialect=dialect, - checkparams={'param_1': 10, 'param_2': 10}) + self.assert_compile( + select([t]).limit(10), + "SELECT col1, col2 FROM (SELECT sometable.col1 AS col1, " + "sometable.col2 AS col2 FROM sometable) WHERE ROWNUM " + "<= :param_1", + dialect=dialect) + + self.assert_compile( + select([t]).offset(10), + "SELECT col1, col2 FROM (SELECT col1, col2, ROWNUM AS ora_rn " + "FROM (SELECT sometable.col1 AS col1, sometable.col2 AS col2 " + "FROM sometable)) WHERE ora_rn > :param_1", + dialect=dialect) + + self.assert_compile( + select([t]).limit(10).offset(10), + "SELECT col1, col2 FROM (SELECT col1, col2, ROWNUM AS ora_rn " + "FROM (SELECT sometable.col1 AS col1, sometable.col2 AS col2 " + "FROM sometable) WHERE ROWNUM <= :param_1 + :param_2) " + "WHERE ora_rn > :param_2", + dialect=dialect, + checkparams={'param_1': 10, 'param_2': 10}) def test_long_labels(self): dialect = default.DefaultDialect() @@ -510,11 +530,9 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): m, Column('id', Integer, primary_key=True), Column('thirty_characters_table_id', - Integer, - ForeignKey('thirty_characters_table_xxxxxx.id'), - primary_key=True - ) - ) + Integer, + ForeignKey('thirty_characters_table_xxxxxx.id'), + primary_key=True)) anon = a_table.alias() self.assert_compile(select([other_table, @@ -551,10 +569,9 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): def test_outer_join(self): table1 = table('mytable', - column('myid', Integer), - column('name', String), - column('description', String), - ) + column('myid', Integer), + column('name', String), + column('description', String)) table2 = table( 'myothertable', @@ -568,11 +585,14 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): column('otherstuff', String), ) - query = select([table1, table2], or_(table1.c.name == 'fred', - table1.c.myid == 10, table2.c.othername != 'jack', - text('EXISTS (select yay from foo where boo = lar)') - ), from_obj=[outerjoin(table1, table2, - table1.c.myid == table2.c.otherid)]) + query = select([table1, table2], + or_(table1.c.name == 'fred', + table1.c.myid == 10, table2.c.othername != 'jack', + text('EXISTS (select yay from foo where boo = lar)') + ), + from_obj=[outerjoin(table1, + table2, + table1.c.myid == table2.c.otherid)]) self.assert_compile(query, 'SELECT mytable.myid, mytable.name, ' 'mytable.description, myothertable.otherid,' @@ -584,9 +604,9 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): 'lar)) AND mytable.myid = ' 'myothertable.otherid(+)', dialect=oracle.OracleDialect(use_ansi=False)) - query = table1.outerjoin(table2, table1.c.myid - == table2.c.otherid).outerjoin(table3, - table3.c.userid == table2.c.otherid) + query = table1.outerjoin(table2, + table1.c.myid == table2.c.otherid) \ + .outerjoin(table3, table3.c.userid == table2.c.otherid) self.assert_compile(query.select(), 'SELECT mytable.myid, mytable.name, ' 'mytable.description, myothertable.otherid,' @@ -607,9 +627,9 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): 'myothertable.otherid AND mytable.myid = ' 'myothertable.otherid(+)', dialect=oracle.dialect(use_ansi=False)) - query = table1.join(table2, table1.c.myid - == table2.c.otherid).join(table3, - table3.c.userid == table2.c.otherid) + query = table1.join(table2, + table1.c.myid == table2.c.otherid) \ + .join(table3, table3.c.userid == table2.c.otherid) self.assert_compile(query.select(), 'SELECT mytable.myid, mytable.name, ' 'mytable.description, myothertable.otherid,' @@ -620,11 +640,11 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): 'myothertable.otherid AND mytable.myid = ' 'myothertable.otherid', dialect=oracle.dialect(use_ansi=False)) - query = table1.join(table2, table1.c.myid - == table2.c.otherid).outerjoin(table3, - table3.c.userid == table2.c.otherid) + query = table1.join(table2, + table1.c.myid == table2.c.otherid) \ + .outerjoin(table3, table3.c.userid == table2.c.otherid) self.assert_compile(query.select().order_by(table1.c.name). - limit(10).offset(5), + limit(10).offset(5), 'SELECT myid, name, description, otherid, ' 'othername, userid, otherstuff FROM ' '(SELECT myid, name, description, otherid, ' @@ -638,16 +658,17 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): 'mytable, myothertable, thirdtable WHERE ' 'thirdtable.userid(+) = ' 'myothertable.otherid AND mytable.myid = ' - 'myothertable.otherid ORDER BY ' - 'mytable.name) WHERE ROWNUM <= :param_1 + :param_2) ' + 'myothertable.otherid ORDER BY mytable.name) ' + 'WHERE ROWNUM <= :param_1 + :param_2) ' 'WHERE ora_rn > :param_2', checkparams={'param_1': 10, 'param_2': 5}, dialect=oracle.dialect(use_ansi=False)) - subq = select([table1]).select_from(table1.outerjoin(table2, - table1.c.myid == table2.c.otherid)).alias() - q = select([table3]).select_from(table3.outerjoin(subq, - table3.c.userid == subq.c.myid)) + subq = select([table1]).select_from( + table1.outerjoin(table2, table1.c.myid == table2.c.otherid)) \ + .alias() + q = select([table3]).select_from( + table3.outerjoin(subq, table3.c.userid == subq.c.myid)) self.assert_compile(q, 'SELECT thirdtable.userid, ' @@ -676,8 +697,8 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): 'SELECT mytable.name FROM mytable WHERE ' 'mytable.name = :name_1', dialect=oracle.dialect(use_ansi=False)) - subq = select([table3.c.otherstuff]).where(table3.c.otherstuff - == table1.c.name).label('bar') + subq = select([table3.c.otherstuff]) \ + .where(table3.c.otherstuff == table1.c.name).label('bar') q = select([table1.c.name, subq]) self.assert_compile(q, 'SELECT mytable.name, (SELECT ' @@ -700,7 +721,6 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): dialect=oracle.OracleDialect(use_ansi=False) ) - j = a.outerjoin(b.join(c, b.c.b == c.c.c), a.c.a == b.c.b) self.assert_compile( @@ -719,7 +739,6 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): dialect=oracle.OracleDialect(use_ansi=False) ) - def test_alias_outer_join(self): address_types = table('address_types', column('id'), column('name')) @@ -727,11 +746,13 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): column('address_type_id'), column('email_address')) at_alias = address_types.alias() - s = select([at_alias, - addresses]).select_from(addresses.outerjoin(at_alias, - addresses.c.address_type_id - == at_alias.c.id)).where(addresses.c.user_id - == 7).order_by(addresses.c.id, address_types.c.id) + s = select([at_alias, addresses]) \ + .select_from( + addresses.outerjoin( + at_alias, + addresses.c.address_type_id == at_alias.c.id)) \ + .where(addresses.c.user_id == 7) \ + .order_by(addresses.c.id, address_types.c.id) self.assert_compile(s, 'SELECT address_types_1.id, ' 'address_types_1.name, addresses.id, ' @@ -748,25 +769,23 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): self.assert_compile( t1.insert().values(c1=1).returning(t1.c.c2, t1.c.c3), "INSERT INTO t1 (c1) VALUES (:c1) RETURNING " - "t1.c2, t1.c3 INTO :ret_0, :ret_1" - ) + "t1.c2, t1.c3 INTO :ret_0, :ret_1") def test_returning_insert_functional(self): - t1 = table('t1', column('c1'), column('c2', String()), column('c3', String())) + t1 = table('t1', + column('c1'), + column('c2', String()), + column('c3', String())) fn = func.lower(t1.c.c2, type_=String()) stmt = t1.insert().values(c1=1).returning(fn, t1.c.c3) compiled = stmt.compile(dialect=oracle.dialect()) - eq_( - compiled._create_result_map(), + eq_(compiled._create_result_map(), {'ret_1': ('ret_1', (t1.c.c3, 'c3', 'c3'), t1.c.c3.type), - 'ret_0': ('ret_0', (fn, 'lower', None), fn.type)} - - ) + 'ret_0': ('ret_0', (fn, 'lower', None), fn.type)}) self.assert_compile( stmt, "INSERT INTO t1 (c1) VALUES (:c1) RETURNING " - "lower(t1.c2), t1.c3 INTO :ret_0, :ret_1" - ) + "lower(t1.c2), t1.c3 INTO :ret_0, :ret_1") def test_returning_insert_labeled(self): t1 = table('t1', column('c1'), column('c2'), column('c3')) @@ -774,8 +793,7 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): t1.insert().values(c1=1).returning( t1.c.c2.label('c2_l'), t1.c.c3.label('c3_l')), "INSERT INTO t1 (c1) VALUES (:c1) RETURNING " - "t1.c2, t1.c3 INTO :ret_0, :ret_1" - ) + "t1.c2, t1.c3 INTO :ret_0, :ret_1") def test_compound(self): t1 = table('t1', column('c1'), column('c2'), column('c3')) @@ -803,9 +821,8 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): def test_create_index_alt_schema(self): m = MetaData() t1 = Table('foo', m, - Column('x', Integer), - schema="alt_schema" - ) + Column('x', Integer), + schema="alt_schema") self.assert_compile( schema.CreateIndex(Index("bar", t1.c.x)), "CREATE INDEX alt_schema.bar ON alt_schema.foo (x)" @@ -814,8 +831,7 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): def test_create_index_expr(self): m = MetaData() t1 = Table('foo', m, - Column('x', Integer) - ) + Column('x', Integer)) self.assert_compile( schema.CreateIndex(Index("bar", t1.c.x > 5)), "CREATE INDEX bar ON foo (x > 5)" @@ -837,7 +853,6 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): "foo (x INTEGER) ON COMMIT PRESERVE ROWS" ) - def test_create_table_compress(self): m = MetaData() tbl1 = Table('testtbl1', m, Column('data', Integer), @@ -881,7 +896,6 @@ class CompatFlagsTest(fixtures.TestBase, AssertsCompiledSQL): dialect._get_default_schema_name = Mock() return dialect - def test_ora8_flags(self): dialect = self._dialect((8, 2, 5)) @@ -899,12 +913,10 @@ class CompatFlagsTest(fixtures.TestBase, AssertsCompiledSQL): self.assert_compile(Unicode(50), "VARCHAR2(50)", dialect=dialect) self.assert_compile(UnicodeText(), "CLOB", dialect=dialect) - dialect = self._dialect((8, 2, 5), implicit_returning=True) dialect.initialize(testing.db.connect()) assert dialect.implicit_returning - def test_default_flags(self): """test with no initialization or server version info""" @@ -993,14 +1005,14 @@ drop synonym %(test_schema)s.local_table; schema = testing.db.dialect.default_schema_name meta = self.metadata parent = Table('parent', meta, - Column('pid', Integer, primary_key=True), - schema=schema - ) + Column('pid', Integer, primary_key=True), + schema=schema) child = Table('child', meta, - Column('cid', Integer, primary_key=True), - Column('pid', Integer, ForeignKey('%s.parent.pid' % schema)), - schema=schema - ) + Column('cid', Integer, primary_key=True), + Column('pid', + Integer, + ForeignKey('%s.parent.pid' % schema)), + schema=schema) meta.create_all() parent.insert().execute({'pid': 1}) child.insert().execute({'cid': 1, 'pid': 1}) @@ -1008,12 +1020,14 @@ drop synonym %(test_schema)s.local_table; def test_reflect_alt_table_owner_local_synonym(self): meta = MetaData(testing.db) - parent = Table('%s_pt' % testing.config.test_schema, meta, autoload=True, - oracle_resolve_synonyms=True) + parent = Table('%s_pt' % testing.config.test_schema, + meta, + autoload=True, + oracle_resolve_synonyms=True) self.assert_compile(parent.select(), - "SELECT %(test_schema)s_pt.id, " - "%(test_schema)s_pt.data FROM %(test_schema)s_pt" - % {"test_schema": testing.config.test_schema}) + "SELECT %(test_schema)s_pt.id, " + "%(test_schema)s_pt.data FROM %(test_schema)s_pt" + % {"test_schema": testing.config.test_schema}) select([parent]).execute().fetchall() def test_reflect_alt_synonym_owner_local_table(self): @@ -1033,13 +1047,12 @@ drop synonym %(test_schema)s.local_table; @testing.provide_metadata def test_create_same_names_implicit_schema(self): meta = self.metadata - parent = Table('parent', meta, - Column('pid', Integer, primary_key=True), - ) + parent = Table('parent', + meta, + Column('pid', Integer, primary_key=True)) child = Table('child', meta, - Column('cid', Integer, primary_key=True), - Column('pid', Integer, ForeignKey('parent.pid')), - ) + Column('cid', Integer, primary_key=True), + Column('pid', Integer, ForeignKey('parent.pid'))) meta.create_all() parent.insert().execute({'pid': 1}) child.insert().execute({'cid': 1, 'pid': 1}) @@ -1153,16 +1166,17 @@ class ConstraintTest(fixtures.TablesTest): def test_oracle_has_no_on_update_cascade(self): bar = Table('bar', self.metadata, - Column('id', Integer, primary_key=True), - Column('foo_id', Integer, - ForeignKey('foo.id', onupdate='CASCADE'))) + Column('id', Integer, primary_key=True), + Column('foo_id', + Integer, + ForeignKey('foo.id', onupdate='CASCADE'))) assert_raises(exc.SAWarning, bar.create) bat = Table('bat', self.metadata, - Column('id', Integer, primary_key=True), - Column('foo_id', Integer), + Column('id', Integer, primary_key=True), + Column('foo_id', Integer), ForeignKeyConstraint(['foo_id'], ['foo.id'], - onupdate='CASCADE')) + onupdate='CASCADE')) assert_raises(exc.SAWarning, bat.create) @@ -1176,9 +1190,8 @@ class TwoPhaseTest(fixtures.TablesTest): @classmethod def define_tables(cls, metadata): Table('datatable', metadata, - Column('id', Integer, primary_key=True), - Column('data', String(50)) - ) + Column('id', Integer, primary_key=True), + Column('data', String(50))) def _connection(self): conn = testing.db.connect() @@ -1190,6 +1203,7 @@ class TwoPhaseTest(fixtures.TablesTest): testing.db.scalar("select count(*) from datatable"), rows ) + def test_twophase_prepare_false(self): conn = self._connection() for i in range(2): @@ -1205,7 +1219,7 @@ class TwoPhaseTest(fixtures.TablesTest): for i in range(2): trans = conn.begin_twophase() conn.execute("insert into datatable (id, data) " - "values (%s, 'somedata')" % i) + "values (%s, 'somedata')" % i) trans.prepare() trans.commit() conn.close() @@ -1215,12 +1229,12 @@ class TwoPhaseTest(fixtures.TablesTest): conn = self._connection() trans = conn.begin_twophase() conn.execute("insert into datatable (id, data) " - "values (%s, 'somedata')" % 1) + "values (%s, 'somedata')" % 1) trans.rollback() trans = conn.begin_twophase() conn.execute("insert into datatable (id, data) " - "values (%s, 'somedata')" % 1) + "values (%s, 'somedata')" % 1) trans.prepare() trans.commit() @@ -1231,11 +1245,12 @@ class TwoPhaseTest(fixtures.TablesTest): conn = self._connection() trans = conn.begin_twophase() conn.execute("insert into datatable (id, data) " - "values (%s, 'somedata')" % 1) + "values (%s, 'somedata')" % 1) trans.commit() conn.close() self._assert_data(1) + class DialectTypesTest(fixtures.TestBase, AssertsCompiledSQL): __dialect__ = oracle.OracleDialect() @@ -1264,7 +1279,6 @@ class DialectTypesTest(fixtures.TestBase, AssertsCompiledSQL): 'STRING' ) - def test_long(self): self.assert_compile(oracle.LONG(), "LONG") @@ -1328,7 +1342,7 @@ class DialectTypesTest(fixtures.TestBase, AssertsCompiledSQL): (oracle.INTERVAL(second_precision=5), 'INTERVAL DAY TO SECOND(5)'), (oracle.INTERVAL(day_precision=2, - second_precision=5), + second_precision=5), 'INTERVAL DAY(2) TO SECOND(5)')]: self.assert_compile(type_, expected) @@ -1338,14 +1352,12 @@ class TypesTest(fixtures.TestBase): __dialect__ = oracle.OracleDialect() __backend__ = True - @testing.fails_on('+zxjdbc', 'zxjdbc lacks the FIXED_CHAR dbapi type') def test_fixed_char(self): m = MetaData(testing.db) t = Table('t1', m, - Column('id', Integer, primary_key=True), - Column('data', CHAR(30), nullable=False) - ) + Column('id', Integer, primary_key=True), + Column('data', CHAR(30), nullable=False)) t.create() try: @@ -1389,9 +1401,7 @@ class TypesTest(fixtures.TestBase): @testing.provide_metadata def test_rowid(self): metadata = self.metadata - t = Table('t1', metadata, - Column('x', Integer) - ) + t = Table('t1', metadata, Column('x', Integer)) t.create() t.insert().execute(x=5) s1 = select([t]) @@ -1402,11 +1412,9 @@ class TypesTest(fixtures.TestBase): # as cx_oracle just treats it as a string, # but we want to make sure the ROWID works... rowid_col = column('rowid', oracle.ROWID) - s3 = select([t.c.x, rowid_col]).\ - where(rowid_col == cast(rowid, oracle.ROWID)) - eq_(s3.select().execute().fetchall(), - [(5, rowid)] - ) + s3 = select([t.c.x, rowid_col]) \ + .where(rowid_col == cast(rowid, oracle.ROWID)) + eq_(s3.select().execute().fetchall(), [(5, rowid)]) @testing.fails_on('+zxjdbc', 'Not yet known how to pass values of the ' @@ -1421,8 +1429,7 @@ class TypesTest(fixtures.TestBase): oracle.INTERVAL(day_precision=3))) metadata.create_all() interval_table.insert().\ - execute(day_interval=datetime.timedelta(days=35, - seconds=5743)) + execute(day_interval=datetime.timedelta(days=35, seconds=5743)) row = interval_table.select().execute().first() eq_(row['day_interval'], datetime.timedelta(days=35, seconds=5743)) @@ -1431,16 +1438,14 @@ class TypesTest(fixtures.TestBase): def test_numerics(self): m = self.metadata t1 = Table('t1', m, - Column('intcol', Integer), - Column('numericcol', Numeric(precision=9, scale=2)), - Column('floatcol1', Float()), - Column('floatcol2', FLOAT()), - Column('doubleprec', oracle.DOUBLE_PRECISION), - Column('numbercol1', oracle.NUMBER(9)), - Column('numbercol2', oracle.NUMBER(9, 3)), - Column('numbercol3', oracle.NUMBER), - - ) + Column('intcol', Integer), + Column('numericcol', Numeric(precision=9, scale=2)), + Column('floatcol1', Float()), + Column('floatcol2', FLOAT()), + Column('doubleprec', oracle.DOUBLE_PRECISION), + Column('numbercol1', oracle.NUMBER(9)), + Column('numbercol2', oracle.NUMBER(9, 3)), + Column('numbercol3', oracle.NUMBER)) t1.create() t1.insert().execute( intcol=1, @@ -1507,12 +1512,11 @@ class TypesTest(fixtures.TestBase): # this test requires cx_oracle 5 foo = Table('foo', metadata, - Column('idata', Integer), - Column('ndata', Numeric(20, 2)), - Column('ndata2', Numeric(20, 2)), - Column('nidata', Numeric(5, 0)), - Column('fdata', Float()), - ) + Column('idata', Integer), + Column('ndata', Numeric(20, 2)), + Column('ndata2', Numeric(20, 2)), + Column('nidata', Numeric(5, 0)), + Column('fdata', Float())) foo.create() foo.insert().execute({ @@ -1525,7 +1529,6 @@ class TypesTest(fixtures.TestBase): stmt = "SELECT idata, ndata, ndata2, nidata, fdata FROM foo" - row = testing.db.execute(stmt).fetchall()[0] eq_( [type(x) for x in row], @@ -1549,11 +1552,11 @@ class TypesTest(fixtures.TestBase): SELECT (SELECT (SELECT idata FROM foo) FROM DUAL) AS idata, (SELECT CAST((SELECT ndata FROM foo) AS NUMERIC(20, 2)) FROM DUAL) - AS ndata, - (SELECT CAST((SELECT ndata2 FROM foo) AS NUMERIC(20, 2)) FROM DUAL) - AS ndata2, - (SELECT CAST((SELECT nidata FROM foo) AS NUMERIC(5, 0)) FROM DUAL) - AS nidata, + AS ndata, + (SELECT CAST((SELECT ndata2 FROM foo) AS NUMERIC(20, 2)) FROM DUAL) + AS ndata2, + (SELECT CAST((SELECT nidata FROM foo) AS NUMERIC(5, 0)) FROM DUAL) + AS nidata, (SELECT CAST((SELECT fdata FROM foo) AS FLOAT) FROM DUAL) AS fdata FROM dual """ @@ -1568,13 +1571,12 @@ class TypesTest(fixtures.TestBase): ) row = testing.db.execute(text(stmt, - typemap={ - 'idata': Integer(), - 'ndata': Numeric(20, 2), - 'ndata2': Numeric(20, 2), - 'nidata': Numeric(5, 0), - 'fdata': Float() - })).fetchall()[0] + typemap={ + 'idata': Integer(), + 'ndata': Numeric(20, 2), + 'ndata2': Numeric(20, 2), + 'nidata': Numeric(5, 0), + 'fdata': Float()})).fetchall()[0] eq_( [type(x) for x in row], [int, decimal.Decimal, decimal.Decimal, decimal.Decimal, float] @@ -1619,13 +1621,13 @@ class TypesTest(fixtures.TestBase): ) row = testing.db.execute(text(stmt, - typemap={ - 'anon_1_idata': Integer(), - 'anon_1_ndata': Numeric(20, 2), - 'anon_1_ndata2': Numeric(20, 2), - 'anon_1_nidata': Numeric(5, 0), - 'anon_1_fdata': Float() - })).fetchall()[0] + typemap={ + 'anon_1_idata': Integer(), + 'anon_1_ndata': Numeric(20, 2), + 'anon_1_ndata2': Numeric(20, 2), + 'anon_1_nidata': Numeric(5, 0), + 'anon_1_fdata': Float() + })).fetchall()[0] eq_( [type(x) for x in row], [int, decimal.Decimal, decimal.Decimal, decimal.Decimal, float] @@ -1636,14 +1638,15 @@ class TypesTest(fixtures.TestBase): decimal.Decimal('53'), 45.683920000000001) ) - row = testing.db.execute(text(stmt, - typemap={ - 'anon_1_idata': Integer(), - 'anon_1_ndata': Numeric(20, 2, asdecimal=False), - 'anon_1_ndata2': Numeric(20, 2, asdecimal=False), - 'anon_1_nidata': Numeric(5, 0, asdecimal=False), - 'anon_1_fdata': Float(asdecimal=True) - })).fetchall()[0] + row = testing.db.execute(text( + stmt, + typemap={ + 'anon_1_idata': Integer(), + 'anon_1_ndata': Numeric(20, 2, asdecimal=False), + 'anon_1_ndata2': Numeric(20, 2, asdecimal=False), + 'anon_1_nidata': Numeric(5, 0, asdecimal=False), + 'anon_1_fdata': Float(asdecimal=True) + })).fetchall()[0] eq_( [type(x) for x in row], [int, float, float, float, decimal.Decimal] @@ -1653,7 +1656,6 @@ class TypesTest(fixtures.TestBase): (5, 45.6, 45, 53, decimal.Decimal('45.68392')) ) - @testing.provide_metadata def test_reflect_dates(self): metadata = self.metadata @@ -1682,10 +1684,9 @@ class TypesTest(fixtures.TestBase): def test_reflect_all_types_schema(self): types_table = Table('all_types', MetaData(testing.db), - Column('owner', String(30), primary_key=True), - Column('type_name', String(30), primary_key=True), - autoload=True, oracle_resolve_synonyms=True - ) + Column('owner', String(30), primary_key=True), + Column('type_name', String(30), primary_key=True), + autoload=True, oracle_resolve_synonyms=True) for row in types_table.select().execute().fetchall(): [row[k] for k in row.keys()] @@ -1693,9 +1694,8 @@ class TypesTest(fixtures.TestBase): def test_raw_roundtrip(self): metadata = self.metadata raw_table = Table('raw', metadata, - Column('id', Integer, primary_key=True), - Column('data', oracle.RAW(35)) - ) + Column('id', Integer, primary_key=True), + Column('data', oracle.RAW(35))) metadata.create_all() testing.db.execute(raw_table.insert(), id=1, data=b("ABCDEF")) eq_( @@ -1706,9 +1706,7 @@ class TypesTest(fixtures.TestBase): @testing.provide_metadata def test_reflect_nvarchar(self): metadata = self.metadata - Table('t', metadata, - Column('data', sqltypes.NVARCHAR(255)) - ) + Table('t', metadata, Column('data', sqltypes.NVARCHAR(255))) metadata.create_all() m2 = MetaData(testing.db) t2 = Table('t', m2, autoload=True) @@ -1727,15 +1725,13 @@ class TypesTest(fixtures.TestBase): eq_(res, data) assert isinstance(res, util.text_type) - @testing.provide_metadata def test_char_length(self): metadata = self.metadata t1 = Table('t1', metadata, - Column("c1", VARCHAR(50)), - Column("c2", NVARCHAR(250)), - Column("c3", CHAR(200)) - ) + Column("c1", VARCHAR(50)), + Column("c2", NVARCHAR(250)), + Column("c3", CHAR(200))) t1.create() m2 = MetaData(testing.db) t2 = Table('t1', m2, autoload=True) @@ -1747,9 +1743,7 @@ class TypesTest(fixtures.TestBase): def test_long_type(self): metadata = self.metadata - t = Table('t', metadata, - Column('data', oracle.LONG) - ) + t = Table('t', metadata, Column('data', oracle.LONG)) metadata.create_all(testing.db) testing.db.execute(t.insert(), data='xyz') eq_( @@ -1777,21 +1771,28 @@ class TypesTest(fixtures.TestBase): def test_lobs_without_convert(self): engine = testing_engine(options=dict(auto_convert_lobs=False)) metadata = MetaData() - t = Table("z_test", metadata, Column('id', Integer, primary_key=True), - Column('data', Text), Column('bindata', LargeBinary)) + t = Table("z_test", + metadata, + Column('id', Integer, primary_key=True), + Column('data', Text), + Column('bindata', LargeBinary)) t.create(engine) try: - engine.execute(t.insert(), id=1, - data='this is text', - bindata=b('this is binary')) + engine.execute(t.insert(), + id=1, + data='this is text', + bindata=b('this is binary')) row = engine.execute(t.select()).first() eq_(row['data'].read(), 'this is text') eq_(row['bindata'].read(), b('this is binary')) finally: t.drop(engine) + class EuroNumericTest(fixtures.TestBase): - """test the numeric output_type_handler when using non-US locale for NLS_LANG.""" + """ + test the numeric output_type_handler when using non-US locale for NLS_LANG. + """ __only_on__ = 'oracle+cx_oracle' __backend__ = True @@ -1813,11 +1814,11 @@ class EuroNumericTest(fixtures.TestBase): ("SELECT 0.1 FROM DUAL", decimal.Decimal("0.1"), {}), ("SELECT 15 FROM DUAL", 15, {}), ("SELECT CAST(15 AS NUMERIC(3, 1)) FROM DUAL", - decimal.Decimal("15"), {}), + decimal.Decimal("15"), {}), ("SELECT CAST(0.1 AS NUMERIC(5, 2)) FROM DUAL", - decimal.Decimal("0.1"), {}), + decimal.Decimal("0.1"), {}), ("SELECT :num FROM DUAL", decimal.Decimal("2.5"), - {'num': decimal.Decimal("2.5")}) + {'num': decimal.Decimal("2.5")}) ]: test_exp = self.engine.scalar(stmt, **kw) eq_( @@ -1833,8 +1834,12 @@ class SystemTableTablenamesTest(fixtures.TestBase): def setup(self): testing.db.execute("create table my_table (id integer)") - testing.db.execute("create global temporary table my_temp_table (id integer)") - testing.db.execute("create table foo_table (id integer) tablespace SYSTEM") + testing.db.execute( + "create global temporary table my_temp_table (id integer)" + ) + testing.db.execute( + "create table foo_table (id integer) tablespace SYSTEM" + ) def teardown(self): testing.db.execute("drop table my_temp_table") @@ -1857,7 +1862,8 @@ class SystemTableTablenamesTest(fixtures.TestBase): engine = testing_engine(options={"exclude_tablespaces": ["FOO"]}) insp = inspect(engine) eq_( - set(insp.get_table_names()).intersection(["my_table", "foo_table"]), + set(insp.get_table_names()).intersection(["my_table", + "foo_table"]), set(["my_table", "foo_table"]) ) @@ -1894,6 +1900,7 @@ class DontReflectIOTTest(fixtures.TestBase): set(['admin_docindex']) ) + class BufferedColumnTest(fixtures.TestBase, AssertsCompiledSQL): __only_on__ = 'oracle' __backend__ = True @@ -1903,9 +1910,8 @@ class BufferedColumnTest(fixtures.TestBase, AssertsCompiledSQL): global binary_table, stream, meta meta = MetaData(testing.db) binary_table = Table('binary_table', meta, - Column('id', Integer, primary_key=True), - Column('data', LargeBinary) - ) + Column('id', Integer, primary_key=True), + Column('data', LargeBinary)) meta.create_all() stream = os.path.join( os.path.dirname(__file__), "..", @@ -1929,9 +1935,10 @@ class BufferedColumnTest(fixtures.TestBase, AssertsCompiledSQL): def test_fetch_single_arraysize(self): eng = testing_engine(options={'arraysize': 1}) result = eng.execute(binary_table.select(). - order_by(binary_table.c.id)).fetchall() + order_by(binary_table.c.id)).fetchall() eq_(result, [(i, stream) for i in range(1, 11)]) + class UnsupportedIndexReflectTest(fixtures.TestBase): __only_on__ = 'oracle' __backend__ = True @@ -1941,8 +1948,7 @@ class UnsupportedIndexReflectTest(fixtures.TestBase): def test_reflect_functional_index(self): metadata = self.metadata Table('test_index_reflect', metadata, - Column('data', String(20), primary_key=True) - ) + Column('data', String(20), primary_key=True)) metadata.create_all() testing.db.execute('CREATE INDEX DATA_IDX ON ' @@ -1958,7 +1964,7 @@ def all_tables_compression_missing(): "select * from v$version"): return True return False - except: + except Exception: return True @@ -1969,7 +1975,7 @@ def all_tables_compress_for_missing(): "select * from v$version"): return True return False - except: + except Exception: return True @@ -2009,8 +2015,6 @@ class TableReflectionTest(fixtures.TestBase): assert tbl.dialect_options['oracle']['compress'] == "OLTP" - - class RoundTripIndexTest(fixtures.TestBase): __only_on__ = 'oracle' __backend__ = True @@ -2020,12 +2024,14 @@ class RoundTripIndexTest(fixtures.TestBase): metadata = self.metadata table = Table("sometable", metadata, - Column("id_a", Unicode(255), primary_key=True), - Column("id_b", Unicode(255), primary_key=True, unique=True), - Column("group", Unicode(255), primary_key=True), - Column("col", Unicode(255)), - UniqueConstraint('col', 'group'), - ) + Column("id_a", Unicode(255), primary_key=True), + Column("id_b", + Unicode(255), + primary_key=True, + unique=True), + Column("group", Unicode(255), primary_key=True), + Column("col", Unicode(255)), + UniqueConstraint('col', 'group')) # "group" is a keyword, so lower case normalind = Index('tableind', table.c.id_b, table.c.group) @@ -2044,8 +2050,9 @@ class RoundTripIndexTest(fixtures.TestBase): inspect.reflect() def obj_definition(obj): - return obj.__class__, tuple([c.name for c in - obj.columns]), getattr(obj, 'unique', None) + return (obj.__class__, + tuple([c.name for c in obj.columns]), + getattr(obj, 'unique', None)) # find what the primary k constraint name should be primaryconsname = testing.db.scalar( @@ -2091,6 +2098,7 @@ class RoundTripIndexTest(fixtures.TestBase): eq_(len(reflectedtable.constraints), 1) eq_(len(reflectedtable.indexes), 5) + class SequenceTest(fixtures.TestBase, AssertsCompiledSQL): def test_basic(self): @@ -2131,9 +2139,10 @@ class ExecuteTest(fixtures.TestBase): # oracle can't actually do the ROWNUM thing with FOR UPDATE # very well. - t = Table('t1', metadata, Column('id', Integer, primary_key=True), - Column('data', Integer) - ) + t = Table('t1', + metadata, + Column('id', Integer, primary_key=True), + Column('data', Integer)) metadata.create_all() t.insert().execute( @@ -2168,8 +2177,9 @@ class UnicodeSchemaTest(fixtures.TestBase): def test_quoted_column_non_unicode(self): metadata = self.metadata table = Table("atable", metadata, - Column("_underscorecolumn", Unicode(255), primary_key=True), - ) + Column("_underscorecolumn", + Unicode(255), + primary_key=True)) metadata.create_all() table.insert().execute( @@ -2184,8 +2194,7 @@ class UnicodeSchemaTest(fixtures.TestBase): def test_quoted_column_unicode(self): metadata = self.metadata table = Table("atable", metadata, - Column(u("méil"), Unicode(255), primary_key=True), - ) + Column(u("méil"), Unicode(255), primary_key=True)) metadata.create_all() table.insert().execute( @@ -2208,14 +2217,13 @@ class DBLinkReflectionTest(fixtures.TestBase): cls.dblink = config.file_config.get('sqla_testing', 'oracle_db_link') # note that the synonym here is still not totally functional - # when accessing via a different username as we do with the multiprocess - # test suite, so testing here is minimal + # when accessing via a different username as we do with the + # multiprocess test suite, so testing here is minimal with testing.db.connect() as conn: - conn.execute( - "create table test_table " - "(id integer primary key, data varchar2(50))") + conn.execute("create table test_table " + "(id integer primary key, data varchar2(50))") conn.execute("create synonym test_table_syn " - "for test_table@%s" % cls.dblink) + "for test_table@%s" % cls.dblink) @classmethod def teardown_class(cls): @@ -2228,7 +2236,7 @@ class DBLinkReflectionTest(fixtures.TestBase): m = MetaData() t = Table('test_table_syn', m, autoload=True, - autoload_with=testing.db, oracle_resolve_synonyms=True) + autoload_with=testing.db, oracle_resolve_synonyms=True) eq_(list(t.c.keys()), ['id', 'data']) eq_(list(t.primary_key), [t.c.id]) diff --git a/test/dialect/test_pyodbc.py b/test/dialect/test_pyodbc.py index f5850fe98..8d817fdb1 100644 --- a/test/dialect/test_pyodbc.py +++ b/test/dialect/test_pyodbc.py @@ -2,6 +2,7 @@ from sqlalchemy.testing import eq_ from sqlalchemy.connectors import pyodbc from sqlalchemy.testing import fixtures + class PyODBCTest(fixtures.TestBase): def test_pyodbc_version(self): connector = pyodbc.PyODBCConnector() @@ -14,4 +15,4 @@ class PyODBCTest(fixtures.TestBase): eq_( connector._parse_dbapi_version(vers), expected - )
\ No newline at end of file + ) diff --git a/test/dialect/test_sqlite.py b/test/dialect/test_sqlite.py index 6735ae30d..4c462aed1 100644 --- a/test/dialect/test_sqlite.py +++ b/test/dialect/test_sqlite.py @@ -490,14 +490,14 @@ class DialectTest(fixtures.TestBase, AssertsExecutionResults): # """id""" integer NOT NULL PRIMARY KEY, # """aid""" integer NULL # REFERENCES """a""" ("""id""") - #) - #''') + # ) + # ''') table1 = Table(r'"a"', metadata, autoload=True) assert '"id"' in table1.c - #table2 = Table(r'"b"', metadata, autoload=True) - #j = table1.join(table2) + # table2 = Table(r'"b"', metadata, autoload=True) + # j = table1.join(table2) # assert j.onclause.compare(table1.c['"id"'] # == table2.c['"aid"']) @@ -751,8 +751,7 @@ class SQLTest(fixtures.TestBase, AssertsCompiledSQL): # test quoting and all that idx2 = Index('test_idx2', tbl.c.data, - sqlite_where=and_(tbl.c.data > 'a', tbl.c.data - < "b's")) + sqlite_where=and_(tbl.c.data > 'a', tbl.c.data < "b's")) self.assert_compile(schema.CreateIndex(idx), 'CREATE INDEX test_idx1 ON testtbl (data) ' 'WHERE data > 5 AND data < 10', @@ -898,9 +897,10 @@ def full_text_search_missing(): testing.db.execute('CREATE VIRTUAL TABLE t using FTS3;') testing.db.execute('DROP TABLE t;') return False - except: + except Exception: return True + metadata = cattable = matchtable = None @@ -952,7 +952,8 @@ class MatchTest(fixtures.TestBase, AssertsCompiledSQL): def test_expression(self): self.assert_compile(matchtable.c.title.match('somstr'), - 'matchtable.title MATCH ?', dialect=sqlite.dialect()) + 'matchtable.title MATCH ?', + dialect=sqlite.dialect()) def test_simple_match(self): results = \ @@ -1198,7 +1199,7 @@ class ConstraintReflectionTest(fixtures.TestBase): "d", "d1", "d2", "c", "b", "a1", "a2"]: try: conn.execute("drop table %s" % name) - except: + except Exception: pass def test_legacy_quoted_identifiers_unit(self): diff --git a/test/dialect/test_suite.py b/test/dialect/test_suite.py index 3820a7721..d4e753656 100644 --- a/test/dialect/test_suite.py +++ b/test/dialect/test_suite.py @@ -1,3 +1 @@ from sqlalchemy.testing.suite import * - - diff --git a/test/dialect/test_sybase.py b/test/dialect/test_sybase.py index d8f7d3aae..fee72a5b7 100644 --- a/test/dialect/test_sybase.py +++ b/test/dialect/test_sybase.py @@ -32,5 +32,3 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): "Sybase ASE does not support OFFSET", stmt.compile, dialect=self.__dialect__ ) - - |