# pyformat is supported for mysql, but skipping because a few driver
# versions have a bug that bombs out on this test. (1.2.2b3, 1.2.2c1, 1.2.2)
- @testing.unsupported('mysql', 'supported but not covered for MySQLdb')
+ @testing.skip_if(lambda: testing.against('mysql'), 'db-api flaky')
@testing.fails_on_everything_except('postgres')
def test_raw_python(self):
for conn in (testing.db, testing.db.connect()):
meta.drop_all()
- @testing.unsupported('oracle', 'FIXME: unknown, confirm not fails_on')
+ @testing.crashes('oracle', 'FIXME: unknown, confirm not fails_on')
def testreserved(self):
# check a table that uses an SQL reserved name doesn't cause an error
meta = MetaData(testing.db)
assert buf.index("CREATE TABLE someschema.table1") > -1
assert buf.index("CREATE TABLE someschema.table2") > -1
- @testing.unsupported('firebird', 'FIXME: unknown- no schema support in db?')
+ @testing.crashes('firebird', 'FIXME: unknown- no schema support in db?')
@testing.fails_on('sqlite')
# fixme: revisit these below.
@testing.fails_on('access', 'oracle', 'mssql', 'sybase')
break
con.close()
- @testing.unsupported('sqlite', 'needs n threads -> 1 :memory: db')
- @testing.unsupported('mssql', 'FIXME: unknown')
- @testing.unsupported('firebird', 'FIXME: unknown')
- @testing.unsupported('sybase', 'FIXME: unknown')
- @testing.unsupported('access', 'FIXME: unknown')
+ @testing.crashes('mssql', 'FIXME: unknown')
+ @testing.crashes('firebird', 'FIXME: unknown')
+ @testing.crashes('sybase', 'FIXME: unknown')
+ @testing.crashes('access', 'FIXME: unknown')
+ @testing.requires.independent_connections
def test_queued_update(self):
"""Test SELECT FOR UPDATE with concurrent modifications.
return errors
- @testing.unsupported('sqlite', 'needs n threads -> 1 memory db')
- @testing.unsupported('mssql', 'FIXME: unknown')
- @testing.unsupported('firebird', 'FIXME: unknown')
- @testing.unsupported('sybase', 'FIXME: unknown')
- @testing.unsupported('access', 'FIXME: unknown')
+ @testing.crashes('mssql', 'FIXME: unknown')
+ @testing.crashes('firebird', 'FIXME: unknown')
+ @testing.crashes('sybase', 'FIXME: unknown')
+ @testing.crashes('access', 'FIXME: unknown')
+ @testing.requires.independent_connections
def test_queued_select(self):
"""Simple SELECT FOR UPDATE conflict test"""
sys.stderr.write("Failure: %s\n" % e)
self.assert_(len(errors) == 0)
- @testing.unsupported('sqlite', 'needs n threads -> 1 memory db')
- @testing.unsupported('mssql', 'FIXME: unknown')
- @testing.unsupported('mysql', 'no support for NOWAIT')
- @testing.unsupported('firebird', 'FIXME: unknown')
- @testing.unsupported('sybase', 'FIXME: unknown')
- @testing.unsupported('access', 'FIXME: unknown')
+ @testing.crashes('mssql', 'FIXME: unknown')
+ @testing.fails_on('mysql') # no support for NOWAIT
+ @testing.crashes('firebird', 'FIXME: unknown')
+ @testing.crashes('sybase', 'FIXME: unknown')
+ @testing.crashes('access', 'FIXME: unknown')
+ @testing.requires.independent_connections
def test_nowait_select(self):
"""Simple SELECT FOR UPDATE NOWAIT conflict test"""
result = ["%d %s" % ( t.id,t.category.name ) for t in l]
eq_(result, [u'1 Some Category', u'3 Some Category'])
- @testing.unsupported('sybase', 'FIXME: unknown, verify not fails_on')
+ @testing.crashes('sybase', 'FIXME: unknown, verify not fails_on')
@testing.resolve_artifact_names
def test_without_outerjoin_literal(self):
s = create_session()
assert res.order_by([Foo.bar])[0].bar == 5
assert res.order_by([sa.desc(Foo.bar)])[0].bar == 95
- @testing.unsupported('mssql', 'FIXME: verify not fails_on()')
+ @testing.crashes('mssql', 'FIXME: verify not fails_on()')
@testing.fails_on('maxdb')
@testing.resolve_artifact_names
def test_slice(self):
from sqlalchemy.orm.util import join, outerjoin, with_parent
+
class QueryTest(FixtureTest):
keep_mappers = True
keep_data = True
assert [User(id=10)] == sess.query(User).outerjoin("addresses", aliased=True).filter(~User.addresses.any()).all()
- @testing.unsupported('maxdb', 'can dump core')
+ @testing.crashes('maxdb', 'can dump core')
def test_has(self):
sess = create_session()
assert [Address(id=5)] == sess.query(Address).filter(Address.user.has(name='fred')).all()
assert len(session.query(User).filter_by(name='Johnny').all()) == 0
session.close()
- @testing.unsupported('mssql', 'test causes mssql to hang')
- @testing.unsupported('sqlite', 'needs true independent connections')
+ @testing.crashes('mssql', 'test causes mssql to hang')
+ @testing.requires.independent_connections
@engines.close_open_connections
@testing.resolve_artifact_names
def test_transaction(self):
assert testing.db.connect().execute("select count(1) from users").scalar() == 1
sess.close()
- @testing.unsupported('mssql', 'test causes mssql to hang')
- @testing.unsupported('sqlite', 'needs true independent connections')
+ @testing.crashes('mssql', 'test causes mssql to hang')
+ @testing.requires.independent_connections
@engines.close_open_connections
@testing.resolve_artifact_names
def test_autoflush(self):
eq_(sess.query(Address).filter(Address.user==u).one(),
Address(email_address='foo'))
- @testing.unsupported('mssql', 'test causes mssql to hang')
- @testing.unsupported('sqlite', 'needs true independent connections')
+ @testing.crashes('mssql', 'test causes mssql to hang')
+ @testing.requires.independent_connections
@engines.close_open_connections
@testing.resolve_artifact_names
def test_autoflush_unbound(self):
from engine import _base as engine_base
import pickleable
+
class UnitOfWorkTest(object):
pass
eq_(u.name, 'test2')
eq_(u.counter, 2)
- @testing.unsupported('mssql', 'FIXME: unknown, verify not fails_on()')
+ @testing.crashes('mssql', 'FIXME: unknown, verify not fails_on()')
@testing.resolve_artifact_names
def test_insert(self):
u = User(name='test', counter=sa.select([5]))
def tearDownAll(self):
metadata.drop_all()
- @testing.unsupported('mssql', 'FIXME: unknown, verify not fails_on')
+ @testing.crashes('mssql', 'FIXME: unknown, verify not fails_on')
def test_basic(self):
t2.insert().execute(nextid=1)
r = t1.insert().execute(data='hi')
r = users.select(limit=3, order_by=[users.c.user_id]).execute().fetchall()
self.assert_(r == [(1, 'john'), (2, 'jack'), (3, 'ed')], repr(r))
- @testing.unsupported('mssql', 'FIXME: guessing')
+ @testing.crashes('mssql', 'FIXME: guessing')
@testing.fails_on('maxdb')
def test_select_limit_offset(self):
users.insert().execute(user_id=1, user_name='john')
self.assertEqual([x.lower() for x in r.keys()], ['user_name', 'user_id'])
self.assertEqual(r.values(), ['foo', 1])
- @testing.unsupported('oracle', 'FIXME: unknown, varify not fails_on()')
- @testing.unsupported('firebird', 'FIXME: unknown, verify not fails_on()')
- @testing.unsupported('maxdb', 'FIXME: unknown, verify not fails_on()')
+ @testing.crashes('oracle', 'FIXME: unknown, varify not fails_on()')
+ @testing.crashes('firebird', 'FIXME: unknown, verify not fails_on()')
+ @testing.crashes('maxdb', 'FIXME: unknown, verify not fails_on()')
def test_column_accessor_shadow(self):
meta = MetaData(testing.db)
shadowed = Table('test_shadowed', meta,
('ccc', 'aaa')]
self.assertEquals(u.alias('bar').select().execute().fetchall(), wanted)
- @testing.unsupported('oracle', 'FIXME: unknown, verify not fails_on')
+ @testing.crashes('oracle', 'FIXME: unknown, verify not fails_on')
@testing.fails_on('mysql')
@testing.fails_on('sqlite')
def test_union_all(self):
found2 = self._fetchall_sorted(e.alias('foo').select().execute())
self.assertEquals(found2, wanted)
- @testing.unsupported('firebird', 'FIXME: unknown, verify not fails_on')
- @testing.unsupported('sybase', 'FIXME: unknown, verify not fails_on')
+ @testing.crashes('firebird', 'FIXME: unknown, verify not fails_on')
+ @testing.crashes('sybase', 'FIXME: unknown, verify not fails_on')
@testing.fails_on('mysql')
def test_intersect(self):
i = intersect(
found2 = self._fetchall_sorted(i.alias('bar').select().execute())
self.assertEquals(found2, wanted)
- @testing.unsupported('firebird', 'FIXME: unknown, verify not fails_on')
- @testing.unsupported('oracle', 'FIXME: unknown, verify not fails_on')
- @testing.unsupported('sybase', 'FIXME: unknown, verify not fails_on')
+ @testing.crashes('firebird', 'FIXME: unknown, verify not fails_on')
+ @testing.crashes('oracle', 'FIXME: unknown, verify not fails_on')
+ @testing.crashes('sybase', 'FIXME: unknown, verify not fails_on')
@testing.fails_on('mysql')
def test_except_style1(self):
e = except_(union(
found = self._fetchall_sorted(e.alias('bar').select().execute())
self.assertEquals(found, wanted)
- @testing.unsupported('firebird', 'FIXME: unknown, verify not fails_on')
- @testing.unsupported('oracle', 'FIXME: unknown, verify not fails_on')
- @testing.unsupported('sybase', 'FIXME: unknown, verify not fails_on')
+ @testing.crashes('firebird', 'FIXME: unknown, verify not fails_on')
+ @testing.crashes('oracle', 'FIXME: unknown, verify not fails_on')
+ @testing.crashes('sybase', 'FIXME: unknown, verify not fails_on')
@testing.fails_on('mysql')
def test_except_style2(self):
e = except_(union(
found2 = self._fetchall_sorted(e.alias('bar').select().execute())
self.assertEquals(found2, wanted)
- @testing.unsupported('firebird', 'FIXME: unknown, verify not fails_on')
- @testing.unsupported('oracle', 'FIXME: unknown, verify not fails_on')
- @testing.unsupported('sybase', 'FIXME: unknown, verify not fails_on')
+ @testing.crashes('firebird', 'FIXME: unknown, verify not fails_on')
+ @testing.crashes('oracle', 'FIXME: unknown, verify not fails_on')
+ @testing.crashes('sybase', 'FIXME: unknown, verify not fails_on')
@testing.fails_on('mysql')
@testing.fails_on('sqlite')
def test_except_style3(self):
self.assertEquals(e.alias('foo').select().execute().fetchall(),
[('ccc',)])
- @testing.unsupported('firebird', 'FIXME: unknown, verify not fails_on')
+ @testing.crashes('firebird', 'FIXME: unknown, verify not fails_on')
@testing.fails_on('mysql')
def test_composite(self):
u = intersect(
self.assertEquals(found, wanted)
- @testing.unsupported('firebird', 'FIXME: unknown, verify not fails_on')
+ @testing.crashes('firebird', 'FIXME: unknown, verify not fails_on')
@testing.fails_on('mysql')
def test_composite_alias(self):
ua = intersect(
from sqlalchemy.sql import compiler
from testlib import *
+
class QuoteTest(TestBase, AssertsCompiledSQL):
def setUpAll(self):
- # TODO: figure out which databases/which identifiers allow special characters to be used,
- # such as: spaces, quote characters, punctuation characters, set up tests for those as
- # well.
+ # TODO: figure out which databases/which identifiers allow special
+ # characters to be used, such as: spaces, quote characters,
+ # punctuation characters, set up tests for those as well.
global table1, table2, table3
metadata = MetaData(testing.db)
table1 = Table('WorstCase1', metadata,
res2 = select([table2.c.d123, table2.c.u123, table2.c.MixedCase], use_labels=True).execute().fetchall()
print res2
assert(res2==[(1,2,3),(2,2,3),(4,3,2)])
-
+
def test_quote_flag(self):
metadata = MetaData()
- t1 = Table('TableOne', metadata,
+ t1 = Table('TableOne', metadata,
Column('ColumnOne', Integer), schema="FooBar")
self.assert_compile(t1.select(), '''SELECT "FooBar"."TableOne"."ColumnOne" FROM "FooBar"."TableOne"''')
metadata = MetaData()
- t1 = Table('t1', metadata,
+ t1 = Table('t1', metadata,
Column('col1', Integer, quote=True), quote=True, schema="foo", quote_schema=True)
self.assert_compile(t1.select(), '''SELECT "foo"."t1"."col1" FROM "foo"."t1"''')
metadata = MetaData()
- t1 = Table('TableOne', metadata,
+ t1 = Table('TableOne', metadata,
Column('ColumnOne', Integer, quote=False), quote=False, schema="FooBar", quote_schema=False)
self.assert_compile(t1.select(), '''SELECT FooBar.TableOne.ColumnOne FROM FooBar.TableOne''')
-
- @testing.unsupported('oracle', 'FIXME: unknown, verify not fails_on')
+
+ @testing.crashes('oracle', 'FIXME: unknown, verify not fails_on')
@testing.requires.subqueries
def testlabels(self):
"""test the quoting of labels.
# note that 'foo' and 'FooCol' are literals already quoted
x = select([sql.literal_column("'foo'").label("somelabel")], from_obj=[table]).alias("AnAlias")
x = x.select()
- self.assert_compile(x,
+ self.assert_compile(x,
'''SELECT "AnAlias".somelabel FROM (SELECT 'foo' AS somelabel FROM "ImATable") AS "AnAlias"''')
x = select([sql.literal_column("'FooCol'").label("SomeLabel")], from_obj=[table])
x = x.select()
- self.assert_compile(x,
+ self.assert_compile(x,
'''SELECT "SomeLabel" FROM (SELECT 'FooCol' AS "SomeLabel" FROM "ImATable")''')
from testlib import *
-
class AdaptTest(TestBase):
def testadapt(self):
e1 = url.URL('postgres').get_dialect()()
testing.db.engine.dialect.convert_unicode = prev_unicode
testing.db.engine.dialect.convert_unicode = prev_assert
- @testing.unsupported('oracle', 'FIXME: unknown, verify not fails_on')
+ @testing.crashes('oracle', 'FIXME: unknown, verify not fails_on')
@testing.fails_on('firebird') # "Data type unknown" on the parameter
def testlength(self):
"""checks the database correctly understands the length of a unicode string"""
"""
from testlib.testing import \
+ _block_unconditionally as no_support, \
_chain_decorators_on, \
- exclude, \
- unsupported
+ exclude
-def sequences(fn):
- """Target database must support SEQUENCEs."""
+def deferrable_constraints(fn):
+ """Target database must support derferable constraints."""
return _chain_decorators_on(
fn,
- unsupported('access', 'no SEQUENCE support'),
- unsupported('mssql', 'no SEQUENCE support'),
- unsupported('mysql', 'no SEQUENCE support'),
- unsupported('sqlite', 'no SEQUENCE support'),
- unsupported('sybase', 'no SEQUENCE support'),
+ no_support('mysql', 'not supported by database'),
+ no_support('mssql', 'not supported by database'),
)
-def savepoints(fn):
- """Target database must support savepoints."""
+def foreign_keys(fn):
+ """Target database must support foreign keys."""
return _chain_decorators_on(
fn,
- unsupported('access', 'FIXME: guessing, needs confirmation'),
- unsupported('mssql', 'FIXME: guessing, needs confirmation'),
- unsupported('sqlite', 'not supported by database'),
- unsupported('sybase', 'FIXME: guessing, needs confirmation'),
- exclude('mysql', '<', (5, 0, 3), 'not supported by database'),
+ no_support('sqlite', 'not supported by database'),
)
-def two_phase_transactions(fn):
- """Target database must support two-phase transactions."""
+def independent_connections(fn):
+ """Target must support simultaneous, independent database connections."""
+
+ # This is also true of some configurations of UnixODBC and probably win32
+ # ODBC as well.
return _chain_decorators_on(
fn,
- unsupported('access', 'FIXME: guessing, needs confirmation'),
- unsupported('firebird', 'no SA implementation'),
- unsupported('maxdb', 'not supported by database'),
- unsupported('mssql', 'FIXME: guessing, needs confirmation'),
- unsupported('oracle', 'no SA implementation'),
- unsupported('sqlite', 'not supported by database'),
- unsupported('sybase', 'FIXME: guessing, needs confirmation'),
- exclude('mysql', '<', (5, 0, 3), 'not supported by database'),
+ no_support('sqlite', 'no driver support')
)
-def unicode_connections(fn):
- """Target driver must support some encoding of Unicode across the wire."""
- # TODO: expand to exclude MySQLdb versions w/ broken unicode
+def row_triggers(fn):
+ """Target must support standard statement-running EACH ROW triggers."""
return _chain_decorators_on(
fn,
- exclude('mysql', '<', (4, 1, 1), 'no unicode connection support'),
+ # no access to same table
+ exclude('mysql', '<', (5, 0, 10), 'not supported by database'),
+ no_support('postgres', 'not supported by database: no statements'),
)
-def unicode_ddl(fn):
- """Target driver must support some encoding of Unicode across the wire."""
- # TODO: expand to exclude MySQLdb versions w/ broken unicode
+def savepoints(fn):
+ """Target database must support savepoints."""
return _chain_decorators_on(
fn,
- unsupported('maxdb', 'database support flakey'),
- unsupported('oracle', 'FIXME: no support in database?'),
- unsupported('sybase', 'FIXME: guessing, needs confirmation'),
- exclude('mysql', '<', (4, 1, 1), 'no unicode connection support'),
+ no_support('access', 'FIXME: guessing, needs confirmation'),
+ no_support('mssql', 'FIXME: guessing, needs confirmation'),
+ no_support('sqlite', 'not supported by database'),
+ no_support('sybase', 'FIXME: guessing, needs confirmation'),
+ exclude('mysql', '<', (5, 0, 3), 'not supported by database'),
+ )
+
+def sequences(fn):
+ """Target database must support SEQUENCEs."""
+ return _chain_decorators_on(
+ fn,
+ no_support('access', 'no SEQUENCE support'),
+ no_support('mssql', 'no SEQUENCE support'),
+ no_support('mysql', 'no SEQUENCE support'),
+ no_support('sqlite', 'no SEQUENCE support'),
+ no_support('sybase', 'no SEQUENCE support'),
)
def subqueries(fn):
exclude('mysql', '<', (4, 1, 1), 'no subquery support'),
)
-def foreign_keys(fn):
- """Target database must support foreign keys."""
+def two_phase_transactions(fn):
+ """Target database must support two-phase transactions."""
return _chain_decorators_on(
fn,
- unsupported('sqlite', 'not supported by database'),
+ no_support('access', 'FIXME: guessing, needs confirmation'),
+ no_support('firebird', 'no SA implementation'),
+ no_support('maxdb', 'not supported by database'),
+ no_support('mssql', 'FIXME: guessing, needs confirmation'),
+ no_support('oracle', 'no SA implementation'),
+ no_support('sqlite', 'not supported by database'),
+ no_support('sybase', 'FIXME: guessing, needs confirmation'),
+ exclude('mysql', '<', (5, 0, 3), 'not supported by database'),
)
-def deferrable_constraints(fn):
- """Target database must support derferable constraints."""
+def unicode_connections(fn):
+ """Target driver must support some encoding of Unicode across the wire."""
+ # TODO: expand to exclude MySQLdb versions w/ broken unicode
return _chain_decorators_on(
fn,
- unsupported('mysql', 'not supported by database'),
- unsupported('mssql', 'not supported by database'),
+ exclude('mysql', '<', (4, 1, 1), 'no unicode connection support'),
)
-def row_triggers(fn):
- """Target must support standard statement-running EACH ROW triggers."""
+def unicode_ddl(fn):
+ """Target driver must support some encoding of Unicode across the wire."""
+ # TODO: expand to exclude MySQLdb versions w/ broken unicode
return _chain_decorators_on(
fn,
- # no access to same table
- exclude('mysql', '<', (5, 0, 10), 'not supported by database'),
- unsupported('postgres', 'not supported by database: no statements'),
+ no_support('maxdb', 'database support flakey'),
+ no_support('oracle', 'FIXME: no support in database?'),
+ no_support('sybase', 'FIXME: guessing, needs confirmation'),
+ exclude('mysql', '<', (4, 1, 1), 'no unicode connection support'),
)
return _function_named(maybe, fn_name)
return decorate
-def unsupported(db, reason):
+def crashes(db, reason):
"""Mark a test as unsupported by a database implementation.
- 'unsupported' tests will be skipped unconditionally. Useful for feature
- tests that cause deadlocks or other fatal problems.
+ 'crashes' tests will be skipped unconditionally. Use for feature tests
+ that cause deadlocks or other fatal problems.
+
"""
+ carp = _should_carp_about_exclusion(reason)
+ def decorate(fn):
+ fn_name = fn.__name__
+ def maybe(*args, **kw):
+ if config.db.name == db:
+ msg = "'%s' unsupported on DB implementation '%s': %s" % (
+ fn_name, config.db.name, reason)
+ print msg
+ if carp:
+ print >> sys.stderr, msg
+ return True
+ else:
+ return fn(*args, **kw)
+ return _function_named(maybe, fn_name)
+ return decorate
+def _block_unconditionally(db, reason):
+ """Mark a test as unsupported by a database implementation.
+
+ Will never run the test against any version of the given database, ever,
+ no matter what. Use when your assumptions are infallible; past, present
+ and future.
+
+ """
carp = _should_carp_about_exclusion(reason)
def decorate(fn):
fn_name = fn.__name__
return _function_named(maybe, fn_name)
return decorate
+
def exclude(db, op, spec, reason):
"""Mark a test as unsupported by specific database server versions.
bind = config.db
return bind.dialect.server_version_info(bind.contextual_connect())
+def skip_if(predicate, reason=None):
+ """Skip a test if predicate is true."""
+ reason = reason or predicate.__name__
+ def decorate(fn):
+ fn_name = fn.__name__
+ def maybe(*args, **kw):
+ if predicate():
+ msg = "'%s' skipped on DB %s version '%s': %s" % (
+ fn_name, config.db.name, _server_version(), reason)
+ print msg
+ return True
+ else:
+ return fn(*args, **kw)
+ return _function_named(maybe, fn_name)
+ return decorate
+
def emits_warning(*messages):
"""Mark a test as emitting a warning.