Here are my test results for the SO test suite. Some sources of errors:
* Many of the asserts use a string where the MySQLdb cursor returns and
integer.
* col.py does not allow for the cursor returning a datetime.timedelta
object. (I
didn't know this ever happened either.)
* There are a few instances where a column is typed as 'text' and then
used as a
key, but mysql doesn't allow this unless you specify a prefix. The
blob/text
types shouldn't be used for simple fields like 'name'.
* A couple asserts compare the row return value to a list, but the row
is a
tuple and will never compare equal with a list.
There's only two tests that looks like they really did fail:
testInnerAggregate
test_asDict
I'd try to fix them myself, but I'd probably break more than I fixed...
Here's the full log:
============================= test process starts
=============================
executable: /tools/aticad/1.0/external/python-2.4.1/bin/python
(2.4.1-final-0)
using py lib:
/tool/tools/aticad/1.0/external/python-2.4.1/lib/python2.4/site-packages
/py-0.9.1-py2.4.egg/py <rev unknown>
test_NoneValuedResultItem.py[1] .
test_SQLMultipleJoin.py[2] ..
test_SQLRelatedJoin.py[2] ..
test_SingleJoin.py[1] .
test_aggregates.py[3] F.F
test_aliases.py[3] ...
test_asdict.py[1] F
test_auto.py[5] .....
test_auto_old.py[1] .
test_basic.py[17] .................
test_blob.py[1] .
test_boundattributes.py[0]
test_cache.py[3] ...
test_combining_joins.py[1] .
test_comparison.py[1] F
test_constraints.py[1] .
test_converters.py[31] ...............................
test_create_drop.py[1] .
test_csvexport.py[3] ...
test_cyclic_reference.py[1] .
test_datetime.py[2] .F
test_decimal.py[2] ..
test_declarative.py[2] ..
test_default_style.py - FAILED TO LOAD MODULE
test_delete.py[5] .....
test_distinct.py[1] .
test_empty.py[1] .
test_enum.py[4] ....
test_events.py[6] ......
test_exceptions.py[1] F
test_expire.py[1] .
test_foreignKey.py[2] ..
test_groupBy.py[2] FF
test_indexes.py[4] ..FF
test_inheritance.py[2] ..
test_joins.py[5] .....
test_joins_conditional.py[7] .......
test_lazy.py[1] .
test_new_joins.py[5] .....
test_parse.py[1] .
test_paste.py[0]
test_picklecol.py[1] .
test_reparent_sqlmeta.py[1] .
test_select.py[15] ...............
test_select_through.py[5] .....
test_setters.py[1] .
test_slice.py[6] ......
test_sorting.py[4] ....
test_sqlbuilder.py[1] .
test_sqlbuilder_dbspecific.py[13] .............
test_sqlbuilder_importproxy.py[4] ....
test_sqlbuilder_joins_instances.py[7] .......
test_sqlite_factory.py[3] ...
test_sqlite_threaded.py[1] .
test_sqlobject_admin.py - FAILED TO LOAD MODULE
test_string_id.py[1] F
test_stringid.py[1] .
test_style.py[1] .
test_subqueries.py[9] .........
test_transactions.py[4] ....
test_unicode.py[2] ..
test_validation.py[3] ...
test_views.py[11] ....FF.F...
________________________________________________________________________
_______
__________________________ entrypoint: test_integer
___________________________
def test_integer():
setupClass(IntAccumulator)
IntAccumulator(value=1)
IntAccumulator(value=2)
IntAccumulator(value=3)
assert IntAccumulator.select().min(IntAccumulator.q.value) == 1
E assert IntAccumulator.select().avg(IntAccumulator.q.value) == 2
> assert '2.0000' == 2
+ where '2.0000' = <SelectResults at
2a9ae10b10>.avg(int_accumulator.value)
+ where <SelectResults at 2a9ae10b10> = IntAccumulator.select()
+ and int_accumulator = IntAccumulator.q
[/home/bgolemon/SQLObject/sqlobject/tests/test_aggregates.py:22]
________________________________________________________________________
_______
____________________________ entrypoint: test_many
____________________________
def test_many():
setupClass(IntAccumulator)
IntAccumulator(value=1)
IntAccumulator(value=1)
IntAccumulator(value=2)
IntAccumulator(value=2)
IntAccumulator(value=3)
IntAccumulator(value=3)
attribute = IntAccumulator.q.value
assert IntAccumulator.select().accumulateMany(
("MIN", attribute), ("AVG", attribute), ("MAX", attribute),
E ("COUNT", attribute), ("SUM", attribute)
> AssertionError: (inconsistently failed then succeeded)
[/home/bgolemon/SQLObject/sqlobject/tests/test_aggregates.py:61]
________________________________________________________________________
_______
___________________________ entrypoint: test_asDict
___________________________
def test_asDict():
setupClass(TestAsDict)
t1 = TestAsDict(name='one', name2='1')
E assert t1.sqlmeta.asDict() == dict(name='one', name2='1', id=1)
> assert {'id': 4L, 'name': 'one', 'name2': '1'} == {'id': 1, 'name':
'one', 'name2': '1'}
+ where {'id': 4L, 'name': 'one', 'name2': '1'} =
<sqlobject.declarative.sqlmeta object at 0x2a9ae28810>.asDict()
+ where <sqlobject.declarative.sqlmeta object at 0x2a9ae28810> =
<TestAsDict 4L name='one' name2='1'>.sqlmeta
+ and {'id': 1, 'name': 'one', 'name2': '1'} = dict(name='one',
name2='1', id=1)
[/home/bgolemon/SQLObject/sqlobject/tests/test_asdict.py:15]
________________________________________________________________________
_______
_____________________________ entrypoint: test_eq
_____________________________
def test_eq():
setupClass(TestComparison)
t1 = TestComparison()
t2 = TestComparison()
TestComparison._connection.cache.clear()
> t3 = TestComparison.get(1)
[/home/bgolemon/SQLObject/sqlobject/tests/test_comparison.py:13]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def get(cls, id, connection=None, selectResults=None):
assert id is not None, 'None is not a possible id for %s' % cls.__name__
id = cls.sqlmeta.idType(id)
if connection is None:
cache = cls._connection.cache
else:
cache = connection.cache
# This whole sequence comes from Cache.CacheFactory's
# behavior, where a None returned means a cache miss.
val = cache.get(id, cls)
if val is None:
try:
val = cls(_SO_fetch_no_create=1)
val._SO_validatorState = SQLObjectState(val)
> val._init(id, connection, selectResults)
[/home/bgolemon/SQLObject/sqlobject/main.py:891]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def _init(self, id, connection=None, selectResults=None):
assert id is not None
# This function gets called only when the object is
# created, unlike __init__ which would be called
# anytime the object was returned from cache.
self.id = id
self._SO_writeLock = threading.Lock()
# If no connection was given, we'll inherit the class
# instance variable which should have a _connection
# attribute.
if connection is not None:
self._connection = connection
# Sometimes we need to know if this instance is
# global or tied to a particular connection.
# This flag tells us that:
self.sqlmeta._perConnection = True
if not selectResults:
dbNames = [col.dbName for col in self.sqlmeta.columnList]
selectResults = self._connection._SO_selectOne(self, dbNames)
if not selectResults:
E raise SQLObjectNotFound, "The object %s by the ID %s does not exist" %
(self.__class__.__name__, self.id)
> SQLObjectNotFound: The object TestComparison by the ID 1 does not
exist
[/home/bgolemon/SQLObject/sqlobject/main.py:932]
________________________________________________________________________
_______
_________________________ entrypoint: test_mxDateTime
_________________________
def test_mxDateTime():
setupClass(DateTime2)
_now = now()
> dt2 = DateTime2(col1=_now, col2=_now, col3=Time(_now.hour,
_now.minute, int(_now.second)))
[/home/bgolemon/SQLObject/sqlobject/tests/test_datetime.py:60]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def __init__(self, **kw):
# If we are the outmost constructor of a hiearchy of
# InheritableSQLObjects (or simlpy _the_ constructor of a "normal"
# SQLObject), we create a threadlocal list that collects the
# RowCreatedSignals, and executes them if this very constructor is left
try:
_postponed_local.postponed_calls
postponed_created = False
except AttributeError:
_postponed_local.postponed_calls = []
postponed_created = True
try:
# We shadow the sqlmeta class with an instance of sqlmeta
# that points to us (our sqlmeta buddy object; where the
# sqlmeta class is our class's buddy class)
self.sqlmeta = self.__class__.sqlmeta(self)
# The get() classmethod/constructor uses a magic keyword
# argument when it wants an empty object, fetched from the
# database. So we have nothing more to do in that case:
if kw.has_key('_SO_fetch_no_create'):
return
post_funcs = []
self.sqlmeta.send(events.RowCreateSignal, kw, post_funcs)
# Pass the connection object along if we were given one.
if kw.has_key('connection'):
self._connection = kw['connection']
self.sqlmeta._perConnection = True
del kw['connection']
self._SO_writeLock = threading.Lock()
if kw.has_key('id'):
id = self.sqlmeta.idType(kw['id'])
del kw['id']
else:
id = None
> self._create(id, **kw)
[/home/bgolemon/SQLObject/sqlobject/main.py:1204]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def _create(self, id, **kw):
self.sqlmeta._creating = True
self._SO_createValues = {}
self._SO_validatorState = SQLObjectState(self)
# First we do a little fix-up on the keywords we were
# passed:
for column in self.sqlmeta.columnList:
# Then we check if the column wasn't passed in, and
# if not we try to get the default.
if not kw.has_key(column.name) and not kw.has_key(column.foreignName):
default = column.default
# If we don't get it, it's an error:
# If we specified an SQL DEFAULT, then we should use that
if default is NoDefault:
if column.defaultSQL is None:
raise TypeError, "%s() did not get expected keyword argument '%s'" %
(self.__class__.__name__, column.name)
else:
# There is defaultSQL for the column - do not put
# the column to kw so that the backend creates the value
continue
# Otherwise we put it in as though they did pass
# that keyword:
kw[column.name] = default
self.set(**kw)
# Then we finalize the process:
> self._SO_finishCreate(id)
[/home/bgolemon/SQLObject/sqlobject/main.py:1252]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def _SO_finishCreate(self, id=None):
# Here's where an INSERT is finalized.
# These are all the column values that were supposed
# to be set, but were delayed until now:
setters = self._SO_createValues.items()
# Here's their database names:
names = [self.sqlmeta.columns[v[0]].dbName for v in setters]
values = [v[1] for v in setters]
# Get rid of _SO_create*, we aren't creating anymore.
# Doesn't have to be threadsafe because we're still in
# new(), which doesn't need to be threadsafe.
self.dirty = False
if not self.sqlmeta.lazyUpdate:
del self._SO_createValues
else:
self._SO_createValues = {}
del self.sqlmeta._creating
# Do the insert -- most of the SQL in this case is left
# up to DBConnection, since getting a new ID is
# non-standard.
id = self._connection.queryInsertID(self,
id, names, values)
cache = self._connection.cache
cache.created(id, self.__class__, self)
> self._init(id)
[/home/bgolemon/SQLObject/sqlobject/main.py:1279]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def _init(self, id, connection=None, selectResults=None):
assert id is not None
# This function gets called only when the object is
# created, unlike __init__ which would be called
# anytime the object was returned from cache.
self.id = id
self._SO_writeLock = threading.Lock()
# If no connection was given, we'll inherit the class
# instance variable which should have a _connection
# attribute.
if connection is not None:
self._connection = connection
# Sometimes we need to know if this instance is
# global or tied to a particular connection.
# This flag tells us that:
self.sqlmeta._perConnection = True
if not selectResults:
dbNames = [col.dbName for col in self.sqlmeta.columnList]
selectResults = self._connection._SO_selectOne(self, dbNames)
if not selectResults:
raise SQLObjectNotFound, "The object %s by the ID %s does not exist" %
(self.__class__.__name__, self.id)
> self._SO_selectInit(selectResults)
[/home/bgolemon/SQLObject/sqlobject/main.py:933]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def _SO_selectInit(self, row):
for col, colValue in zip(self.sqlmeta.columnList, row):
if col.to_python:
> colValue = col.to_python(colValue, self._SO_validatorState)
[/home/bgolemon/SQLObject/sqlobject/main.py:1136]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def to_python(self, value, state):
if value is None:
return None
if isinstance(value, (DateTimeType, TimeType,
sqlbuilder.SQLExpression)):
return value
if isinstance(value, datetime.datetime):
return DateTime.DateTime(value.year, value.month, value.day,
value.hour, value.minute, value.second)
elif isinstance(value, datetime.date):
return DateTime.Date(value.year, value.month, value.day)
elif isinstance(value, datetime.time):
return DateTime.Time(value.hour, value.minute, value.second)
try:
stime = time.strptime(value, self.format)
except:
raise validators.Invalid("expected a date/time string of the '%s' format
in the DateTimeCol '%s', got %s %r instead" % \
E (self.format, self.name, type(value), value), value, state)
> Invalid: expected a date/time string of the '%H:%M:%S' format in the
DateTimeCol 'col3', got <type 'datetime.timedelta'>
datetime.timedelta(0, 42857) instead
[/home/bgolemon/SQLObject/sqlobject/col.py:1081]
________________________________________________________________________
_______
________________________________ entrypoint:
_________________________________
"""
from sqlobject import *
from sqlobject.tests.dbtest import *
from sqlobject.styles import Style, MixedCaseUnderscoreStyle,
MixedCaseStyle
> class SOStyleTest(SQLObject):
[/home/bgolemon/SQLObject/sqlobject/tests/test_default_style.py:9]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
class SOStyleTest(SQLObject):
a = StringCol()
st2 = ForeignKey('SOStyleTest2')
> class sqlmeta(sqlmeta):
[/home/bgolemon/SQLObject/sqlobject/tests/test_default_style.py:12]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
class sqlmeta(sqlmeta):
E style = AnotherStyle()
> NameError: name 'AnotherStyle' is not defined
[/home/bgolemon/SQLObject/sqlobject/tests/test_default_style.py:13]
________________________________________________________________________
_______
_________________________ entrypoint: test_exceptions
_________________________
def test_exceptions():
if not supports("exceptions"):
return
> setupClass(TestException)
[/home/bgolemon/SQLObject/sqlobject/tests/test_exceptions.py:15]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def setupClass(soClasses, force=False):
"""
Makes sure the classes have a corresponding and correct table.
This won't recreate the table if it already exists. It will check
that the table is properly defined (in case you change your table
definition).
You can provide a single class or a list of classes; if a list
then classes will be created in the order you provide, and
destroyed in the opposite order. So if class A depends on class
B, then do setupClass([B, A]) and B won't be destroyed or cleared
until after A is destroyed or cleared.
If force is true, then the database will be recreated no matter
what.
"""
global hub
if not isinstance(soClasses, (list, tuple)):
soClasses = [soClasses]
connection = getConnection()
for soClass in soClasses:
## This would be an alternate way to register connections...
#try:
# hub
#except NameError:
# hub = sqlobject.dbconnection.ConnectionHub()
#soClass._connection = hub
#hub.threadConnection = connection
#hub.processConnection = connection
soClass._connection = connection
> installOrClear(soClasses, force=force)
[/home/bgolemon/SQLObject/sqlobject/tests/dbtest.py:82]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def installOrClear(cls, soClasses, force=False):
cls.setup()
reversed = list(soClasses)[:]
reversed.reverse()
# If anything needs to be dropped, they all must be dropped
# But if we're forcing it, then we'll always drop
if force:
any_drops = True
else:
any_drops = False
for soClass in reversed:
table = soClass.sqlmeta.table
if not soClass._connection.tableExists(table):
continue
items = list(cls.selectBy(
tableName=table,
connectionURI=soClass._connection.uri()))
if items:
instance = items[0]
sql = instance.createSQL
else:
sql = None
newSQL, constraints = soClass.createTableSQL()
if sql != newSQL:
if sql is not None:
instance.destroySelf()
any_drops = True
break
for soClass in reversed:
if soClass._connection.tableExists(soClass.sqlmeta.table):
if any_drops:
cls.drop(soClass)
else:
cls.clear(soClass)
for soClass in soClasses:
table = soClass.sqlmeta.table
if not soClass._connection.tableExists(table):
> cls.install(soClass)
[/home/bgolemon/SQLObject/sqlobject/tests/dbtest.py:163]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def install(cls, soClass):
"""
Creates the given table in its database.
"""
sql = getattr(soClass, soClass._connection.dbName + 'Create',
None)
all_extra = []
if sql:
soClass._connection.query(sql)
else:
sql, extra_sql = soClass.createTableSQL()
> soClass.createTable(applyConstraints=False)
[/home/bgolemon/SQLObject/sqlobject/tests/dbtest.py:177]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def createTable(cls, ifNotExists=False, createJoinTables=True,
createIndexes=True, applyConstraints=True,
connection=None):
conn = connection or cls._connection
if ifNotExists and cls.tableExists(connection=conn):
return
extra_sql = []
post_funcs = []
cls.sqlmeta.send(events.CreateTableSignal, cls, connection,
extra_sql, post_funcs)
> constraints = conn.createTable(cls)
[/home/bgolemon/SQLObject/sqlobject/main.py:1396]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def createTable(self, soClass):
createSql, constraints = self.createTableSQL(soClass)
> self.query(createSql)
[/home/bgolemon/SQLObject/sqlobject/dbconnection.py:432]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def query(self, s):
> return self._runWithConnection(self._query, s)
[/home/bgolemon/SQLObject/sqlobject/dbconnection.py:336]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def _runWithConnection(self, meth, *args):
conn = self.getConnection()
try:
> val = meth(conn, *args)
[/home/bgolemon/SQLObject/sqlobject/dbconnection.py:249]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def _query(self, conn, s):
if self.debug:
self.printDebug(conn, s, 'Query')
> self._executeRetry(conn, conn.cursor(), s)
[/home/bgolemon/SQLObject/sqlobject/dbconnection.py:333]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def _executeRetry(self, conn, cursor, query):
# When a server connection is lost and a query is attempted, most of
# the time the query will raise a SERVER_LOST exception, then at the
# second attempt to execute it, the mysql lib will reconnect and
# succeed. However is a few cases, the first attempt raises the
# SERVER_GONE exception, the second attempt the SERVER_LOST exception
# and only the third succeeds. Thus the 3 in the loop count.
# If it doesn't reconnect even after 3 attempts, while the database is
# up and running, it is because a 5.0.3 (or newer) server is used
# which no longer permits autoreconnects by default. In that case a
# reconnect flag must be set when making the connection to indicate
# that autoreconnecting is desired. In MySQLdb 1.2.2 or newer this is
# done by calling ping(True) on the connection.
for count in range(3):
try:
return cursor.execute(query)
except MySQLdb.OperationalError, e:
if e.args[0] in (MySQLdb.constants.CR.SERVER_GONE_ERROR,
MySQLdb.constants.CR.SERVER_LOST):
if count == 2:
raise OperationalError(ErrorMessage(e))
if self.debug:
self.printDebug(conn, str(e), 'ERROR')
else:
E raise OperationalError(ErrorMessage(e))
> OperationalError: BLOB/TEXT column 'name' used in key specification
without a key length
[/home/bgolemon/SQLObject/sqlobject/mysql/mysqlconnection.py:119]
________________________________________________________________________
_______
__________________________ entrypoint: test_groupBy
___________________________
def test_groupBy():
setupClass(GroupbyTest)
GroupbyTest(name='a', value=1)
GroupbyTest(name='a', value=2)
GroupbyTest(name='b', value=1)
connection = getConnection()
select = Select([GroupbyTest.q.name, func.COUNT(GroupbyTest.q.value)],
groupBy=GroupbyTest.q.name,
orderBy=GroupbyTest.q.name)
sql = connection.sqlrepr(select)
rows = connection.queryAll(sql)
E assert rows == [('a', 2), ('b', 1)]
> assert (('a', 2L), ('b', 1L)) == [('a', 2), ('b', 1)]
[/home/bgolemon/SQLObject/sqlobject/tests/test_groupBy.py:25]
________________________________________________________________________
_______
________________________ entrypoint: test_groupBy_list
________________________
def test_groupBy_list():
setupClass(GroupbyTest)
GroupbyTest(name='a', value=1)
GroupbyTest(name='a', value=2)
GroupbyTest(name='b', value=1)
connection = getConnection()
select = Select([GroupbyTest.q.name, GroupbyTest.q.value],
groupBy=[GroupbyTest.q.name, GroupbyTest.q.value],
orderBy=[GroupbyTest.q.name, GroupbyTest.q.value])
sql = connection.sqlrepr(select)
rows = connection.queryAll(sql)
E assert rows == [('a', 1), ('a', 2), ('b', 1)]
> assert (('a', 1L), ('a', 2L), ('b', 1L)) == [('a', 1), ('a', 2), ('b',
1)]
[/home/bgolemon/SQLObject/sqlobject/tests/test_groupBy.py:39]
________________________________________________________________________
_______
________________________ entrypoint: test_index_get_1
_________________________
def test_index_get_1():
> setupClass(PersonIndexGet, force=True)
[/home/bgolemon/SQLObject/sqlobject/tests/test_indexes.py:53]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def setupClass(soClasses, force=False):
"""
Makes sure the classes have a corresponding and correct table.
This won't recreate the table if it already exists. It will check
that the table is properly defined (in case you change your table
definition).
You can provide a single class or a list of classes; if a list
then classes will be created in the order you provide, and
destroyed in the opposite order. So if class A depends on class
B, then do setupClass([B, A]) and B won't be destroyed or cleared
until after A is destroyed or cleared.
If force is true, then the database will be recreated no matter
what.
"""
global hub
if not isinstance(soClasses, (list, tuple)):
soClasses = [soClasses]
connection = getConnection()
for soClass in soClasses:
## This would be an alternate way to register connections...
#try:
# hub
#except NameError:
# hub = sqlobject.dbconnection.ConnectionHub()
#soClass._connection = hub
#hub.threadConnection = connection
#hub.processConnection = connection
soClass._connection = connection
> installOrClear(soClasses, force=force)
[/home/bgolemon/SQLObject/sqlobject/tests/dbtest.py:82]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def installOrClear(cls, soClasses, force=False):
cls.setup()
reversed = list(soClasses)[:]
reversed.reverse()
# If anything needs to be dropped, they all must be dropped
# But if we're forcing it, then we'll always drop
if force:
any_drops = True
else:
any_drops = False
for soClass in reversed:
table = soClass.sqlmeta.table
if not soClass._connection.tableExists(table):
continue
items = list(cls.selectBy(
tableName=table,
connectionURI=soClass._connection.uri()))
if items:
instance = items[0]
sql = instance.createSQL
else:
sql = None
newSQL, constraints = soClass.createTableSQL()
if sql != newSQL:
if sql is not None:
instance.destroySelf()
any_drops = True
break
for soClass in reversed:
if soClass._connection.tableExists(soClass.sqlmeta.table):
if any_drops:
cls.drop(soClass)
else:
cls.clear(soClass)
for soClass in soClasses:
table = soClass.sqlmeta.table
if not soClass._connection.tableExists(table):
> cls.install(soClass)
[/home/bgolemon/SQLObject/sqlobject/tests/dbtest.py:163]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def install(cls, soClass):
"""
Creates the given table in its database.
"""
sql = getattr(soClass, soClass._connection.dbName + 'Create',
None)
all_extra = []
if sql:
soClass._connection.query(sql)
else:
sql, extra_sql = soClass.createTableSQL()
> soClass.createTable(applyConstraints=False)
[/home/bgolemon/SQLObject/sqlobject/tests/dbtest.py:177]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def createTable(cls, ifNotExists=False, createJoinTables=True,
createIndexes=True, applyConstraints=True,
connection=None):
conn = connection or cls._connection
if ifNotExists and cls.tableExists(connection=conn):
return
extra_sql = []
post_funcs = []
cls.sqlmeta.send(events.CreateTableSignal, cls, connection,
extra_sql, post_funcs)
constraints = conn.createTable(cls)
if applyConstraints:
for constraint in constraints:
conn.query(constraint)
else:
extra_sql.extend(constraints)
if createJoinTables:
cls.createJoinTables(ifNotExists=ifNotExists,
connection=conn)
if createIndexes:
cls.createIndexes(ifNotExists=ifNotExists,
> connection=conn)
[/home/bgolemon/SQLObject/sqlobject/main.py:1407]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def createIndexes(cls, ifNotExists=False, connection=None):
conn = connection or cls._connection
for index in cls.sqlmeta.indexes:
if not index:
continue
> conn._SO_createIndex(cls, index)
[/home/bgolemon/SQLObject/sqlobject/main.py:1450]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def _SO_createIndex(self, soClass, index):
> self.query(self.createIndexSQL(soClass, index))
[/home/bgolemon/SQLObject/sqlobject/dbconnection.py:425]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def query(self, s):
> return self._runWithConnection(self._query, s)
[/home/bgolemon/SQLObject/sqlobject/dbconnection.py:336]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def _runWithConnection(self, meth, *args):
conn = self.getConnection()
try:
> val = meth(conn, *args)
[/home/bgolemon/SQLObject/sqlobject/dbconnection.py:249]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def _query(self, conn, s):
if self.debug:
self.printDebug(conn, s, 'Query')
> self._executeRetry(conn, conn.cursor(), s)
[/home/bgolemon/SQLObject/sqlobject/dbconnection.py:333]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def _executeRetry(self, conn, cursor, query):
# When a server connection is lost and a query is attempted, most of
# the time the query will raise a SERVER_LOST exception, then at the
# second attempt to execute it, the mysql lib will reconnect and
# succeed. However is a few cases, the first attempt raises the
# SERVER_GONE exception, the second attempt the SERVER_LOST exception
# and only the third succeeds. Thus the 3 in the loop count.
# If it doesn't reconnect even after 3 attempts, while the database is
# up and running, it is because a 5.0.3 (or newer) server is used
# which no longer permits autoreconnects by default. In that case a
# reconnect flag must be set when making the connection to indicate
# that autoreconnecting is desired. In MySQLdb 1.2.2 or newer this is
# done by calling ping(True) on the connection.
for count in range(3):
try:
return cursor.execute(query)
except MySQLdb.OperationalError, e:
if e.args[0] in (MySQLdb.constants.CR.SERVER_GONE_ERROR,
MySQLdb.constants.CR.SERVER_LOST):
if count == 2:
raise OperationalError(ErrorMessage(e))
if self.debug:
self.printDebug(conn, str(e), 'ERROR')
else:
E raise OperationalError(ErrorMessage(e))
> OperationalError: BLOB/TEXT column 'first_name' used in key
specification without a key length
[/home/bgolemon/SQLObject/sqlobject/mysql/mysqlconnection.py:119]
________________________________________________________________________
_______
________________________ entrypoint: test_index_get_2
_________________________
def test_index_get_2():
> setupClass([PersonIndexGet2, AddressIndexGet2])
[/home/bgolemon/SQLObject/sqlobject/tests/test_indexes.py:104]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def setupClass(soClasses, force=False):
"""
Makes sure the classes have a corresponding and correct table.
This won't recreate the table if it already exists. It will check
that the table is properly defined (in case you change your table
definition).
You can provide a single class or a list of classes; if a list
then classes will be created in the order you provide, and
destroyed in the opposite order. So if class A depends on class
B, then do setupClass([B, A]) and B won't be destroyed or cleared
until after A is destroyed or cleared.
If force is true, then the database will be recreated no matter
what.
"""
global hub
if not isinstance(soClasses, (list, tuple)):
soClasses = [soClasses]
connection = getConnection()
for soClass in soClasses:
## This would be an alternate way to register connections...
#try:
# hub
#except NameError:
# hub = sqlobject.dbconnection.ConnectionHub()
#soClass._connection = hub
#hub.threadConnection = connection
#hub.processConnection = connection
soClass._connection = connection
> installOrClear(soClasses, force=force)
[/home/bgolemon/SQLObject/sqlobject/tests/dbtest.py:82]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def installOrClear(cls, soClasses, force=False):
cls.setup()
reversed = list(soClasses)[:]
reversed.reverse()
# If anything needs to be dropped, they all must be dropped
# But if we're forcing it, then we'll always drop
if force:
any_drops = True
else:
any_drops = False
for soClass in reversed:
table = soClass.sqlmeta.table
if not soClass._connection.tableExists(table):
continue
items = list(cls.selectBy(
tableName=table,
connectionURI=soClass._connection.uri()))
if items:
instance = items[0]
sql = instance.createSQL
else:
sql = None
newSQL, constraints = soClass.createTableSQL()
if sql != newSQL:
if sql is not None:
instance.destroySelf()
any_drops = True
break
for soClass in reversed:
if soClass._connection.tableExists(soClass.sqlmeta.table):
if any_drops:
cls.drop(soClass)
else:
cls.clear(soClass)
for soClass in soClasses:
table = soClass.sqlmeta.table
if not soClass._connection.tableExists(table):
> cls.install(soClass)
[/home/bgolemon/SQLObject/sqlobject/tests/dbtest.py:163]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def install(cls, soClass):
"""
Creates the given table in its database.
"""
sql = getattr(soClass, soClass._connection.dbName + 'Create',
None)
all_extra = []
if sql:
soClass._connection.query(sql)
else:
sql, extra_sql = soClass.createTableSQL()
> soClass.createTable(applyConstraints=False)
[/home/bgolemon/SQLObject/sqlobject/tests/dbtest.py:177]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def createTable(cls, ifNotExists=False, createJoinTables=True,
createIndexes=True, applyConstraints=True,
connection=None):
conn = connection or cls._connection
if ifNotExists and cls.tableExists(connection=conn):
return
extra_sql = []
post_funcs = []
cls.sqlmeta.send(events.CreateTableSignal, cls, connection,
extra_sql, post_funcs)
> constraints = conn.createTable(cls)
[/home/bgolemon/SQLObject/sqlobject/main.py:1396]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def createTable(self, soClass):
createSql, constraints = self.createTableSQL(soClass)
> self.query(createSql)
[/home/bgolemon/SQLObject/sqlobject/dbconnection.py:432]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def query(self, s):
> return self._runWithConnection(self._query, s)
[/home/bgolemon/SQLObject/sqlobject/dbconnection.py:336]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def _runWithConnection(self, meth, *args):
conn = self.getConnection()
try:
> val = meth(conn, *args)
[/home/bgolemon/SQLObject/sqlobject/dbconnection.py:249]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def _query(self, conn, s):
if self.debug:
self.printDebug(conn, s, 'Query')
> self._executeRetry(conn, conn.cursor(), s)
[/home/bgolemon/SQLObject/sqlobject/dbconnection.py:333]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def _executeRetry(self, conn, cursor, query):
# When a server connection is lost and a query is attempted, most of
# the time the query will raise a SERVER_LOST exception, then at the
# second attempt to execute it, the mysql lib will reconnect and
# succeed. However is a few cases, the first attempt raises the
# SERVER_GONE exception, the second attempt the SERVER_LOST exception
# and only the third succeeds. Thus the 3 in the loop count.
# If it doesn't reconnect even after 3 attempts, while the database is
# up and running, it is because a 5.0.3 (or newer) server is used
# which no longer permits autoreconnects by default. In that case a
# reconnect flag must be set when making the connection to indicate
# that autoreconnecting is desired. In MySQLdb 1.2.2 or newer this is
# done by calling ping(True) on the connection.
for count in range(3):
try:
return cursor.execute(query)
except MySQLdb.OperationalError, e:
if e.args[0] in (MySQLdb.constants.CR.SERVER_GONE_ERROR,
MySQLdb.constants.CR.SERVER_LOST):
if count == 2:
raise OperationalError(ErrorMessage(e))
if self.debug:
self.printDebug(conn, str(e), 'ERROR')
else:
E raise OperationalError(ErrorMessage(e))
> OperationalError: BLOB/TEXT column 'name' used in key specification
without a key length
[/home/bgolemon/SQLObject/sqlobject/mysql/mysqlconnection.py:119]
________________________________________________________________________
_______
________________________________ entrypoint:
_________________________________
"""
from sqlobject import *
> sqlhub.processConnection =
connectionForURI('postgres://pgsql@/db_test')
[/home/bgolemon/SQLObject/sqlobject/tests/test_sqlobject_admin.py:7]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def connectionForURI(self, uri, **args):
if args:
if '?' not in uri:
uri += '?'
uri += urllib.urlencode(args)
if self.cachedURIs.has_key(uri):
return self.cachedURIs[uri]
if uri.find(':') != -1:
scheme, rest = uri.split(':', 1)
connCls = self.dbConnectionForScheme(scheme)
> conn = connCls.connectionFromURI(uri)
[/home/bgolemon/SQLObject/sqlobject/dbconnection.py:922]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def connectionFromURI(cls, uri):
user, password, host, port, path, args = cls._parseURI(uri)
path = path.strip('/')
if (host is None) and path.count('/'): # Non-default unix socket
path_parts = path.split('/')
host = '/' + '/'.join(path_parts[:-1])
path = path_parts[-1]
> return cls(host=host, port=port, db=path, user=user,
password=password, **args)
[/home/bgolemon/SQLObject/sqlobject/postgres/pgconnection.py:97]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def __init__(self, dsn=None, host=None, port=None, db=None,
user=None, password=None, usePygresql=False, unicodeCols=False,
**kw):
global psycopg, pgdb
self.usePygresql = usePygresql
if usePygresql:
if pgdb is None:
import pgdb
self.module = pgdb
else:
if psycopg is None:
try:
import psycopg2 as psycopg
except ImportError:
E import psycopg
> ImportError: No module named psycopg
[/home/bgolemon/SQLObject/sqlobject/postgres/pgconnection.py:29]
________________________________________________________________________
_______
_________________________ entrypoint: test_string_id
__________________________
def test_string_id():
conn = getConnection()
TestStringID.setConnection(conn)
TestStringID.dropTable(ifExists=True)
assert not conn.tableExists(TestStringID.sqlmeta.table)
> TestStringID.createTable()
[/home/bgolemon/SQLObject/sqlobject/tests/test_string_id.py:15]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def createTable(cls, ifNotExists=False, createJoinTables=True,
createIndexes=True, applyConstraints=True,
connection=None):
conn = connection or cls._connection
if ifNotExists and cls.tableExists(connection=conn):
return
extra_sql = []
post_funcs = []
cls.sqlmeta.send(events.CreateTableSignal, cls, connection,
extra_sql, post_funcs)
> constraints = conn.createTable(cls)
[/home/bgolemon/SQLObject/sqlobject/main.py:1396]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def createTable(self, soClass):
createSql, constraints = self.createTableSQL(soClass)
> self.query(createSql)
[/home/bgolemon/SQLObject/sqlobject/dbconnection.py:432]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def query(self, s):
> return self._runWithConnection(self._query, s)
[/home/bgolemon/SQLObject/sqlobject/dbconnection.py:336]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def _runWithConnection(self, meth, *args):
conn = self.getConnection()
try:
> val = meth(conn, *args)
[/home/bgolemon/SQLObject/sqlobject/dbconnection.py:249]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def _query(self, conn, s):
if self.debug:
self.printDebug(conn, s, 'Query')
> self._executeRetry(conn, conn.cursor(), s)
[/home/bgolemon/SQLObject/sqlobject/dbconnection.py:333]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def _executeRetry(self, conn, cursor, query):
# When a server connection is lost and a query is attempted, most of
# the time the query will raise a SERVER_LOST exception, then at the
# second attempt to execute it, the mysql lib will reconnect and
# succeed. However is a few cases, the first attempt raises the
# SERVER_GONE exception, the second attempt the SERVER_LOST exception
# and only the third succeeds. Thus the 3 in the loop count.
# If it doesn't reconnect even after 3 attempts, while the database is
# up and running, it is because a 5.0.3 (or newer) server is used
# which no longer permits autoreconnects by default. In that case a
# reconnect flag must be set when making the connection to indicate
# that autoreconnecting is desired. In MySQLdb 1.2.2 or newer this is
# done by calling ping(True) on the connection.
for count in range(3):
try:
return cursor.execute(query)
except MySQLdb.OperationalError, e:
if e.args[0] in (MySQLdb.constants.CR.SERVER_GONE_ERROR,
MySQLdb.constants.CR.SERVER_LOST):
if count == 2:
raise OperationalError(ErrorMessage(e))
if self.debug:
self.printDebug(conn, str(e), 'ERROR')
else:
E raise OperationalError(ErrorMessage(e))
> OperationalError: BLOB/TEXT column 'test_id_here' used in key
specification without a key length
[/home/bgolemon/SQLObject/sqlobject/mysql/mysqlconnection.py:119]
________________________________________________________________________
_______
____________________________ entrypoint: testGetVP
____________________________
def testGetVP():
checkAttr(ViewPhone, phones[0].id, 'number', phones[0].number)
> checkAttr(ViewPhone, phones[0].id, 'minutes',
phones[0].calls.sum(PhoneCall.q.minutes))
[/home/bgolemon/SQLObject/sqlobject/tests/test_views.py:102]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def checkAttr(cls, id, attr, value):
E assert getattr(cls.get(id), attr) == value
> assert 25 == '25'
+ where 25 = getattr(<ViewPhone 7 minutes=25 numberOfCalls=2L
number='1234567890' phoneNumberID=7L>, 'minutes')
+ where <ViewPhone 7 minutes=25 numberOfCalls=2L number='1234567890'
phoneNumberID=7L> = <class
'sqlobject.tests.test_views.ViewPhone'>.get(7L)
[/home/bgolemon/SQLObject/sqlobject/tests/test_views.py:90]
________________________________________________________________________
_______
___________________________ entrypoint: testGetVPM
____________________________
def testGetVPM():
checkAttr(ViewPhoneMore, phones[0].id, 'number', phones[0].number)
> checkAttr(ViewPhoneMore, phones[0].id, 'minutesCalled',
phones[0].incoming.sum(PhoneCall.q.minutes))
[/home/bgolemon/SQLObject/sqlobject/tests/test_views.py:107]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def checkAttr(cls, id, attr, value):
E assert getattr(cls.get(id), attr) == value
> assert 35 == '35'
+ where 35 = getattr(<ViewPhoneMore 7 number='1234567890' timesCalled=2L
timesCalledLong=1L minutesCalled=35>, 'minutesCalled')
+ where <ViewPhoneMore 7 number='1234567890' timesCalled=2L
timesCalledLong=1L minutesCalled=35> = <class
'sqlobject.tests.test_views.ViewPhoneMore'>.get(7L)
[/home/bgolemon/SQLObject/sqlobject/tests/test_views.py:90]
________________________________________________________________________
_______
_______________________ entrypoint: testInnerAggregate
________________________
def testInnerAggregate():
> checkAttr(ViewPhoneInnerAggregate, phones[0].id, 'twiceMinutes',
phones[0].calls.sum(PhoneCall.q.minutes)*2)
[/home/bgolemon/SQLObject/sqlobject/tests/test_views.py:118]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_ _ _ _
def checkAttr(cls, id, attr, value):
E assert getattr(cls.get(id), attr) == value
> assert 50 == '2525'
+ where 50 = getattr(<ViewPhoneInnerAggregate 7 numberOfCalls=2L
minutes=25 number='1234567890' phoneNumberID=7L twiceMinutes=50>,
'twiceMinutes')
+ where <ViewPhoneInnerAggregate 7 numberOfCalls=2L minutes=25
number='1234567890' phoneNumberID=7L twiceMinutes=50> = <class
'sqlobject.tests.test_views.ViewPhoneInnerAggregate'>.get(7L)
[/home/bgolemon/SQLObject/sqlobject/tests/test_views.py:90]
________________________________________________________________________
_______
=========== tests finished: 213 passed, 16 failed in 158.12 seconds
===========
--Buck
-------------------------------------------------------------------------
Check out the new SourceForge.net Marketplace.
It's the best place to buy or sell services for
just about anything Open Source.
http://sourceforge.net/services/buy/index.php
_______________________________________________
sqlobject-discuss mailing list
sqlobject-discuss@lists.sourceforge.net
https://lists.sourceforge.net/lists/listinfo/sqlobject-discuss
