mirror of
https://github.com/sqlalchemy/sqlalchemy.git
synced 2026-05-12 11:49:59 -04:00
Mike Bayer
c0b5a0446b
tests extend from either TestBase or ORMTest, using additional mixins for special assertion methods as needed
805 lines
33 KiB
Python
805 lines
33 KiB
Python
import testenv; testenv.configure_for_tests()
|
|
import datetime, os, pickleable, re
|
|
from sqlalchemy import *
|
|
from sqlalchemy import exceptions, types, util
|
|
from sqlalchemy.sql import operators
|
|
import sqlalchemy.engine.url as url
|
|
from sqlalchemy.databases import mssql, oracle, mysql, postgres, firebird
|
|
from testlib import *
|
|
|
|
|
|
|
|
class AdaptTest(TestBase):
|
|
def testadapt(self):
|
|
e1 = url.URL('postgres').get_dialect()()
|
|
e2 = url.URL('mysql').get_dialect()()
|
|
e3 = url.URL('sqlite').get_dialect()()
|
|
e4 = url.URL('firebird').get_dialect()()
|
|
|
|
type = String(40)
|
|
|
|
t1 = type.dialect_impl(e1)
|
|
t2 = type.dialect_impl(e2)
|
|
t3 = type.dialect_impl(e3)
|
|
t4 = type.dialect_impl(e4)
|
|
|
|
impls = [t1, t2, t3, t4]
|
|
for i,ta in enumerate(impls):
|
|
for j,tb in enumerate(impls):
|
|
if i == j:
|
|
assert ta == tb # call me paranoid... :)
|
|
else:
|
|
assert ta != tb
|
|
|
|
def testmsnvarchar(self):
|
|
dialect = mssql.MSSQLDialect()
|
|
# run the test twice to insure the caching step works too
|
|
for x in range(0, 1):
|
|
col = Column('', Unicode(length=10))
|
|
dialect_type = col.type.dialect_impl(dialect)
|
|
assert isinstance(dialect_type, mssql.MSNVarchar)
|
|
assert dialect_type.get_col_spec() == 'NVARCHAR(10)'
|
|
|
|
def testoracletext(self):
|
|
dialect = oracle.OracleDialect()
|
|
class MyDecoratedType(types.TypeDecorator):
|
|
impl = String
|
|
def copy(self):
|
|
return MyDecoratedType()
|
|
|
|
col = Column('', MyDecoratedType)
|
|
dialect_type = col.type.dialect_impl(dialect)
|
|
assert isinstance(dialect_type.impl, oracle.OracleText), repr(dialect_type.impl)
|
|
|
|
|
|
def testoracletimestamp(self):
|
|
dialect = oracle.OracleDialect()
|
|
t1 = oracle.OracleTimestamp
|
|
t2 = oracle.OracleTimestamp()
|
|
t3 = types.TIMESTAMP
|
|
assert isinstance(dialect.type_descriptor(t1), oracle.OracleTimestamp)
|
|
assert isinstance(dialect.type_descriptor(t2), oracle.OracleTimestamp)
|
|
assert isinstance(dialect.type_descriptor(t3), oracle.OracleTimestamp)
|
|
|
|
def testmysqlbinary(self):
|
|
dialect = mysql.MySQLDialect()
|
|
t1 = mysql.MSVarBinary
|
|
t2 = mysql.MSVarBinary()
|
|
assert isinstance(dialect.type_descriptor(t1), mysql.MSVarBinary)
|
|
assert isinstance(dialect.type_descriptor(t2), mysql.MSVarBinary)
|
|
|
|
def teststringadapt(self):
|
|
"""test that String with no size becomes TEXT, *all* others stay as varchar/String"""
|
|
|
|
oracle_dialect = oracle.OracleDialect()
|
|
mysql_dialect = mysql.MySQLDialect()
|
|
postgres_dialect = postgres.PGDialect()
|
|
firebird_dialect = firebird.FBDialect()
|
|
|
|
for dialect, start, test in [
|
|
(oracle_dialect, String(), oracle.OracleText),
|
|
(oracle_dialect, VARCHAR(), oracle.OracleString),
|
|
(oracle_dialect, String(50), oracle.OracleString),
|
|
(oracle_dialect, Unicode(), oracle.OracleText),
|
|
(oracle_dialect, UnicodeText(), oracle.OracleText),
|
|
(oracle_dialect, NCHAR(), oracle.OracleString),
|
|
(mysql_dialect, String(), mysql.MSText),
|
|
(mysql_dialect, VARCHAR(), mysql.MSString),
|
|
(mysql_dialect, String(50), mysql.MSString),
|
|
(mysql_dialect, Unicode(), mysql.MSText),
|
|
(mysql_dialect, UnicodeText(), mysql.MSText),
|
|
(mysql_dialect, NCHAR(), mysql.MSNChar),
|
|
(postgres_dialect, String(), postgres.PGText),
|
|
(postgres_dialect, VARCHAR(), postgres.PGString),
|
|
(postgres_dialect, String(50), postgres.PGString),
|
|
(postgres_dialect, Unicode(), postgres.PGText),
|
|
(postgres_dialect, UnicodeText(), postgres.PGText),
|
|
(postgres_dialect, NCHAR(), postgres.PGString),
|
|
(firebird_dialect, String(), firebird.FBText),
|
|
(firebird_dialect, VARCHAR(), firebird.FBString),
|
|
(firebird_dialect, String(50), firebird.FBString),
|
|
(firebird_dialect, Unicode(), firebird.FBText),
|
|
(firebird_dialect, UnicodeText(), firebird.FBText),
|
|
(firebird_dialect, NCHAR(), firebird.FBString),
|
|
]:
|
|
assert isinstance(start.dialect_impl(dialect), test), "wanted %r got %r" % (test, start.dialect_impl(dialect))
|
|
|
|
|
|
|
|
class UserDefinedTest(TestBase):
|
|
"""tests user-defined types."""
|
|
|
|
def testbasic(self):
|
|
print users.c.goofy4.type
|
|
print users.c.goofy4.type.dialect_impl(testing.db.dialect)
|
|
print users.c.goofy4.type.dialect_impl(testing.db.dialect).get_col_spec()
|
|
|
|
def testprocessing(self):
|
|
|
|
global users
|
|
users.insert().execute(user_id = 2, goofy = 'jack', goofy2='jack', goofy3='jack', goofy4=u'jack', goofy5=u'jack', goofy6='jack', goofy7=u'jack', goofy8=12, goofy9=12)
|
|
users.insert().execute(user_id = 3, goofy = 'lala', goofy2='lala', goofy3='lala', goofy4=u'lala', goofy5=u'lala', goofy6='lala', goofy7=u'lala', goofy8=15, goofy9=15)
|
|
users.insert().execute(user_id = 4, goofy = 'fred', goofy2='fred', goofy3='fred', goofy4=u'fred', goofy5=u'fred', goofy6='fred', goofy7=u'fred', goofy8=9, goofy9=9)
|
|
|
|
l = users.select().execute().fetchall()
|
|
for assertstr, assertint, assertint2, row in zip(
|
|
["BIND_INjackBIND_OUT", "BIND_INlalaBIND_OUT", "BIND_INfredBIND_OUT"],
|
|
[1200, 1500, 900],
|
|
[1800, 2250, 1350],
|
|
l
|
|
|
|
):
|
|
for col in row[1:8]:
|
|
self.assertEquals(col, assertstr)
|
|
self.assertEquals(row[8], assertint)
|
|
self.assertEquals(row[9], assertint2)
|
|
for col in (row[4], row[5], row[7]):
|
|
assert isinstance(col, unicode)
|
|
|
|
def setUpAll(self):
|
|
global users, metadata
|
|
|
|
class MyType(types.TypeEngine):
|
|
def get_col_spec(self):
|
|
return "VARCHAR(100)"
|
|
def bind_processor(self, dialect):
|
|
def process(value):
|
|
return "BIND_IN"+ value
|
|
return process
|
|
def result_processor(self, dialect):
|
|
def process(value):
|
|
return value + "BIND_OUT"
|
|
return process
|
|
def adapt(self, typeobj):
|
|
return typeobj()
|
|
|
|
class MyDecoratedType(types.TypeDecorator):
|
|
impl = String
|
|
def bind_processor(self, dialect):
|
|
impl_processor = super(MyDecoratedType, self).bind_processor(dialect) or (lambda value:value)
|
|
def process(value):
|
|
return "BIND_IN"+ impl_processor(value)
|
|
return process
|
|
def result_processor(self, dialect):
|
|
impl_processor = super(MyDecoratedType, self).result_processor(dialect) or (lambda value:value)
|
|
def process(value):
|
|
return impl_processor(value) + "BIND_OUT"
|
|
return process
|
|
def copy(self):
|
|
return MyDecoratedType()
|
|
|
|
class MyNewUnicodeType(types.TypeDecorator):
|
|
impl = Unicode
|
|
|
|
def process_bind_param(self, value, dialect):
|
|
return "BIND_IN" + value
|
|
|
|
def process_result_value(self, value, dialect):
|
|
return value + "BIND_OUT"
|
|
|
|
def copy(self):
|
|
return MyNewUnicodeType(self.impl.length)
|
|
|
|
class MyNewIntType(types.TypeDecorator):
|
|
impl = Integer
|
|
|
|
def process_bind_param(self, value, dialect):
|
|
return value * 10
|
|
|
|
def process_result_value(self, value, dialect):
|
|
return value * 10
|
|
|
|
def copy(self):
|
|
return MyNewIntType()
|
|
|
|
class MyNewIntSubClass(MyNewIntType):
|
|
def process_result_value(self, value, dialect):
|
|
return value * 15
|
|
|
|
def copy(self):
|
|
return MyNewIntSubClass()
|
|
|
|
class MyUnicodeType(types.TypeDecorator):
|
|
impl = Unicode
|
|
|
|
def bind_processor(self, dialect):
|
|
impl_processor = super(MyUnicodeType, self).bind_processor(dialect) or (lambda value:value)
|
|
|
|
def process(value):
|
|
return "BIND_IN"+ impl_processor(value)
|
|
return process
|
|
|
|
def result_processor(self, dialect):
|
|
impl_processor = super(MyUnicodeType, self).result_processor(dialect) or (lambda value:value)
|
|
def process(value):
|
|
return impl_processor(value) + "BIND_OUT"
|
|
return process
|
|
|
|
def copy(self):
|
|
return MyUnicodeType(self.impl.length)
|
|
|
|
class LegacyType(types.TypeEngine):
|
|
def get_col_spec(self):
|
|
return "VARCHAR(100)"
|
|
def convert_bind_param(self, value, dialect):
|
|
return "BIND_IN"+ value
|
|
def convert_result_value(self, value, dialect):
|
|
return value + "BIND_OUT"
|
|
def adapt(self, typeobj):
|
|
return typeobj()
|
|
|
|
class LegacyUnicodeType(types.TypeDecorator):
|
|
impl = Unicode
|
|
|
|
def convert_bind_param(self, value, dialect):
|
|
return "BIND_IN" + super(LegacyUnicodeType, self).convert_bind_param(value, dialect)
|
|
|
|
def convert_result_value(self, value, dialect):
|
|
return super(LegacyUnicodeType, self).convert_result_value(value, dialect) + "BIND_OUT"
|
|
|
|
def copy(self):
|
|
return LegacyUnicodeType(self.impl.length)
|
|
|
|
metadata = MetaData(testing.db)
|
|
users = Table('type_users', metadata,
|
|
Column('user_id', Integer, primary_key = True),
|
|
# totall custom type
|
|
Column('goofy', MyType, nullable = False),
|
|
|
|
# decorated type with an argument, so its a String
|
|
Column('goofy2', MyDecoratedType(50), nullable = False),
|
|
|
|
# decorated type without an argument, it will adapt_args to TEXT
|
|
Column('goofy3', MyDecoratedType, nullable = False),
|
|
|
|
Column('goofy4', MyUnicodeType, nullable = False),
|
|
Column('goofy5', LegacyUnicodeType, nullable = False),
|
|
Column('goofy6', LegacyType, nullable = False),
|
|
Column('goofy7', MyNewUnicodeType, nullable = False),
|
|
Column('goofy8', MyNewIntType, nullable = False),
|
|
Column('goofy9', MyNewIntSubClass, nullable = False),
|
|
|
|
)
|
|
|
|
metadata.create_all()
|
|
|
|
def tearDownAll(self):
|
|
metadata.drop_all()
|
|
|
|
class ColumnsTest(TestBase, AssertsExecutionResults):
|
|
|
|
def testcolumns(self):
|
|
expectedResults = { 'int_column': 'int_column INTEGER',
|
|
'smallint_column': 'smallint_column SMALLINT',
|
|
'varchar_column': 'varchar_column VARCHAR(20)',
|
|
'numeric_column': 'numeric_column NUMERIC(12, 3)',
|
|
'float_column': 'float_column FLOAT(25)',
|
|
}
|
|
|
|
db = testing.db
|
|
if testing.against('sqlite', 'oracle'):
|
|
expectedResults['float_column'] = 'float_column NUMERIC(25, 2)'
|
|
|
|
if testing.against('maxdb'):
|
|
expectedResults['numeric_column'] = (
|
|
expectedResults['numeric_column'].replace('NUMERIC', 'FIXED'))
|
|
|
|
print db.engine.__module__
|
|
testTable = Table('testColumns', MetaData(db),
|
|
Column('int_column', Integer),
|
|
Column('smallint_column', SmallInteger),
|
|
Column('varchar_column', String(20)),
|
|
Column('numeric_column', Numeric(12,3)),
|
|
Column('float_column', Float(25)),
|
|
)
|
|
|
|
for aCol in testTable.c:
|
|
self.assertEquals(
|
|
expectedResults[aCol.name],
|
|
db.dialect.schemagenerator(db.dialect, db, None, None).\
|
|
get_column_specification(aCol))
|
|
|
|
class UnicodeTest(TestBase, AssertsExecutionResults):
|
|
"""tests the Unicode type. also tests the TypeDecorator with instances in the types package."""
|
|
def setUpAll(self):
|
|
global unicode_table
|
|
metadata = MetaData(testing.db)
|
|
unicode_table = Table('unicode_table', metadata,
|
|
Column('id', Integer, Sequence('uni_id_seq', optional=True), primary_key=True),
|
|
Column('unicode_varchar', Unicode(250)),
|
|
Column('unicode_text', UnicodeText),
|
|
Column('plain_varchar', String(250))
|
|
)
|
|
unicode_table.create()
|
|
def tearDownAll(self):
|
|
unicode_table.drop()
|
|
|
|
def tearDown(self):
|
|
unicode_table.delete().execute()
|
|
|
|
def testbasic(self):
|
|
assert unicode_table.c.unicode_varchar.type.length == 250
|
|
rawdata = 'Alors vous imaginez ma surprise, au lever du jour, quand une dr\xc3\xb4le de petit voix m\xe2\x80\x99a r\xc3\xa9veill\xc3\xa9. Elle disait: \xc2\xab S\xe2\x80\x99il vous pla\xc3\xaet\xe2\x80\xa6 dessine-moi un mouton! \xc2\xbb\n'
|
|
unicodedata = rawdata.decode('utf-8')
|
|
unicode_table.insert().execute(unicode_varchar=unicodedata,
|
|
unicode_text=unicodedata,
|
|
plain_varchar=rawdata)
|
|
x = unicode_table.select().execute().fetchone()
|
|
print 0, repr(unicodedata)
|
|
print 1, repr(x['unicode_varchar'])
|
|
print 2, repr(x['unicode_text'])
|
|
print 3, repr(x['plain_varchar'])
|
|
self.assert_(isinstance(x['unicode_varchar'], unicode) and x['unicode_varchar'] == unicodedata)
|
|
self.assert_(isinstance(x['unicode_text'], unicode) and x['unicode_text'] == unicodedata)
|
|
if isinstance(x['plain_varchar'], unicode):
|
|
# SQLLite and MSSQL return non-unicode data as unicode
|
|
self.assert_(testing.against('sqlite', 'mssql'))
|
|
self.assert_(x['plain_varchar'] == unicodedata)
|
|
print "it's %s!" % testing.db.name
|
|
else:
|
|
self.assert_(not isinstance(x['plain_varchar'], unicode) and x['plain_varchar'] == rawdata)
|
|
|
|
def testassert(self):
|
|
try:
|
|
unicode_table.insert().execute(unicode_varchar='not unicode')
|
|
assert False
|
|
except exceptions.SAWarning, e:
|
|
assert str(e) == "Unicode type received non-unicode bind param value 'not unicode'", str(e)
|
|
|
|
unicode_engine = engines.utf8_engine(options={'convert_unicode':True,
|
|
'assert_unicode':True})
|
|
try:
|
|
try:
|
|
unicode_engine.execute(unicode_table.insert(), plain_varchar='im not unicode')
|
|
assert False
|
|
except exceptions.InvalidRequestError, e:
|
|
assert str(e) == "Unicode type received non-unicode bind param value 'im not unicode'"
|
|
|
|
@testing.emits_warning('.*non-unicode bind')
|
|
def warns():
|
|
# test that data still goes in if warning is emitted....
|
|
unicode_table.insert().execute(unicode_varchar='not unicode')
|
|
assert (select([unicode_table.c.unicode_varchar]).execute().fetchall() == [('not unicode', )])
|
|
warns()
|
|
|
|
finally:
|
|
unicode_engine.dispose()
|
|
|
|
@testing.fails_on('oracle')
|
|
def testblanks(self):
|
|
unicode_table.insert().execute(unicode_varchar=u'')
|
|
assert select([unicode_table.c.unicode_varchar]).scalar() == u''
|
|
|
|
def testengineparam(self):
|
|
"""tests engine-wide unicode conversion"""
|
|
prev_unicode = testing.db.engine.dialect.convert_unicode
|
|
prev_assert = testing.db.engine.dialect.assert_unicode
|
|
try:
|
|
testing.db.engine.dialect.convert_unicode = True
|
|
testing.db.engine.dialect.assert_unicode = False
|
|
rawdata = 'Alors vous imaginez ma surprise, au lever du jour, quand une dr\xc3\xb4le de petit voix m\xe2\x80\x99a r\xc3\xa9veill\xc3\xa9. Elle disait: \xc2\xab S\xe2\x80\x99il vous pla\xc3\xaet\xe2\x80\xa6 dessine-moi un mouton! \xc2\xbb\n'
|
|
unicodedata = rawdata.decode('utf-8')
|
|
unicode_table.insert().execute(unicode_varchar=unicodedata,
|
|
unicode_text=unicodedata,
|
|
plain_varchar=rawdata)
|
|
x = unicode_table.select().execute().fetchone()
|
|
print 0, repr(unicodedata)
|
|
print 1, repr(x['unicode_varchar'])
|
|
print 2, repr(x['unicode_text'])
|
|
print 3, repr(x['plain_varchar'])
|
|
self.assert_(isinstance(x['unicode_varchar'], unicode) and x['unicode_varchar'] == unicodedata)
|
|
self.assert_(isinstance(x['unicode_text'], unicode) and x['unicode_text'] == unicodedata)
|
|
self.assert_(isinstance(x['plain_varchar'], unicode) and x['plain_varchar'] == unicodedata)
|
|
finally:
|
|
testing.db.engine.dialect.convert_unicode = prev_unicode
|
|
testing.db.engine.dialect.convert_unicode = prev_assert
|
|
|
|
@testing.unsupported('oracle')
|
|
def testlength(self):
|
|
"""checks the database correctly understands the length of a unicode string"""
|
|
teststr = u'aaa\x1234'
|
|
self.assert_(testing.db.func.length(teststr).scalar() == len(teststr))
|
|
|
|
class BinaryTest(TestBase, AssertsExecutionResults):
|
|
def setUpAll(self):
|
|
global binary_table, MyPickleType
|
|
|
|
class MyPickleType(types.TypeDecorator):
|
|
impl = PickleType
|
|
|
|
def process_bind_param(self, value, dialect):
|
|
if value:
|
|
value.stuff = 'this is modified stuff'
|
|
return value
|
|
|
|
def process_result_value(self, value, dialect):
|
|
if value:
|
|
value.stuff = 'this is the right stuff'
|
|
return value
|
|
|
|
binary_table = Table('binary_table', MetaData(testing.db),
|
|
Column('primary_id', Integer, Sequence('binary_id_seq', optional=True), primary_key=True),
|
|
Column('data', Binary),
|
|
Column('data_slice', Binary(100)),
|
|
Column('misc', String(30)),
|
|
# construct PickleType with non-native pickle module, since cPickle uses relative module
|
|
# loading and confuses this test's parent package 'sql' with the 'sqlalchemy.sql' package relative
|
|
# to the 'types' module
|
|
Column('pickled', PickleType),
|
|
Column('mypickle', MyPickleType)
|
|
)
|
|
binary_table.create()
|
|
|
|
def tearDown(self):
|
|
binary_table.delete().execute()
|
|
|
|
def tearDownAll(self):
|
|
binary_table.drop()
|
|
|
|
def testbinary(self):
|
|
testobj1 = pickleable.Foo('im foo 1')
|
|
testobj2 = pickleable.Foo('im foo 2')
|
|
testobj3 = pickleable.Foo('im foo 3')
|
|
|
|
stream1 =self.load_stream('binary_data_one.dat')
|
|
stream2 =self.load_stream('binary_data_two.dat')
|
|
binary_table.insert().execute(primary_id=1, misc='binary_data_one.dat', data=stream1, data_slice=stream1[0:100], pickled=testobj1, mypickle=testobj3)
|
|
binary_table.insert().execute(primary_id=2, misc='binary_data_two.dat', data=stream2, data_slice=stream2[0:99], pickled=testobj2)
|
|
binary_table.insert().execute(primary_id=3, misc='binary_data_two.dat', data=None, data_slice=stream2[0:99], pickled=None)
|
|
|
|
for stmt in (
|
|
binary_table.select(order_by=binary_table.c.primary_id),
|
|
text("select * from binary_table order by binary_table.primary_id", typemap={'pickled':PickleType, 'mypickle':MyPickleType}, bind=testing.db)
|
|
):
|
|
l = stmt.execute().fetchall()
|
|
print type(stream1), type(l[0]['data']), type(l[0]['data_slice'])
|
|
print len(stream1), len(l[0]['data']), len(l[0]['data_slice'])
|
|
self.assertEquals(list(stream1), list(l[0]['data']))
|
|
self.assertEquals(list(stream1[0:100]), list(l[0]['data_slice']))
|
|
self.assertEquals(list(stream2), list(l[1]['data']))
|
|
self.assertEquals(testobj1, l[0]['pickled'])
|
|
self.assertEquals(testobj2, l[1]['pickled'])
|
|
self.assertEquals(testobj3.moredata, l[0]['mypickle'].moredata)
|
|
self.assertEquals(l[0]['mypickle'].stuff, 'this is the right stuff')
|
|
|
|
def load_stream(self, name, len=12579):
|
|
f = os.path.join(os.path.dirname(testenv.__file__), name)
|
|
# put a number less than the typical MySQL default BLOB size
|
|
return file(f).read(len)
|
|
|
|
class ExpressionTest(TestBase, AssertsExecutionResults):
|
|
def setUpAll(self):
|
|
global test_table, meta
|
|
|
|
class MyCustomType(types.TypeEngine):
|
|
def get_col_spec(self):
|
|
return "INT"
|
|
def bind_processor(self, dialect):
|
|
def process(value):
|
|
return value * 10
|
|
return process
|
|
def result_processor(self, dialect):
|
|
def process(value):
|
|
return value / 10
|
|
return process
|
|
def adapt_operator(self, op):
|
|
return {operators.add:operators.sub, operators.sub:operators.add}.get(op, op)
|
|
|
|
meta = MetaData(testing.db)
|
|
test_table = Table('test', meta,
|
|
Column('id', Integer, primary_key=True),
|
|
Column('data', String(30)),
|
|
Column('atimestamp', Date),
|
|
Column('avalue', MyCustomType))
|
|
|
|
meta.create_all()
|
|
|
|
test_table.insert().execute({'id':1, 'data':'somedata', 'atimestamp':datetime.date(2007, 10, 15), 'avalue':25})
|
|
|
|
def tearDownAll(self):
|
|
meta.drop_all()
|
|
|
|
def test_control(self):
|
|
assert testing.db.execute("select avalue from test").scalar() == 250
|
|
|
|
assert test_table.select().execute().fetchall() == [(1, 'somedata', datetime.date(2007, 10, 15), 25)]
|
|
|
|
def test_bind_adapt(self):
|
|
expr = test_table.c.atimestamp == bindparam("thedate")
|
|
assert expr.right.type.__class__ == test_table.c.atimestamp.type.__class__
|
|
|
|
assert testing.db.execute(test_table.select().where(expr), {"thedate":datetime.date(2007, 10, 15)}).fetchall() == [(1, 'somedata', datetime.date(2007, 10, 15), 25)]
|
|
|
|
expr = test_table.c.avalue == bindparam("somevalue")
|
|
assert expr.right.type.__class__ == test_table.c.avalue.type.__class__
|
|
assert testing.db.execute(test_table.select().where(expr), {"somevalue":25}).fetchall() == [(1, 'somedata', datetime.date(2007, 10, 15), 25)]
|
|
|
|
|
|
def test_operator_adapt(self):
|
|
"""test type-based overloading of operators"""
|
|
|
|
# test string concatenation
|
|
expr = test_table.c.data + "somedata"
|
|
assert testing.db.execute(select([expr])).scalar() == "somedatasomedata"
|
|
|
|
expr = test_table.c.id + 15
|
|
assert testing.db.execute(select([expr])).scalar() == 16
|
|
|
|
# test custom operator conversion
|
|
expr = test_table.c.avalue + 40
|
|
assert expr.type.__class__ is test_table.c.avalue.type.__class__
|
|
|
|
# + operator converted to -
|
|
# value is calculated as: (250 - (40 * 10)) / 10 == -15
|
|
assert testing.db.execute(select([expr.label('foo')])).scalar() == -15
|
|
|
|
# this one relies upon anonymous labeling to assemble result
|
|
# processing rules on the column.
|
|
assert testing.db.execute(select([expr])).scalar() == -15
|
|
|
|
class DateTest(TestBase, AssertsExecutionResults):
|
|
def setUpAll(self):
|
|
global users_with_date, insert_data
|
|
|
|
db = testing.db
|
|
if testing.against('oracle'):
|
|
import sqlalchemy.databases.oracle as oracle
|
|
insert_data = [
|
|
[7, 'jack',
|
|
datetime.datetime(2005, 11, 10, 0, 0),
|
|
datetime.date(2005,11,10),
|
|
datetime.datetime(2005, 11, 10, 0, 0, 0, 29384)],
|
|
[8, 'roy',
|
|
datetime.datetime(2005, 11, 10, 11, 52, 35),
|
|
datetime.date(2005,10,10),
|
|
datetime.datetime(2006, 5, 10, 15, 32, 47, 6754)],
|
|
[9, 'foo',
|
|
datetime.datetime(2006, 11, 10, 11, 52, 35),
|
|
datetime.date(1970,4,1),
|
|
datetime.datetime(2004, 9, 18, 4, 0, 52, 1043)],
|
|
[10, 'colber', None, None, None]
|
|
]
|
|
fnames = ['user_id', 'user_name', 'user_datetime',
|
|
'user_date', 'user_time']
|
|
|
|
collist = [Column('user_id', INT, primary_key=True),
|
|
Column('user_name', VARCHAR(20)),
|
|
Column('user_datetime', DateTime),
|
|
Column('user_date', Date),
|
|
Column('user_time', TIMESTAMP)]
|
|
else:
|
|
datetime_micro = 54839
|
|
time_micro = 999
|
|
|
|
# Missing or poor microsecond support:
|
|
if testing.against('mssql', 'mysql', 'firebird'):
|
|
datetime_micro, time_micro = 0, 0
|
|
# No microseconds for TIME
|
|
elif testing.against('maxdb'):
|
|
time_micro = 0
|
|
|
|
insert_data = [
|
|
[7, 'jack',
|
|
datetime.datetime(2005, 11, 10, 0, 0),
|
|
datetime.date(2005, 11, 10),
|
|
datetime.time(12, 20, 2)],
|
|
[8, 'roy',
|
|
datetime.datetime(2005, 11, 10, 11, 52, 35),
|
|
datetime.date(2005, 10, 10),
|
|
datetime.time(0, 0, 0)],
|
|
[9, 'foo',
|
|
datetime.datetime(2005, 11, 10, 11, 52, 35, datetime_micro),
|
|
datetime.date(1970, 4, 1),
|
|
datetime.time(23, 59, 59, time_micro)],
|
|
[10, 'colber', None, None, None]
|
|
]
|
|
fnames = ['user_id', 'user_name', 'user_datetime',
|
|
'user_date', 'user_time']
|
|
|
|
collist = [Column('user_id', INT, primary_key=True),
|
|
Column('user_name', VARCHAR(20)),
|
|
Column('user_datetime', DateTime(timezone=False)),
|
|
Column('user_date', Date),
|
|
Column('user_time', Time)]
|
|
|
|
users_with_date = Table('query_users_with_date',
|
|
MetaData(testing.db), *collist)
|
|
users_with_date.create()
|
|
insert_dicts = [dict(zip(fnames, d)) for d in insert_data]
|
|
|
|
for idict in insert_dicts:
|
|
users_with_date.insert().execute(**idict)
|
|
|
|
def tearDownAll(self):
|
|
users_with_date.drop()
|
|
|
|
def testdate(self):
|
|
global insert_data
|
|
|
|
l = map(list, users_with_date.select().execute().fetchall())
|
|
self.assert_(l == insert_data,
|
|
'DateTest mismatch: got:%s expected:%s' % (l, insert_data))
|
|
|
|
def testtextdate(self):
|
|
x = testing.db.text(
|
|
"select user_datetime from query_users_with_date",
|
|
typemap={'user_datetime':DateTime}).execute().fetchall()
|
|
|
|
print repr(x)
|
|
self.assert_(isinstance(x[0][0], datetime.datetime))
|
|
|
|
x = testing.db.text(
|
|
"select * from query_users_with_date where user_datetime=:somedate",
|
|
bindparams=[bindparam('somedate', type_=types.DateTime)]).execute(
|
|
somedate=datetime.datetime(2005, 11, 10, 11, 52, 35)).fetchall()
|
|
print repr(x)
|
|
|
|
def testdate2(self):
|
|
meta = MetaData(testing.db)
|
|
t = Table('testdate', meta,
|
|
Column('id', Integer,
|
|
Sequence('datetest_id_seq', optional=True),
|
|
primary_key=True),
|
|
Column('adate', Date), Column('adatetime', DateTime))
|
|
t.create(checkfirst=True)
|
|
try:
|
|
d1 = datetime.date(2007, 10, 30)
|
|
t.insert().execute(adate=d1, adatetime=d1)
|
|
d2 = datetime.datetime(2007, 10, 30)
|
|
t.insert().execute(adate=d2, adatetime=d2)
|
|
|
|
x = t.select().execute().fetchall()[0]
|
|
self.assert_(x.adate.__class__ == datetime.date)
|
|
self.assert_(x.adatetime.__class__ == datetime.datetime)
|
|
|
|
t.delete().execute()
|
|
|
|
# test mismatched date/datetime
|
|
t.insert().execute(adate=d2, adatetime=d2)
|
|
self.assertEquals(select([t.c.adate, t.c.adatetime], t.c.adate==d1).execute().fetchall(), [(d1, d2)])
|
|
self.assertEquals(select([t.c.adate, t.c.adatetime], t.c.adate==d1).execute().fetchall(), [(d1, d2)])
|
|
|
|
finally:
|
|
t.drop(checkfirst=True)
|
|
|
|
class StringTest(TestBase, AssertsExecutionResults):
|
|
def test_nolen_string_deprecated(self):
|
|
metadata = MetaData(testing.db)
|
|
foo =Table('foo', metadata,
|
|
Column('one', String))
|
|
|
|
# no warning
|
|
select([func.count("*")], bind=testing.db).execute()
|
|
|
|
try:
|
|
# warning during CREATE
|
|
foo.create()
|
|
assert False
|
|
except exceptions.SADeprecationWarning, e:
|
|
assert "Using String type with no length" in str(e)
|
|
assert re.search(r'\bone\b', str(e))
|
|
|
|
bar = Table('bar', metadata, Column('one', String(40)))
|
|
|
|
try:
|
|
# no warning
|
|
bar.create()
|
|
|
|
# no warning for non-lengthed string
|
|
select([func.count("*")], from_obj=bar).execute()
|
|
finally:
|
|
bar.drop()
|
|
|
|
def _missing_decimal():
|
|
"""Python implementation supports decimals"""
|
|
try:
|
|
import decimal
|
|
return False
|
|
except ImportError:
|
|
return True
|
|
|
|
class NumericTest(TestBase, AssertsExecutionResults):
|
|
def setUpAll(self):
|
|
global numeric_table, metadata
|
|
metadata = MetaData(testing.db)
|
|
numeric_table = Table('numeric_table', metadata,
|
|
Column('id', Integer, Sequence('numeric_id_seq', optional=True), primary_key=True),
|
|
Column('numericcol', Numeric(asdecimal=False)),
|
|
Column('floatcol', Float),
|
|
Column('ncasdec', Numeric),
|
|
Column('fcasdec', Float(asdecimal=True))
|
|
)
|
|
metadata.create_all()
|
|
|
|
def tearDownAll(self):
|
|
metadata.drop_all()
|
|
|
|
def tearDown(self):
|
|
numeric_table.delete().execute()
|
|
|
|
@testing.fails_if(_missing_decimal)
|
|
def test_decimal(self):
|
|
from decimal import Decimal
|
|
numeric_table.insert().execute(
|
|
numericcol=3.5, floatcol=5.6, ncasdec=12.4, fcasdec=15.75)
|
|
numeric_table.insert().execute(
|
|
numericcol=Decimal("3.5"), floatcol=Decimal("5.6"),
|
|
ncasdec=Decimal("12.4"), fcasdec=Decimal("15.75"))
|
|
|
|
l = numeric_table.select().execute().fetchall()
|
|
print l
|
|
rounded = [
|
|
(l[0][0], l[0][1], round(l[0][2], 5), l[0][3], l[0][4]),
|
|
(l[1][0], l[1][1], round(l[1][2], 5), l[1][3], l[1][4]),
|
|
]
|
|
assert rounded == [
|
|
(1, 3.5, 5.6, Decimal("12.4"), Decimal("15.75")),
|
|
(2, 3.5, 5.6, Decimal("12.4"), Decimal("15.75")),
|
|
]
|
|
|
|
@testing.emits_warning('True Decimal types not available')
|
|
def test_decimal_fallback(self):
|
|
from sqlalchemy.util import Decimal # could be Decimal or float
|
|
|
|
numeric_table.insert().execute(ncasdec=12.4, fcasdec=15.75)
|
|
numeric_table.insert().execute(ncasdec=Decimal("12.4"),
|
|
fcasdec=Decimal("15.75"))
|
|
|
|
for row in numeric_table.select().execute().fetchall():
|
|
assert isinstance(row['ncasdec'], util.decimal_type)
|
|
assert isinstance(row['fcasdec'], util.decimal_type)
|
|
|
|
|
|
class IntervalTest(TestBase, AssertsExecutionResults):
|
|
def setUpAll(self):
|
|
global interval_table, metadata
|
|
metadata = MetaData(testing.db)
|
|
interval_table = Table("intervaltable", metadata,
|
|
Column("id", Integer, Sequence('interval_id_seq', optional=True), primary_key=True),
|
|
Column("interval", Interval),
|
|
)
|
|
metadata.create_all()
|
|
|
|
def tearDown(self):
|
|
interval_table.delete().execute()
|
|
|
|
def tearDownAll(self):
|
|
metadata.drop_all()
|
|
|
|
def test_roundtrip(self):
|
|
delta = datetime.datetime(2006, 10, 5) - datetime.datetime(2005, 8, 17)
|
|
interval_table.insert().execute(interval=delta)
|
|
assert interval_table.select().execute().fetchone()['interval'] == delta
|
|
|
|
def test_null(self):
|
|
interval_table.insert().execute(id=1, inverval=None)
|
|
assert interval_table.select().execute().fetchone()['interval'] is None
|
|
|
|
class BooleanTest(TestBase, AssertsExecutionResults):
|
|
def setUpAll(self):
|
|
global bool_table
|
|
metadata = MetaData(testing.db)
|
|
bool_table = Table('booltest', metadata,
|
|
Column('id', Integer, primary_key=True),
|
|
Column('value', Boolean))
|
|
bool_table.create()
|
|
def tearDownAll(self):
|
|
bool_table.drop()
|
|
def testbasic(self):
|
|
bool_table.insert().execute(id=1, value=True)
|
|
bool_table.insert().execute(id=2, value=False)
|
|
bool_table.insert().execute(id=3, value=True)
|
|
bool_table.insert().execute(id=4, value=True)
|
|
bool_table.insert().execute(id=5, value=True)
|
|
|
|
res = bool_table.select(bool_table.c.value==True).execute().fetchall()
|
|
print res
|
|
assert(res==[(1, True),(3, True),(4, True),(5, True)])
|
|
|
|
res2 = bool_table.select(bool_table.c.value==False).execute().fetchall()
|
|
print res2
|
|
assert(res2==[(2, False)])
|
|
|
|
if __name__ == "__main__":
|
|
testenv.main()
|