diff options
Diffstat (limited to 'test/dialect')
| -rw-r--r-- | test/dialect/test_firebird.py | 6 | ||||
| -rw-r--r-- | test/dialect/test_mssql.py | 86 | ||||
| -rw-r--r-- | test/dialect/test_mxodbc.py | 2 | ||||
| -rw-r--r-- | test/dialect/test_mysql.py | 52 | ||||
| -rw-r--r-- | test/dialect/test_oracle.py | 148 | ||||
| -rw-r--r-- | test/dialect/test_postgresql.py | 112 | ||||
| -rw-r--r-- | test/dialect/test_sqlite.py | 20 |
7 files changed, 213 insertions, 213 deletions
diff --git a/test/dialect/test_firebird.py b/test/dialect/test_firebird.py index 814c267b5..ce708936b 100644 --- a/test/dialect/test_firebird.py +++ b/test/dialect/test_firebird.py @@ -169,7 +169,7 @@ CREATE DOMAIN DOM_ID INTEGER NOT NULL CREATE TABLE A ( ID DOM_ID /* INTEGER NOT NULL */ DEFAULT 0 ) """ - + # the 'default' keyword is lower case here TABLE_B = """\ CREATE TABLE B ( @@ -222,14 +222,14 @@ ID DOM_ID /* INTEGER NOT NULL */ default 0 ) table_a = Table('a', metadata, autoload=True) eq_(table_a.c.id.server_default.arg.text, "0") - + def test_lowercase_default_name(self): metadata = MetaData(testing.db) table_b = Table('b', metadata, autoload=True) eq_(table_b.c.id.server_default.arg.text, "0") - + class CompileTest(TestBase, AssertsCompiledSQL): diff --git a/test/dialect/test_mssql.py b/test/dialect/test_mssql.py index 68203cfea..6cc327151 100644 --- a/test/dialect/test_mssql.py +++ b/test/dialect/test_mssql.py @@ -31,15 +31,15 @@ class CompileTest(TestBase, AssertsCompiledSQL): 'UPDATE sometable SET somecolumn=:somecolum' 'n WHERE sometable.somecolumn = ' ':somecolumn_1', dict(somecolumn=10)) - + # TODO: should this be for *all* MS-SQL dialects ? def test_mxodbc_binds(self): """mxodbc uses MS-SQL native binds, which aren't allowed in various places.""" - + mxodbc_dialect = mxodbc.dialect() t = table('sometable', column('foo')) - + for expr, compile in [ ( select([literal("x"), literal("y")]), @@ -61,7 +61,7 @@ class CompileTest(TestBase, AssertsCompiledSQL): ) ]: self.assert_compile(expr, compile, dialect=mxodbc_dialect) - + def test_in_with_subqueries(self): """Test that when using subqueries in a binary expression the == and != are changed to IN and NOT IN respectively. @@ -151,7 +151,7 @@ class CompileTest(TestBase, AssertsCompiledSQL): 'remotetable_1.value FROM mytable JOIN ' 'remote_owner.remotetable AS remotetable_1 ' 'ON remotetable_1.rem_id = mytable.myid') - + self.assert_compile(select([table4.c.rem_id, table4.c.value]).apply_labels().union(select([table1.c.myid, table1.c.description]).apply_labels()).alias().select(), @@ -163,8 +163,8 @@ class CompileTest(TestBase, AssertsCompiledSQL): "SELECT mytable.myid AS mytable_myid, mytable.description " "AS mytable_description FROM mytable) AS anon_1" ) - - + + def test_delete_schema(self): metadata = MetaData() tbl = Table('test', metadata, Column('id', Integer, @@ -337,9 +337,9 @@ class CompileTest(TestBase, AssertsCompiledSQL): def test_limit_using_top(self): t = table('t', column('x', Integer), column('y', Integer)) - + s = select([t]).where(t.c.x==5).order_by(t.c.y).limit(10) - + self.assert_compile( s, "SELECT TOP 10 t.x, t.y FROM t WHERE t.x = :x_1 ORDER BY t.y", @@ -348,9 +348,9 @@ class CompileTest(TestBase, AssertsCompiledSQL): def test_offset_using_window(self): t = table('t', column('x', Integer), column('y', Integer)) - + s = select([t]).where(t.c.x==5).order_by(t.c.y).offset(20) - + self.assert_compile( s, "SELECT anon_1.x, anon_1.y FROM (SELECT t.x AS x, t.y " @@ -362,9 +362,9 @@ class CompileTest(TestBase, AssertsCompiledSQL): def test_limit_offset_using_window(self): t = table('t', column('x', Integer), column('y', Integer)) - + s = select([t]).where(t.c.x==5).order_by(t.c.y).limit(10).offset(20) - + self.assert_compile( s, "SELECT anon_1.x, anon_1.y " @@ -375,9 +375,9 @@ class CompileTest(TestBase, AssertsCompiledSQL): "WHERE mssql_rn > :mssql_rn_1 AND mssql_rn <= :mssql_rn_2", {u'mssql_rn_1': 20, u'mssql_rn_2': 30, u'x_1': 5} ) - - - + + + class IdentityInsertTest(TestBase, AssertsCompiledSQL): __only_on__ = 'mssql' __dialect__ = mssql.MSDialect() @@ -496,7 +496,7 @@ class ReflectionTest(TestBase, ComparesTables): and table2.c['col1'].default assert sequence.start == 2 assert sequence.increment == 3 - + @testing.emits_warning("Did not recognize") @testing.provide_metadata def test_skip_types(self): @@ -509,14 +509,14 @@ class ReflectionTest(TestBase, ComparesTables): @testing.provide_metadata def test_indexes_cols(self): - + t1 = Table('t', metadata, Column('x', Integer), Column('y', Integer)) Index('foo', t1.c.x, t1.c.y) metadata.create_all() - + m2 = MetaData() t2 = Table('t', m2, autoload=True, autoload_with=testing.db) - + eq_( set(list(t2.indexes)[0].columns), set([t2.c['x'], t2.c.y]) @@ -524,38 +524,38 @@ class ReflectionTest(TestBase, ComparesTables): @testing.provide_metadata def test_indexes_cols_with_commas(self): - + t1 = Table('t', metadata, Column('x, col', Integer, key='x'), Column('y', Integer) ) Index('foo', t1.c.x, t1.c.y) metadata.create_all() - + m2 = MetaData() t2 = Table('t', m2, autoload=True, autoload_with=testing.db) - + eq_( set(list(t2.indexes)[0].columns), set([t2.c['x, col'], t2.c.y]) ) - + @testing.provide_metadata def test_indexes_cols_with_spaces(self): - + t1 = Table('t', metadata, Column('x col', Integer, key='x'), Column('y', Integer)) Index('foo', t1.c.x, t1.c.y) metadata.create_all() - + m2 = MetaData() t2 = Table('t', m2, autoload=True, autoload_with=testing.db) - + eq_( set(list(t2.indexes)[0].columns), set([t2.c['x col'], t2.c.y]) ) - + class QueryUnicodeTest(TestBase): __only_on__ = 'mssql' @@ -587,24 +587,24 @@ class QueryTest(TestBase): def test_fetchid_trigger(self): """ Verify identity return value on inserting to a trigger table. - + MSSQL's OUTPUT INSERTED clause does not work for the case of a table having an identity (autoincrement) primary key column, and which also has a trigger configured to fire upon each insert and subsequently perform an insert into a different table. - + SQLALchemy's MSSQL dialect by default will attempt to use an OUTPUT_INSERTED clause, which in this case will raise the following error: - + ProgrammingError: (ProgrammingError) ('42000', 334, "[Microsoft][SQL Server Native Client 10.0][SQL Server]The target table 't1' of the DML statement cannot have any enabled triggers if the statement contains an OUTPUT clause without INTO clause.", 7748) 'INSERT INTO t1 (descr) OUTPUT inserted.id VALUES (?)' ('hello',) - + This test verifies a workaround, which is to rely on the older SCOPE_IDENTITY() call, which still works for this scenario. To enable the workaround, the Table must be instantiated @@ -759,10 +759,10 @@ class SchemaTest(TestBase): dialect = mssql.dialect() self.ddl_compiler = dialect.ddl_compiler(dialect, schema.CreateTable(t)) - + def _column_spec(self): return self.ddl_compiler.get_column_specification(self.column) - + def test_that_mssql_default_nullability_emits_null(self): eq_("test_column VARCHAR(max) NULL", self._column_spec()) @@ -998,7 +998,7 @@ class ParseConnectTest(TestBase, AssertsCompiledSQL): connection = dialect.create_connect_args(u) eq_([['DRIVER={SQL Server};Server=hostspec;Database=database;UI' 'D=username;PWD=password'], {}], connection) - + def test_pymssql_port_setting(self): dialect = pymssql.dialect() @@ -1017,7 +1017,7 @@ class ParseConnectTest(TestBase, AssertsCompiledSQL): [[], {'host': 'somehost:5000', 'password': 'tiger', 'user': 'scott', 'database': 'test'}], connection ) - + @testing.only_on(['mssql+pyodbc', 'mssql+pymssql'], "FreeTDS specific test") def test_bad_freetds_warning(self): engine = engines.testing_engine() @@ -1040,7 +1040,7 @@ class TypesTest(TestBase, AssertsExecutionResults, ComparesTables): def teardown(self): metadata.drop_all() - + @testing.fails_on_everything_except('mssql+pyodbc', 'this is some pyodbc-specific feature') def test_decimal_notation(self): @@ -1424,7 +1424,7 @@ class TypesTest(TestBase, AssertsExecutionResults, ComparesTables): 'NCHAR(1)'), (mssql.MSNChar, [1], {'collation': 'Latin1_General_CI_AS'}, 'NCHAR(1) COLLATE Latin1_General_CI_AS'), - + (mssql.MSString, [], {}, 'VARCHAR(max)'), (mssql.MSString, [1], {}, @@ -1535,7 +1535,7 @@ class TypesTest(TestBase, AssertsExecutionResults, ComparesTables): elif c.name.startswith('int_n'): assert not c.autoincrement, name assert tbl._autoincrement_column is not c, name - + # mxodbc can't handle scope_identity() with DEFAULT VALUES if testing.db.driver == 'mxodbc': @@ -1548,7 +1548,7 @@ class TypesTest(TestBase, AssertsExecutionResults, ComparesTables): : False}), engines.testing_engine(options={'implicit_returning' : True})] - + for counter, engine in enumerate(eng): engine.execute(tbl.insert()) if 'int_y' in tbl.c: @@ -1563,9 +1563,9 @@ class TypesTest(TestBase, AssertsExecutionResults, ComparesTables): class BinaryTest(TestBase, AssertsExecutionResults): """Test the Binary and VarBinary types""" - + __only_on__ = 'mssql' - + @classmethod def setup_class(cls): global binary_table, MyPickleType @@ -1583,7 +1583,7 @@ class BinaryTest(TestBase, AssertsExecutionResults): value.stuff = 'this is the right stuff' return value - binary_table = Table( + binary_table = Table( 'binary_table', MetaData(testing.db), Column('primary_id', Integer, Sequence('binary_id_seq', diff --git a/test/dialect/test_mxodbc.py b/test/dialect/test_mxodbc.py index 36cfc9b08..58ceaf91c 100644 --- a/test/dialect/test_mxodbc.py +++ b/test/dialect/test_mxodbc.py @@ -10,7 +10,7 @@ class MockDBAPI(object): self.log = [] def connect(self, *args, **kwargs): return MockConnection(self) - + class MockConnection(object): def __init__(self, parent): self.parent = parent diff --git a/test/dialect/test_mysql.py b/test/dialect/test_mysql.py index 7d5cffbe2..2fe9e7533 100644 --- a/test/dialect/test_mysql.py +++ b/test/dialect/test_mysql.py @@ -19,7 +19,7 @@ class TypesTest(TestBase, AssertsExecutionResults, AssertsCompiledSQL): __only_on__ = 'mysql' __dialect__ = mysql.dialect() - + @testing.uses_deprecated('Manually quoting ENUM value literals') def test_basic(self): meta1 = MetaData(testing.db) @@ -485,22 +485,22 @@ class TypesTest(TestBase, AssertsExecutionResults, AssertsCompiledSQL): # if needed, can break out the eq_() just to check for # timestamps that are within a few seconds of "now" # using timedelta. - + now = testing.db.execute("select now()").scalar() - + # TIMESTAMP without NULL inserts current time when passed # NULL. when not passed, generates 0000-00-00 quite # annoyingly. ts_table.insert().execute({'t1':now, 't2':None}) ts_table.insert().execute({'t1':None, 't2':None}) - + eq_( ts_table.select().execute().fetchall(), [(now, now), (None, now)] ) finally: meta.drop_all() - + def test_year(self): """Exercise YEAR.""" @@ -623,7 +623,7 @@ class TypesTest(TestBase, AssertsExecutionResults, AssertsCompiledSQL): assert_raises(exc.SQLError, enum_table.insert().execute, e1=None, e2=None, e3=None, e4=None) - + assert_raises(exc.InvalidRequestError, enum_table.insert().execute, e1='c', e2='c', e2generic='c', e3='c', e4='c', e5='c', e5generic='c', e6='c') @@ -663,7 +663,7 @@ class TypesTest(TestBase, AssertsExecutionResults, AssertsCompiledSQL): eq_(res, expected) enum_table.drop() - + def test_unicode_enum(self): unicode_engine = utf8_engine() metadata = MetaData(unicode_engine) @@ -696,7 +696,7 @@ class TypesTest(TestBase, AssertsExecutionResults, AssertsCompiledSQL): (u'réveillé', u'drôle') #, u'S’il') # eh ? finally: metadata.drop_all() - + def test_enum_compile(self): e1 = Enum('x', 'y', 'z', name='somename') t1 = Table('sometable', MetaData(), Column('somecolumn', e1)) @@ -709,7 +709,7 @@ class TypesTest(TestBase, AssertsExecutionResults, AssertsCompiledSQL): "CREATE TABLE sometable (somecolumn " "VARCHAR(1), CHECK (somecolumn IN ('x', " "'y', 'z')))") - + @testing.exclude('mysql', '<', (4,), "3.23 can't handle an ENUM of ''") @testing.uses_deprecated('Manually quoting ENUM value literals') def test_enum_parse(self): @@ -1041,7 +1041,7 @@ class SQLTest(TestBase, AssertsCompiledSQL): eq_( gen(True, ['high_priority', sql.text('sql_cache')]), 'SELECT high_priority sql_cache DISTINCT q') - + def test_backslash_escaping(self): self.assert_compile( sql.column('foo').like('bar', escape='\\'), @@ -1055,7 +1055,7 @@ class SQLTest(TestBase, AssertsCompiledSQL): "foo LIKE %s ESCAPE '\\'", dialect=dialect ) - + def test_limit(self): t = sql.table('t', sql.column('col1'), sql.column('col2')) @@ -1068,13 +1068,13 @@ class SQLTest(TestBase, AssertsCompiledSQL): select([t]).limit(10), "SELECT t.col1, t.col2 FROM t LIMIT %s", {'param_1':10}) - + self.assert_compile( select([t]).offset(10), "SELECT t.col1, t.col2 FROM t LIMIT %s, 18446744073709551615", {'param_1':10} ) - + def test_varchar_raise(self): for type_ in ( String, @@ -1087,7 +1087,7 @@ class SQLTest(TestBase, AssertsCompiledSQL): ): type_ = sqltypes.to_instance(type_) assert_raises(exc.InvalidRequestError, type_.compile, dialect=mysql.dialect()) - + def test_update_limit(self): t = sql.table('t', sql.column('col1'), sql.column('col2')) @@ -1113,7 +1113,7 @@ class SQLTest(TestBase, AssertsCompiledSQL): def test_sysdate(self): self.assert_compile(func.sysdate(), "SYSDATE()") - + def test_cast(self): t = sql.table('t', sql.column('col')) m = mysql @@ -1205,7 +1205,7 @@ class SQLTest(TestBase, AssertsCompiledSQL): for type_, expected in specs: self.assert_compile(cast(t.c.col, type_), expected) - + def test_no_cast_pre_4(self): self.assert_compile( cast(Column('foo', Integer), String), @@ -1218,7 +1218,7 @@ class SQLTest(TestBase, AssertsCompiledSQL): "foo", dialect=dialect ) - + def test_extract(self): t = sql.table('t', sql.column('col1')) @@ -1231,24 +1231,24 @@ class SQLTest(TestBase, AssertsCompiledSQL): self.assert_compile( select([extract('milliseconds', t.c.col1)]), "SELECT EXTRACT(millisecond FROM t.col1) AS anon_1 FROM t") - + def test_too_long_index(self): exp = 'ix_zyrenian_zyme_zyzzogeton_zyzzogeton_zyrenian_zyme_zyz_5cd2' tname = 'zyrenian_zyme_zyzzogeton_zyzzogeton' cname = 'zyrenian_zyme_zyzzogeton_zo' - + t1 = Table(tname, MetaData(), Column(cname, Integer, index=True), ) ix1 = list(t1.indexes)[0] - + self.assert_compile( schema.CreateIndex(ix1), "CREATE INDEX %s " "ON %s (%s)" % (exp, tname, cname), dialect=mysql.dialect() ) - + def test_innodb_autoincrement(self): t1 = Table('sometable', MetaData(), Column('assigned_id', Integer(), primary_key=True, autoincrement=False), @@ -1273,7 +1273,7 @@ class SQLTest(TestBase, AssertsCompiledSQL): class SQLModeDetectionTest(TestBase): __only_on__ = 'mysql' - + def _options(self, modes): def connect(con, record): cursor = con.cursor() @@ -1286,7 +1286,7 @@ class SQLModeDetectionTest(TestBase): ] }) return e - + def test_backslash_escapes(self): engine = self._options(['NO_BACKSLASH_ESCAPES']) c = engine.connect() @@ -1314,7 +1314,7 @@ class SQLModeDetectionTest(TestBase): assert not engine.dialect._backslash_escapes c.close() engine.dispose() - + class RawReflectionTest(TestBase): def setup(self): dialect = mysql.dialect() @@ -1346,7 +1346,7 @@ class ExecutionTest(TestBase): meta.reflect(cx) eq_(cx.dialect._connection_charset, charset) cx.close() - + def test_sysdate(self): d = testing.db.scalar(func.sysdate()) assert isinstance(d, datetime.datetime) @@ -1402,7 +1402,7 @@ class MatchTest(TestBase, AssertsCompiledSQL): self.assert_compile( matchtable.c.title.match('somstr'), "MATCH (matchtable.title) AGAINST (%s IN BOOLEAN MODE)" % format) - + @testing.fails_on('mysql+mysqldb', 'uses format') @testing.fails_on('mysql+oursql', 'uses format') @testing.fails_on('mysql+pyodbc', 'uses format') diff --git a/test/dialect/test_oracle.py b/test/dialect/test_oracle.py index d842c7fc2..a4c3f0a83 100644 --- a/test/dialect/test_oracle.py +++ b/test/dialect/test_oracle.py @@ -120,7 +120,7 @@ class CompileTest(TestBase, AssertsCompiledSQL): 'col2 FROM sometable ORDER BY ' 'sometable.col2) WHERE ROWNUM <= :ROWNUM_1 ' 'FOR UPDATE') - + s = select([t], for_update=True).limit(10).offset(20).order_by(t.c.col2) self.assert_compile(s, @@ -131,14 +131,14 @@ class CompileTest(TestBase, AssertsCompiledSQL): 'sometable.col2) WHERE ROWNUM <= ' ':ROWNUM_1) WHERE ora_rn > :ora_rn_1 FOR ' 'UPDATE') - - + + def test_long_labels(self): dialect = default.DefaultDialect() dialect.max_identifier_length = 30 - + ora_dialect = oracle.dialect() - + m = MetaData() a_table = Table( 'thirty_characters_table_xxxxxx', @@ -156,7 +156,7 @@ class CompileTest(TestBase, AssertsCompiledSQL): primary_key=True ) ) - + anon = a_table.alias() self.assert_compile(select([other_table, anon]). @@ -189,7 +189,7 @@ class CompileTest(TestBase, AssertsCompiledSQL): 'thirty_characters_table__1.id = ' 'other_thirty_characters_table_.thirty_char' 'acters_table_id', dialect=ora_dialect) - + def test_outer_join(self): table1 = table('mytable', column('myid', Integer), @@ -283,12 +283,12 @@ class CompileTest(TestBase, AssertsCompiledSQL): 'mytable.name) WHERE ROWNUM <= :ROWNUM_1) ' 'WHERE ora_rn > :ora_rn_1', 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)) - + self.assert_compile(q, 'SELECT thirdtable.userid, ' 'thirdtable.otherstuff FROM thirdtable ' @@ -299,7 +299,7 @@ class CompileTest(TestBase, AssertsCompiledSQL): 'mytable.myid = myothertable.otherid) ' 'anon_1 ON thirdtable.userid = anon_1.myid' , dialect=oracle.dialect(use_ansi=True)) - + self.assert_compile(q, 'SELECT thirdtable.userid, ' 'thirdtable.otherstuff FROM thirdtable, ' @@ -310,7 +310,7 @@ class CompileTest(TestBase, AssertsCompiledSQL): '+)) anon_1 WHERE thirdtable.userid = ' 'anon_1.myid(+)', dialect=oracle.dialect(use_ansi=False)) - + q = select([table1.c.name]).where(table1.c.name == 'foo') self.assert_compile(q, 'SELECT mytable.name FROM mytable WHERE ' @@ -326,7 +326,7 @@ class CompileTest(TestBase, AssertsCompiledSQL): 'mytable.name) AS bar FROM mytable', dialect=oracle.dialect(use_ansi=False)) - + def test_alias_outer_join(self): address_types = table('address_types', column('id'), column('name')) @@ -362,11 +362,11 @@ class CompileTest(TestBase, AssertsCompiledSQL): class CompatFlagsTest(TestBase, AssertsCompiledSQL): __only_on__ = 'oracle' - + def test_ora8_flags(self): def server_version_info(self): return (8, 2, 5) - + dialect = oracle.dialect(dbapi=testing.db.dialect.dbapi) dialect._get_server_version_info = server_version_info @@ -389,7 +389,7 @@ class CompatFlagsTest(TestBase, AssertsCompiledSQL): dialect._get_server_version_info = server_version_info dialect.initialize(testing.db.connect()) assert dialect.implicit_returning - + def test_default_flags(self): """test with no initialization or server version info""" @@ -400,7 +400,7 @@ class CompatFlagsTest(TestBase, AssertsCompiledSQL): self.assert_compile(String(50),"VARCHAR(50 CHAR)",dialect=dialect) self.assert_compile(Unicode(50),"NVARCHAR2(50)",dialect=dialect) self.assert_compile(UnicodeText(),"NCLOB",dialect=dialect) - + def test_ora10_flags(self): def server_version_info(self): return (10, 2, 5) @@ -413,23 +413,23 @@ class CompatFlagsTest(TestBase, AssertsCompiledSQL): self.assert_compile(String(50),"VARCHAR(50 CHAR)",dialect=dialect) self.assert_compile(Unicode(50),"NVARCHAR2(50)",dialect=dialect) self.assert_compile(UnicodeText(),"NCLOB",dialect=dialect) - - + + class MultiSchemaTest(TestBase, AssertsCompiledSQL): __only_on__ = 'oracle' - + @classmethod def setup_class(cls): # currently assuming full DBA privs for the user. # don't really know how else to go here unless # we connect as the other user. - + for stmt in """ create table test_schema.parent( id integer primary key, data varchar2(50) ); - + create table test_schema.child( id integer primary key, data varchar2(50), @@ -441,14 +441,14 @@ create synonym test_schema.ctable for test_schema.child; -- can't make a ref from local schema to the -- remote schema's table without this, --- *and* cant give yourself a grant ! +-- *and* cant give yourself a grant ! -- so we give it to public. ideas welcome. grant references on test_schema.parent to public; grant references on test_schema.child to public; """.split(";"): if stmt.strip(): testing.db.execute(stmt) - + @classmethod def teardown_class(cls): for stmt in """ @@ -459,7 +459,7 @@ drop synonym test_schema.ptable; """.split(";"): if stmt.strip(): testing.db.execute(stmt) - + def test_create_same_names_explicit_schema(self): schema = testing.db.dialect.default_schema_name meta = MetaData(testing.db) @@ -597,7 +597,7 @@ class ConstraintTest(TestBase): ForeignKeyConstraint(['foo_id'], ['foo.id'], onupdate='CASCADE')) assert_raises(exc.SAWarning, bat.create) - + class TypesTest(TestBase, AssertsCompiledSQL): __only_on__ = 'oracle' __dialect__ = oracle.OracleDialect() @@ -620,7 +620,7 @@ class TypesTest(TestBase, AssertsCompiledSQL): b = bindparam("foo", u"hello world!") assert b.type.dialect_impl(dialect).get_dbapi_type(dbapi) == 'STRING' - + @testing.fails_on('+zxjdbc', 'zxjdbc lacks the FIXED_CHAR dbapi type') def test_fixed_char(self): m = MetaData(testing.db) @@ -628,7 +628,7 @@ class TypesTest(TestBase, AssertsCompiledSQL): Column('id', Integer, primary_key=True), Column('data', CHAR(30), nullable=False) ) - + t.create() try: t.insert().execute( @@ -640,17 +640,17 @@ class TypesTest(TestBase, AssertsCompiledSQL): eq_(t.select().where(t.c.data=='value 2').execute().fetchall(), [(2, 'value 2 ')] ) - + m2 = MetaData(testing.db) t2 = Table('t1', m2, autoload=True) assert type(t2.c.data.type) is CHAR eq_(t2.select().where(t2.c.data=='value 2').execute().fetchall(), [(2, 'value 2 ')] ) - + finally: t.drop() - + def test_type_adapt(self): dialect = cx_oracle.dialect() @@ -686,7 +686,7 @@ class TypesTest(TestBase, AssertsCompiledSQL): assert isinstance(x, int) finally: t1.drop() - + @testing.provide_metadata def test_rowid(self): t = Table('t1', metadata, @@ -697,7 +697,7 @@ class TypesTest(TestBase, AssertsCompiledSQL): s1 = select([t]) s2 = select([column('rowid')]).select_from(s1) rowid = s2.scalar() - + # the ROWID type is not really needed here, # as cx_oracle just treats it as a string, # but we want to make sure the ROWID works... @@ -707,7 +707,7 @@ class TypesTest(TestBase, AssertsCompiledSQL): eq_(s3.select().execute().fetchall(), [(5, rowid)] ) - + @testing.fails_on('+zxjdbc', 'Not yet known how to pass values of the ' 'INTERVAL type') @@ -738,7 +738,7 @@ class TypesTest(TestBase, AssertsCompiledSQL): seconds=5743)) finally: metadata.drop_all() - + def test_numerics(self): m = MetaData(testing.db) t1 = Table('t1', m, @@ -750,7 +750,7 @@ class TypesTest(TestBase, AssertsCompiledSQL): Column('numbercol1', oracle.NUMBER(9)), Column('numbercol2', oracle.NUMBER(9, 3)), Column('numbercol3', oracle.NUMBER), - + ) t1.create() try: @@ -764,7 +764,7 @@ class TypesTest(TestBase, AssertsCompiledSQL): numbercol2=14.85, numbercol3=15.76 ) - + m2 = MetaData(testing.db) t2 = Table('t1', m2, autoload=True) @@ -788,17 +788,17 @@ class TypesTest(TestBase, AssertsCompiledSQL): finally: t1.drop() - + @testing.provide_metadata def test_numerics_broken_inspection(self): """Numeric scenarios where Oracle type info is 'broken', returning us precision, scale of the form (0, 0) or (0, -127). We convert to Decimal and let int()/float() processors take over. - + """ - + # this test requires cx_oracle 5 - + foo = Table('foo', metadata, Column('idata', Integer), Column('ndata', Numeric(20, 2)), @@ -807,7 +807,7 @@ class TypesTest(TestBase, AssertsCompiledSQL): Column('fdata', Float()), ) foo.create() - + foo.insert().execute( {'idata':5, 'ndata':decimal.Decimal("45.6"), 'ndata2':decimal.Decimal("45.0"), 'nidata':decimal.Decimal('53'), 'fdata':45.68392}, @@ -822,8 +822,8 @@ class TypesTest(TestBase, AssertsCompiledSQL): fdata FROM foo """ - - + + row = testing.db.execute(stmt).fetchall()[0] eq_([type(x) for x in row], [int, decimal.Decimal, decimal.Decimal, int, float]) eq_( @@ -857,7 +857,7 @@ class TypesTest(TestBase, AssertsCompiledSQL): row, (5, decimal.Decimal('45.6'), 45, 53, decimal.Decimal('45.68392')) ) - + row = testing.db.execute(text(stmt, typemap={ 'idata':Integer(), @@ -870,7 +870,7 @@ class TypesTest(TestBase, AssertsCompiledSQL): eq_(row, (5, decimal.Decimal('45.6'), decimal.Decimal('45'), decimal.Decimal('53'), 45.683920000000001) ) - + stmt = """ SELECT anon_1.idata AS anon_1_idata, @@ -924,7 +924,7 @@ class TypesTest(TestBase, AssertsCompiledSQL): (5, 45.6, 45, 53, decimal.Decimal('45.68392')) ) - + def test_reflect_dates(self): metadata = MetaData(testing.db) Table( @@ -946,10 +946,10 @@ class TypesTest(TestBase, AssertsCompiledSQL): assert isinstance(t1.c.d3.type, TIMESTAMP) assert t1.c.d3.type.timezone assert isinstance(t1.c.d4.type, oracle.INTERVAL) - + finally: metadata.drop_all() - + def test_reflect_raw(self): types_table = Table('all_types', MetaData(testing.db), Column('owner', String(30), primary_key=True), @@ -984,7 +984,7 @@ class TypesTest(TestBase, AssertsCompiledSQL): assert isinstance(res, unicode) finally: metadata.drop_all() - + def test_char_length(self): self.assert_compile(VARCHAR(50),"VARCHAR(50 CHAR)") @@ -994,7 +994,7 @@ class TypesTest(TestBase, AssertsCompiledSQL): self.assert_compile(NVARCHAR(50),"NVARCHAR2(50)") self.assert_compile(CHAR(50),"CHAR(50)") - + metadata = MetaData(testing.db) t1 = Table('t1', metadata, Column("c1", VARCHAR(50)), @@ -1043,24 +1043,24 @@ class TypesTest(TestBase, AssertsCompiledSQL): eq_(row['bindata'].read(), 'this is binary') finally: t.drop(engine) - + class EuroNumericTest(TestBase): """test the numeric output_type_handler when using non-US locale for NLS_LANG.""" - + __only_on__ = 'oracle+cx_oracle' - + def setup(self): self.old_nls_lang = os.environ.get('NLS_LANG', False) os.environ['NLS_LANG'] = "GERMAN" self.engine = testing_engine() - + def teardown(self): if self.old_nls_lang is not False: os.environ['NLS_LANG'] = self.old_nls_lang else: del os.environ['NLS_LANG'] self.engine.dispose() - + @testing.provide_metadata def test_output_type_handler(self): for stmt, exp, kw in [ @@ -1076,8 +1076,8 @@ class EuroNumericTest(TestBase): exp ) assert type(test_exp) is type(exp) - - + + class DontReflectIOTTest(TestBase): """test that index overflow tables aren't included in table_names.""" @@ -1097,10 +1097,10 @@ class DontReflectIOTTest(TestBase): PCTTHRESHOLD 20 OVERFLOW TABLESPACE users """) - + def teardown(self): testing.db.execute("drop table admin_docindex") - + def test_reflect_all(self): m = MetaData(testing.db) m.reflect() @@ -1108,7 +1108,7 @@ class DontReflectIOTTest(TestBase): set(t.name for t in m.tables.values()), set(['admin_docindex']) ) - + class BufferedColumnTest(TestBase, AssertsCompiledSQL): __only_on__ = 'oracle' @@ -1145,7 +1145,7 @@ class BufferedColumnTest(TestBase, AssertsCompiledSQL): class UnsupportedIndexReflectTest(TestBase): __only_on__ = 'oracle' - + def setup(self): global metadata metadata = MetaData(testing.db) @@ -1153,16 +1153,16 @@ class UnsupportedIndexReflectTest(TestBase): Column('data', String(20), primary_key=True) ) metadata.create_all() - + def teardown(self): metadata.drop_all() - + def test_reflect_functional_index(self): testing.db.execute('CREATE INDEX DATA_IDX ON ' 'TEST_INDEX_REFLECT (UPPER(DATA))') m2 = MetaData(testing.db) t2 = Table('test_index_reflect', m2, autoload=True) - + class RoundTripIndexTest(TestBase): __only_on__ = 'oracle' @@ -1236,7 +1236,7 @@ class RoundTripIndexTest(TestBase): metadata.drop_all() - + class SequenceTest(TestBase, AssertsCompiledSQL): def test_basic(self): @@ -1250,8 +1250,8 @@ class SequenceTest(TestBase, AssertsCompiledSQL): seq = Sequence('My_Seq', schema='Some_Schema') assert dialect.identifier_preparer.format_sequence(seq) \ == '"Some_Schema"."My_Seq"' - - + + class ExecuteTest(TestBase): __only_on__ = 'oracle' @@ -1259,7 +1259,7 @@ class ExecuteTest(TestBase): def test_basic(self): eq_(testing.db.execute('/*+ this is a comment */ SELECT 1 FROM ' 'DUAL').fetchall(), [(1, )]) - + def test_sequences_are_integers(self): seq = Sequence('foo_seq') seq.create(testing.db) @@ -1269,17 +1269,17 @@ class ExecuteTest(TestBase): assert type(val) is int finally: seq.drop(testing.db) - + @testing.provide_metadata def test_limit_offset_for_update(self): # 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) ) metadata.create_all() - + t.insert().execute( {'id':1, 'data':1}, {'id':2, 'data':7}, @@ -1287,7 +1287,7 @@ class ExecuteTest(TestBase): {'id':4, 'data':15}, {'id':5, 'data':32}, ) - + # here, we can't use ORDER BY. eq_( t.select(for_update=True).limit(2).execute().fetchall(), @@ -1302,5 +1302,5 @@ class ExecuteTest(TestBase): "ORA-02014", t.select(for_update=True).limit(2).offset(3).execute ) - - + + diff --git a/test/dialect/test_postgresql.py b/test/dialect/test_postgresql.py index cfccb9bb1..a8c63c566 100644 --- a/test/dialect/test_postgresql.py +++ b/test/dialect/test_postgresql.py @@ -56,7 +56,7 @@ class CompileTest(TestBase, AssertsCompiledSQL): 'RETURNING length(mytable.name) AS length_1' , dialect=dialect) - + def test_insert_returning(self): dialect = postgresql.dialect() table1 = table('mytable', @@ -83,7 +83,7 @@ class CompileTest(TestBase, AssertsCompiledSQL): 'INSERT INTO mytable (name) VALUES ' '(%(name)s) RETURNING length(mytable.name) ' 'AS length_1', dialect=dialect) - + @testing.uses_deprecated('.*argument is deprecated. Please use ' 'statement.returning.*') def test_old_returning_names(self): @@ -109,7 +109,7 @@ class CompileTest(TestBase, AssertsCompiledSQL): 'INSERT INTO mytable (name) VALUES ' '(%(name)s) RETURNING mytable.myid, ' 'mytable.name', dialect=dialect) - + def test_create_partial_index(self): m = MetaData() tbl = Table('testtbl', m, Column('data', Integer)) @@ -228,7 +228,7 @@ class FloatCoercionTest(TablesTest, AssertsExecutionResults): {'data':52}, {'data':9}, ) - + @testing.resolve_artifact_names def test_float_coercion(self): for type_, result in [ @@ -251,7 +251,7 @@ class FloatCoercionTest(TablesTest, AssertsExecutionResults): ]) ).scalar() eq_(round_decimal(ret, 9), result) - + @testing.provide_metadata def test_arrays(self): t1 = Table('t', metadata, @@ -267,7 +267,7 @@ class FloatCoercionTest(TablesTest, AssertsExecutionResults): row, ([5], [5], [6], [decimal.Decimal("6.4")]) ) - + class EnumTest(TestBase, AssertsExecutionResults, AssertsCompiledSQL): __only_on__ = 'postgresql' @@ -296,7 +296,7 @@ class EnumTest(TestBase, AssertsExecutionResults, AssertsCompiledSQL): "CREATE TABLE sometable (somecolumn " "VARCHAR(1), CHECK (somecolumn IN ('x', " "'y', 'z')))") - + @testing.fails_on('postgresql+zxjdbc', 'zxjdbc fails on ENUM: column "XXX" is of type ' 'XXX but expression is of type character varying') @@ -319,7 +319,7 @@ class EnumTest(TestBase, AssertsExecutionResults, AssertsCompiledSQL): finally: metadata.drop_all() metadata.drop_all() - + def test_name_required(self): metadata = MetaData(testing.db) etype = Enum('four', 'five', 'six', metadata=metadata) @@ -341,7 +341,7 @@ class EnumTest(TestBase, AssertsExecutionResults, AssertsCompiledSQL): Enum(u'réveillé', u'drôle', u'S’il', name='onetwothreetype')) ) - + metadata.create_all() try: t1.insert().execute(value=u'drôle') @@ -411,24 +411,24 @@ class EnumTest(TestBase, AssertsExecutionResults, AssertsCompiledSQL): metadata.drop_all() assert not testing.db.dialect.has_type(testing.db, 'fourfivesixtype') - + def test_no_support(self): def server_version_info(self): return (8, 2) - + e = engines.testing_engine() dialect = e.dialect dialect._get_server_version_info = server_version_info - + assert dialect.supports_native_enum e.connect() assert not dialect.supports_native_enum - + # initialize is called again on new pool e.dispose() e.connect() assert not dialect.supports_native_enum - + def test_reflection(self): metadata = MetaData(testing.db) @@ -476,14 +476,14 @@ class EnumTest(TestBase, AssertsExecutionResults, AssertsCompiledSQL): metadata.drop_all() class NumericInterpretationTest(TestBase): - - + + def test_numeric_codes(self): from sqlalchemy.dialects.postgresql import pg8000, psycopg2, base from sqlalchemy.util.compat import decimal - + for dialect in (pg8000.dialect(), psycopg2.dialect()): - + typ = Numeric().dialect_impl(dialect) for code in base._INT_TYPES + base._FLOAT_TYPES + \ base._DECIMAL_TYPES: @@ -492,7 +492,7 @@ class NumericInterpretationTest(TestBase): if proc is not None: val = proc(val) assert val in (23.7, decimal.Decimal("23.7")) - + class InsertTest(TestBase, AssertsExecutionResults): __only_on__ = 'postgresql' @@ -540,7 +540,7 @@ class InsertTest(TestBase, AssertsExecutionResults): assert_raises_message(exc.DBAPIError, 'violates not-null constraint', eng.execute, t2.insert()) - + def test_sequence_insert(self): table = Table('testtable', metadata, Column('id', Integer, Sequence('my_seq'), primary_key=True), @@ -975,7 +975,7 @@ class DomainReflectionTest(TestBase, AssertsExecutionResults): con.execute("DROP TABLE enum_test") con.execute("DROP DOMAIN enumdomain") con.execute("DROP TYPE testtype") - + def test_table_is_reflected(self): metadata = MetaData(testing.db) table = Table('testtable', metadata, autoload=True) @@ -990,7 +990,7 @@ class DomainReflectionTest(TestBase, AssertsExecutionResults): "Reflected default value didn't equal expected value") assert not table.columns.answer.nullable, \ 'Expected reflected column to not be nullable.' - + def test_enum_domain_is_reflected(self): metadata = MetaData(testing.db) table = Table('enum_test', metadata, autoload=True) @@ -998,7 +998,7 @@ class DomainReflectionTest(TestBase, AssertsExecutionResults): table.c.data.type.enums, ('test', ) ) - + def test_table_is_reflected_test_schema(self): metadata = MetaData(testing.db) table = Table('testtable', metadata, autoload=True, @@ -1356,7 +1356,7 @@ class MiscTest(TestBase, AssertsExecutionResults, AssertsCompiledSQL): 'Skipped unsupported reflection of ' 'expression-based index idx3' ]) - + @testing.fails_on('postgresql+pypostgresql', 'pypostgresql bombs on multiple calls') @@ -1370,7 +1370,7 @@ class MiscTest(TestBase, AssertsExecutionResults, AssertsCompiledSQL): isolation_level='SERIALIZABLE') eq_(eng.execute('show transaction isolation level').scalar(), 'serializable') - + # check that it stays conn = eng.connect() eq_(conn.execute('show transaction isolation level').scalar(), @@ -1381,7 +1381,7 @@ class MiscTest(TestBase, AssertsExecutionResults, AssertsCompiledSQL): eq_(conn.execute('show transaction isolation level').scalar(), 'serializable') conn.close() - + eng = create_engine(testing.db.url, isolation_level='FOO') if testing.db.driver == 'zxjdbc': exception_cls = eng.dialect.dbapi.Error @@ -1399,11 +1399,11 @@ class MiscTest(TestBase, AssertsExecutionResults, AssertsCompiledSQL): stmt = text("select cast('hi' as char) as hi", typemap={'hi' : Numeric}) assert_raises(exc.InvalidRequestError, testing.db.execute, stmt) - + class TimezoneTest(TestBase): """Test timezone-aware datetimes. - + psycopg will return a datetime with a tzinfo attached to it, if postgresql returns it. python then will not let you compare a datetime with a tzinfo to a datetime that doesnt have one. this @@ -1526,7 +1526,7 @@ class TimePrecisionTest(TestBase, AssertsCompiledSQL): eq_(t2.c.c6.type.timezone, True) finally: t1.drop() - + class ArrayTest(TestBase, AssertsExecutionResults): __only_on__ = 'postgresql' @@ -1654,7 +1654,7 @@ class ArrayTest(TestBase, AssertsExecutionResults): foo.id = 2 sess.add(foo) sess.flush() - + @testing.provide_metadata def test_tuple_flag(self): assert_raises_message( @@ -1662,7 +1662,7 @@ class ArrayTest(TestBase, AssertsExecutionResults): "mutable must be set to False if as_tuple is True.", postgresql.ARRAY, Integer, mutable=True, as_tuple=True) - + t1 = Table('t1', metadata, Column('id', Integer, primary_key=True), Column('data', postgresql.ARRAY(String(5), as_tuple=True, mutable=False)), @@ -1672,7 +1672,7 @@ class ArrayTest(TestBase, AssertsExecutionResults): testing.db.execute(t1.insert(), id=1, data=["1","2","3"], data2=[5.4, 5.6]) testing.db.execute(t1.insert(), id=2, data=["4", "5", "6"], data2=[1.0]) testing.db.execute(t1.insert(), id=3, data=[["4", "5"], ["6", "7"]], data2=[[5.4, 5.6], [1.0, 1.1]]) - + r = testing.db.execute(t1.select().order_by(t1.c.id)).fetchall() eq_( r, @@ -1687,16 +1687,16 @@ class ArrayTest(TestBase, AssertsExecutionResults): set(row[1] for row in r), set([('1', '2', '3'), ('4', '5', '6'), (('4', '5'), ('6', '7'))]) ) - - - + + + class TimestampTest(TestBase, AssertsExecutionResults): __only_on__ = 'postgresql' def test_timestamp(self): engine = testing.db connection = engine.connect() - + s = select(["timestamp '2007-12-25'"]) result = connection.execute(s).first() eq_(result[0], datetime.datetime(2007, 12, 25, 0, 0)) @@ -1846,15 +1846,15 @@ class ServerSideCursorsTest(TestBase, AssertsExecutionResults): class SpecialTypesTest(TestBase, ComparesTables): """test DDL and reflection of PG-specific types """ - + __only_on__ = 'postgresql' __excluded_on__ = (('postgresql', '<', (8, 3, 0)),) - + @classmethod def setup_class(cls): global metadata, table metadata = MetaData(testing.db) - + # create these types so that we can issue # special SQL92 INTERVAL syntax class y2m(types.UserDefinedType, postgresql.INTERVAL): @@ -1864,7 +1864,7 @@ class SpecialTypesTest(TestBase, ComparesTables): class d2s(types.UserDefinedType, postgresql.INTERVAL): def get_col_spec(self): return "INTERVAL DAY TO SECOND" - + table = Table('sometable', metadata, Column('id', postgresql.PGUuid, primary_key=True), Column('flag', postgresql.PGBit), @@ -1877,31 +1877,31 @@ class SpecialTypesTest(TestBase, ComparesTables): Column('month_interval', d2s()), Column('precision_interval', postgresql.INTERVAL(precision=3)) ) - + metadata.create_all() - + # cheat so that the "strict type check" # works table.c.year_interval.type = postgresql.INTERVAL() table.c.month_interval.type = postgresql.INTERVAL() - + @classmethod def teardown_class(cls): metadata.drop_all() - + def test_reflection(self): m = MetaData(testing.db) t = Table('sometable', m, autoload=True) - + self.assert_tables_equal(table, t, strict_types=True) assert t.c.plain_interval.type.precision is None assert t.c.precision_interval.type.precision == 3 class UUIDTest(TestBase): """Test the bind/return values of the UUID type.""" - + __only_on__ = 'postgresql' - + @testing.requires.python25 @testing.fails_on('postgresql+pg8000', 'No support for UUID type') def test_uuid_string(self): @@ -1913,7 +1913,7 @@ class UUIDTest(TestBase): str(uuid.uuid4()), str(uuid.uuid4()) ) - + @testing.requires.python25 @testing.fails_on('postgresql+pg8000', 'No support for UUID type') def test_uuid_uuid(self): @@ -1925,7 +1925,7 @@ class UUIDTest(TestBase): uuid.uuid4(), uuid.uuid4() ) - + def test_no_uuid_available(self): from sqlalchemy.dialects.postgresql import base uuid_type = base._python_UUID @@ -1937,14 +1937,14 @@ class UUIDTest(TestBase): ) finally: base._python_UUID = uuid_type - + def setup(self): self.conn = testing.db.connect() trans = self.conn.begin() - + def teardown(self): self.conn.close() - + def _test_round_trip(self, utable, value1, value2): utable.create(self.conn) self.conn.execute(utable.insert(), {'data':value1}) @@ -1955,8 +1955,8 @@ class UUIDTest(TestBase): ) eq_(r.fetchone()[0], value2) eq_(r.fetchone(), None) - - + + class MatchTest(TestBase, AssertsCompiledSQL): __only_on__ = 'postgresql' @@ -2063,9 +2063,9 @@ class MatchTest(TestBase, AssertsCompiledSQL): class TupleTest(TestBase): __only_on__ = 'postgresql' - + def test_tuple_containment(self): - + for test, exp in [ ([('a', 'b')], True), ([('a', 'c')], False), diff --git a/test/dialect/test_sqlite.py b/test/dialect/test_sqlite.py index 34f5927ed..e3618f841 100644 --- a/test/dialect/test_sqlite.py +++ b/test/dialect/test_sqlite.py @@ -16,7 +16,7 @@ class TestTypes(TestBase, AssertsExecutionResults): def test_boolean(self): """Test that the boolean only treats 1 as True - + """ meta = MetaData(testing.db) @@ -214,7 +214,7 @@ class DialectTest(TestBase, AssertsExecutionResults): def test_extra_reserved_words(self): """Tests reserved words in identifiers. - + 'true', 'false', and 'column' are undocumented reserved words when used as column identifiers (as of 3.5.1). Covering them here to ensure they remain in place if the dialect's @@ -318,18 +318,18 @@ class DialectTest(TestBase, AssertsExecutionResults): except exc.DBAPIError: pass raise - + def test_pool_class(self): e = create_engine('sqlite+pysqlite://') assert e.pool.__class__ is pool.SingletonThreadPool e = create_engine('sqlite+pysqlite:///:memory:') assert e.pool.__class__ is pool.SingletonThreadPool - + e = create_engine('sqlite+pysqlite:///foo.db') assert e.pool.__class__ is pool.NullPool - - + + def test_dont_reflect_autoindex(self): meta = MetaData(testing.db) t = Table('foo', meta, Column('bar', String, primary_key=True)) @@ -414,7 +414,7 @@ class SQLTest(TestBase, AssertsCompiledSQL): Column('id', Integer, primary_key=True), Column('t1_id', Integer, ForeignKey('master.t1.id')), ) - + # schema->schema, generate REFERENCES with no schema name self.assert_compile( schema.CreateTable(t2), @@ -423,7 +423,7 @@ class SQLTest(TestBase, AssertsCompiledSQL): "t1_id INTEGER, " "PRIMARY KEY (id), " "FOREIGN KEY(t1_id) REFERENCES t1 (id)" - ")" + ")" ) # schema->different schema, don't generate REFERENCES @@ -433,7 +433,7 @@ class SQLTest(TestBase, AssertsCompiledSQL): "id INTEGER NOT NULL, " "t1_id INTEGER, " "PRIMARY KEY (id)" - ")" + ")" ) # same for local schema @@ -443,7 +443,7 @@ class SQLTest(TestBase, AssertsCompiledSQL): "id INTEGER NOT NULL, " "t1_id INTEGER, " "PRIMARY KEY (id)" - ")" + ")" ) |