adapter = sql_util.ClauseAdapter(target)
def col_label(col):
- adapted = adapter.traverse(c)
- if isinstance(c, expression._Label):
+ adapted = adapter.traverse(col)
+ if isinstance(col, expression._Label):
return adapted.label(c.key)
else:
return self.label_select_column(None, adapted, asfrom=False)
ddl_events = ('before-create', 'after-create', 'before-drop', 'after-drop')
- def __new__(cls, name, metadata, *args, **kw):
+ def __new__(cls, *args, **kw):
+ if not args:
+ # python3k pickle seems to call this
+ return object.__new__(cls)
+
+ try:
+ name, metadata, args = args[0], args[1], args[2:]
+ except IndexError:
+ raise TypeError("Table() takes at least two arguments")
+
schema = kw.get('schema', None)
useexisting = kw.pop('useexisting', False)
mustexist = kw.pop('mustexist', False)
from cStringIO import StringIO
from sqlalchemy.test import config, assertsql, util as testutil
-from sqlalchemy.util import function_named
+from sqlalchemy.util import function_named, py3k
from engines import drop_all_tables
from sqlalchemy import exc as sa_exc, util, types as sqltypes, schema, pool
from nose import SkipTest
+
_ops = { '<': operator.lt,
'>': operator.gt,
'==': operator.eq,
c = clause.compile(dialect=dialect, **kw)
+ # Py3K
+ ## I kid you not.
+ ##
+ ## 1. Doesn't work:
+ ## http://mail.python.org/pipermail/python-3000/2008-February/012144.html
+ ##
+ ## 2. no more setdefaultencoding(). (although this is undocumented)
+ ##
+ ## 3. Therefore:
+ ## http://docs.python.org/3.1/library/sys.html#sys.stdin
+ ##
+ #sys.stdout.buffer.write(("\nSQL String:\n" + str(c) + repr(getattr(c, 'params', {}))).encode('utf-8'))
+ # Py2K
print "\nSQL String:\n" + str(c) + repr(getattr(c, 'params', {}))
-
+ # end Py2K
+
cc = re.sub(r'[\n\t]', '', str(c))
-
+
eq_(cc, result, "%r != %r on dialect %r" % (cc, result, dialect))
if checkparams is not None:
def dictlike_iteritems(dictlike):
"""Return a (key, value) iterator for almost any dict-like object."""
+ # Py3K
+ #if hasattr(dictlike, 'items'):
+ # return dictlike.items()
+ # Py2K
if hasattr(dictlike, 'iteritems'):
return dictlike.iteritems()
elif hasattr(dictlike, 'items'):
return iter(dictlike.items())
+ # end Py2K
getter = getattr(dictlike, '__getitem__', getattr(dictlike, 'get', None))
if getter is None:
if len(self) < len(other):
return False
- for m in itertools.ifilterfalse(self._members.has_key,
+ for m in itertools.ifilterfalse(self._members.__contains__,
other._members.iterkeys()):
return False
return True
Column('c1', Integer, primary_key=True),
Column('c2', String(30)))
- @profiling.function_call_count(72, {'2.4': 45, '3.0':77})
+ @profiling.function_call_count(72, {'2.4': 45, '3.0':77, '3.1':77})
def test_insert(self):
t1.insert().compile()
def test_update(self):
t1.update().compile()
- @profiling.function_call_count(195, versions={'2.4':118, '3.0':208})
+ @profiling.function_call_count(195, versions={'2.4':118, '3.0':208, '3.1':208})
def test_select(self):
s = select([t1], t1.c.c2==t2.c.c1)
s.compile()
use_threadlocal=True)
- @profiling.function_call_count(54, {'2.4': 36, '3.0':57})
+ @profiling.function_call_count(54, {'2.4': 36, '3.0':57, '3.1':57})
def test_first_connect(self):
conn = pool.connect()
def test_second_samethread_connect(self):
conn = pool.connect()
- @profiling.function_call_count(5, {'2.4': 3, '3.0':6})
+ @profiling.function_call_count(5, {'2.4': 3, '3.0':6, '3.1':6})
def go():
return pool.connect()
c2 = go()
def test_object(self):
self._notok(object())
+ # Py2K
def test_duck_1(self):
class duck1(object):
def iteritems(duck):
return iter(self.baseline)
self._ok(duck1())
+ # end Py2K
- # Py2K
def test_duck_2(self):
class duck2(object):
def items(duck):
return list(self.baseline)
self._ok(duck2())
- # end Py2K
# Py2K
def test_duck_3(self):
errors, threads = [], []
for i in xrange(thread_count):
- thread = threading.Thread(target=self.overlap,
+ thrd = threading.Thread(target=self.overlap,
args=(groups.pop(0), errors, update_style))
- thread.start()
- threads.append(thread)
- for thread in threads:
- thread.join()
+ thrd.start()
+ threads.append(thrd)
+ for thrd in threads:
+ thrd.join()
return errors
# ensure that examples with external dependencies are not run if those dependencies are
# not present (i.e. elementtree, postgis)
def test_examples(self):
- pass
+ pass
#for module in find_modules():
# check_import.description = module
# yield check_import, module