class Predicate(object):
@classmethod
def as_predicate(cls, predicate):
- if isinstance(predicate, Predicate):
+ if isinstance(predicate, skip_if):
+ return predicate.predicate
+ elif isinstance(predicate, Predicate):
return predicate
elif isinstance(predicate, list):
return OrPredicate([cls.as_predicate(pred) for pred in predicate])
return exclusions.open()
+ @property
+ def on_update_cascade(self):
+ """"target database must support ON UPDATE..CASCADE behavior in
+ foreign keys."""
+
+ return exclusions.open()
+
+ @property
+ def deferrable_fks(self):
+ return exclusions.closed()
+
+ @property
+ def on_update_or_deferrable_fks(self):
+ # TODO: exclusions should be composable,
+ # somehow only_if([x, y]) isn't working here, negation/conjunctions
+ # getting confused.
+ return exclusions.only_if(
+ lambda: self.on_update_cascade.enabled or self.deferrable_fks.enabled
+ )
+
+
@property
def self_referential_foreign_keys(self):
"""Target database must support self-referential foreign keys."""
"""
return exclusions.open()
+ @property
+ def unicode_ddl(self):
+ """Target driver must support some degree of non-ascii symbol names."""
+ return exclusions.closed()
+
@property
def datetime(self):
"""target dialect supports representation of Python
"""target database must use a plain percent '%' as the 'modulus'
operator."""
return exclusions.closed()
+
+ @property
+ def unicode_connections(self):
+ """Target driver must support non-ASCII characters being passed at all."""
+ return exclusions.open()
+
+ @property
+ def skip_mysql_on_windows(self):
+ """Catchall for a large variety of MySQL on Windows failures"""
+ return exclusions.open()
+
+ def _has_mysql_on_windows(self):
+ return False
+
+ def _has_mysql_fully_case_sensitive(self):
+ return False
])) > 0, '%s(%s), %s(%s)' % (col.name,
col.type, cols[i]['name'], ctype))
+ if not col.primary_key:
+ assert cols[i]['default'] is None
+
@testing.requires.table_reflection
def test_get_columns(self):
self._test_get_columns()
def test_get_table_oid_with_schema(self):
self._test_get_table_oid('users', schema='test_schema')
+ @testing.provide_metadata
+ def test_autoincrement_col(self):
+ """test that 'autoincrement' is reflected according to sqla's policy.
+
+ Don't mark this test as unsupported for any backend !
+
+ (technically it fails with MySQL InnoDB since "id" comes before "id2")
+
+ A backend is better off not returning "autoincrement" at all,
+ instead of potentially returning "False" for an auto-incrementing
+ primary key column.
+
+ """
+
+ meta = self.metadata
+ insp = inspect(meta.bind)
+
+ for tname, cname in [
+ ('users', 'user_id'),
+ ('email_addresses', 'address_id'),
+ ('dingalings', 'dingaling_id'),
+ ]:
+ cols = insp.get_columns(tname)
+ id_ = dict((c['name'], c) for c in cols)[cname]
+ assert id_.get('autoincrement', True)
+
+
__all__ = ('ComponentReflectionTest', 'HasTableTest')
from ..assertions import eq_
from ..config import requirements
from sqlalchemy import Integer, Unicode, UnicodeText, select
-from sqlalchemy import Date, DateTime, Time, MetaData, String
+from sqlalchemy import Date, DateTime, Time, MetaData, String, Text
from ..schema import Table, Column
import datetime
def test_empty_strings_text(self):
self._test_empty_strings()
+class TextTest(fixtures.TablesTest):
+ @classmethod
+ def define_tables(cls, metadata):
+ Table('text_table', metadata,
+ Column('id', Integer, primary_key=True,
+ test_needs_autoincrement=True),
+ Column('text_data', Text),
+ )
+
+ def test_text_roundtrip(self):
+ text_table = self.tables.text_table
+
+ config.db.execute(
+ text_table.insert(),
+ {"text_data": 'some text'}
+ )
+ row = config.db.execute(
+ select([text_table.c.text_data])
+ ).first()
+ eq_(row, ('some text',))
+
+ def test_text_empty_strings(self):
+ text_table = self.tables.text_table
+
+ config.db.execute(
+ text_table.insert(),
+ {"text_data": ''}
+ )
+ row = config.db.execute(
+ select([text_table.c.text_data])
+ ).first()
+ eq_(row, ('',))
+
class StringTest(fixtures.TestBase):
@requirements.unbounded_varchar
__all__ = ('UnicodeVarcharTest', 'UnicodeTextTest',
- 'DateTest', 'DateTimeTest',
+ 'DateTest', 'DateTimeTest', 'TextTest',
'DateTimeHistoricTest', 'DateTimeCoercedToDateTimeTest',
'TimeMicrosecondsTest', 'TimeTest', 'DateTimeMicrosecondsTest',
'DateHistoricTest', 'StringTest')
id INTEGER NOT NULL,
isbn VARCHAR(50) NOT NULL,
title VARCHAR(100) NOT NULL,
- series INTEGER,
- series_id INTEGER,
+ series INTEGER NOT NULL,
+ series_id INTEGER NOT NULL,
UNIQUE(series, series_id),
PRIMARY KEY(id, isbn)
)""")
# are really limited unless you're on PG or SQLite
# forget about it on these backends
- if testing.against('sybase', 'maxdb', 'oracle'):
+ if not testing.requires.unicode_ddl.enabled:
names = no_multibyte_period
# mysql can't handle casing usually
elif testing.against("mysql") and \
from sqlalchemy.testing import fixtures
from test.orm import _fixtures
+def _backend_specific_fk_args():
+ if testing.requires.deferrable_fks.enabled:
+ fk_args = dict(deferrable=True, initially='deferred')
+ elif not testing.requires.on_update_cascade.enabled:
+ fk_args = dict()
+ else:
+ fk_args = dict(onupdate='cascade')
+ return fk_args
+
class NaturalPKTest(fixtures.MappedTest):
# MySQL 5.5 on Windows crashes (the entire server, not the client)
# if you screw around with ON UPDATE CASCADE type of stuff.
- __requires__ = 'skip_mysql_on_windows',
+ __requires__ = 'skip_mysql_on_windows', 'on_update_or_deferrable_fks'
@classmethod
def define_tables(cls, metadata):
- if testing.against('oracle'):
- fk_args = dict(deferrable=True, initially='deferred')
- else:
- fk_args = dict(onupdate='cascade')
+ fk_args = _backend_specific_fk_args()
users = Table('users', metadata,
Column('username', String(50), primary_key=True),
assert sess.query(User).get('ed').fullname == 'jack'
- @testing.fails_on('sqlite', 'sqlite doesnt support ON UPDATE CASCADE')
- @testing.fails_on('oracle', 'oracle doesnt support ON UPDATE CASCADE')
+ @testing.requires.on_update_cascade
def test_onetomany_passive(self):
self._test_onetomany(True)
u1 = sess.query(User).get('fred')
eq_(User(username='fred', fullname='jack'), u1)
- @testing.fails_on('sqlite', 'sqlite doesnt support ON UPDATE CASCADE')
- @testing.fails_on('oracle', 'oracle doesnt support ON UPDATE CASCADE')
+ @testing.requires.on_update_cascade
def test_manytoone_passive(self):
self._test_manytoone(True)
sess.query(Address).all())
- @testing.fails_on('sqlite', 'sqlite doesnt support ON UPDATE CASCADE')
- @testing.fails_on('oracle', 'oracle doesnt support ON UPDATE CASCADE')
+ @testing.requires.on_update_cascade
def test_onetoone_passive(self):
self._test_onetoone(True)
sess.expunge_all()
eq_([Address(username='ed')], sess.query(Address).all())
- @testing.fails_on('sqlite', 'sqlite doesnt support ON UPDATE CASCADE')
- @testing.fails_on('oracle', 'oracle doesnt support ON UPDATE CASCADE')
+ @testing.requires.on_update_cascade
def test_bidirectional_passive(self):
self._test_bidirectional(True)
mapper(User, users)
mapper(Address, addresses, properties={
- 'user':relationship(User, passive_updates=passive_updates,
+ 'user': relationship(User, passive_updates=passive_updates,
backref='addresses')})
sess = create_session()
sess.query(Address).all())
- @testing.fails_on('sqlite', 'sqlite doesnt support ON UPDATE CASCADE')
- @testing.fails_on('oracle', 'oracle doesnt support ON UPDATE CASCADE')
+ @testing.requires.on_update_cascade
def test_manytomany_passive(self):
self._test_manytomany(True)
class SelfReferentialTest(fixtures.MappedTest):
# mssql, mysql don't allow
# ON UPDATE on self-referential keys
- __unsupported_on__ = ('mssql','mysql')
+ __unsupported_on__ = ('mssql', 'mysql')
+
+ __requires__ = 'on_update_or_deferrable_fks',
@classmethod
def define_tables(cls, metadata):
- if testing.against('oracle'):
- fk_args = dict(deferrable=True, initially='deferred')
- else:
- fk_args = dict(onupdate='cascade')
+ fk_args = _backend_specific_fk_args()
Table('nodes', metadata,
Column('name', String(50), primary_key=True),
for n in sess.query(Node).filter(
Node.name.in_(['n11', 'n12', 'n13']))])
- @testing.fails_on('sqlite', 'sqlite doesnt support ON UPDATE CASCADE')
- @testing.fails_on('oracle', 'oracle doesnt support ON UPDATE CASCADE')
+ @testing.requires.on_update_cascade
def test_many_to_one_passive(self):
self._test_many_to_one(True)
class NonPKCascadeTest(fixtures.MappedTest):
- __requires__ = 'skip_mysql_on_windows',
+ __requires__ = 'skip_mysql_on_windows', 'on_update_or_deferrable_fks'
@classmethod
def define_tables(cls, metadata):
- if testing.against('oracle'):
- fk_args = dict(deferrable=True, initially='deferred')
- else:
- fk_args = dict(onupdate='cascade')
+ fk_args = _backend_specific_fk_args()
Table('users', metadata,
Column('id', Integer, primary_key=True,
class Address(cls.Comparable):
pass
- @testing.fails_on('sqlite', 'sqlite doesnt support ON UPDATE CASCADE')
- @testing.fails_on('oracle', 'oracle doesnt support ON UPDATE CASCADE')
+ @testing.requires.on_update_cascade
def test_onetomany_passive(self):
self._test_onetomany(True)
@classmethod
def define_tables(cls, metadata):
- if testing.against('oracle'):
- fk_args = dict(deferrable=True, initially='deferred')
- else:
- fk_args = dict(onupdate='cascade')
+ fk_args = _backend_specific_fk_args()
Table('users', metadata,
Column('username', String(50), primary_key=True),
class Address(cls.Comparable):
pass
- @testing.fails_on('sqlite', 'sqlite doesnt support ON UPDATE CASCADE')
- @testing.fails_on('oracle', 'oracle doesnt support ON UPDATE CASCADE')
+ @testing.requires.on_update_cascade
def test_onetomany_passive(self):
self._test_onetomany(True)
u2.addresses.append(a1)
sess.flush()
- @testing.fails_on('oracle', 'oracle doesnt support ON UPDATE CASCADE '
- 'but requires referential integrity')
- @testing.fails_on('sqlite', 'sqlite doesnt support ON UPDATE CASCADE')
+ @testing.requires.on_update_cascade
def test_change_m2o_passive(self):
self._test_change_m2o(True)
@classmethod
def define_tables(cls, metadata):
- if testing.against('oracle'):
- fk_args = dict(deferrable=True, initially='deferred')
- else:
- fk_args = dict(onupdate='cascade')
+ fk_args = _backend_specific_fk_args()
Table('person', metadata,
Column('name', String(50), primary_key=True),
class Manager(Person):
pass
- @testing.fails_on('sqlite', 'sqlite doesnt support ON UPDATE CASCADE')
- @testing.fails_on('oracle', 'oracle doesnt support ON UPDATE CASCADE')
+ @testing.requires.on_update_cascade
def test_pk_passive(self):
self._test_pk(True)
def test_pk_nonpassive(self):
self._test_pk(False)
- @testing.fails_on('sqlite', 'sqlite doesnt support ON UPDATE CASCADE')
- @testing.fails_on('oracle', 'oracle doesnt support ON UPDATE CASCADE')
+ @testing.requires.on_update_cascade
def test_fk_passive(self):
self._test_fk(True)
def test_basic(self):
User = self.classes.User
- assert [User(id=7), User(id=8), User(id=9),User(id=10)] == create_session().query(User).all()
+ users = create_session().query(User).all()
+ eq_(
+ [User(id=7), User(id=8), User(id=9),User(id=10)],
+ users
+ )
- @testing.fails_on('maxdb', 'FIXME: unknown')
- def test_limit(self):
+ @testing.requires.offset
+ def test_limit_offset(self):
User = self.classes.User
- assert [User(id=8), User(id=9)] == create_session().query(User).order_by(User.id).limit(2).offset(1).all()
+ sess = create_session()
+
+ assert [User(id=8), User(id=9)] == sess.query(User).order_by(User.id).limit(2).offset(1).all()
+
+ assert [User(id=8), User(id=9)] == list(sess.query(User).order_by(User.id)[1:3])
- assert [User(id=8), User(id=9)] == list(create_session().query(User).order_by(User.id)[1:3])
+ assert User(id=8) == sess.query(User).order_by(User.id)[1]
- assert User(id=8) == create_session().query(User).order_by(User.id)[1]
+ assert [] == sess.query(User).order_by(User.id)[3:3]
+ assert [] == sess.query(User).order_by(User.id)[0:0]
- assert [] == create_session().query(User).order_by(User.id)[3:3]
- assert [] == create_session().query(User).order_by(User.id)[0:0]
@testing.requires.boolean_col_expressions
def test_exists(self):
no_support('sqlite', 'not supported by database')
)
+ @property
+ def on_update_cascade(self):
+ """target database must support ON UPDATE..CASCADE behavior in
+ foreign keys."""
+
+ return skip_if(
+ ['sqlite', 'oracle'],
+ 'target backend does not support ON UPDATE CASCADE'
+ )
+
+ @property
+ def deferrable_fks(self):
+ """target database must support deferrable fks"""
+
+ return only_on(['oracle'])
+
+
@property
def unbounded_varchar(self):
"""Target database must support VARCHAR with no length"""
@property
def unicode_data(self):
+ """target drive must support unicode data stored in columns."""
return skip_if([
no_support("sybase", "no unicode driver support")
])
@property
def unicode_ddl(self):
- """Target driver must support some encoding of Unicode across the wire."""
+ """Target driver must support some degree of non-ascii symbol names."""
# TODO: expand to exclude MySQLdb versions w/ broken unicode
return skip_if([
no_support('maxdb', 'database support flakey'),
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
