any setuptools-installed installations (setuptools places .egg files ahead of
plain directories, even if on PYTHONPATH, unfortunately).
+NEW INSTRUCTIONS FOR ABSOLUTE MODULE IMPORTS
+--------------------------------------------
+Py3K imports modules absolutely. This means the old way of running unittests,
+such as "python test/sql/query.py", can break things badly. How ? By name
+conflicts in our test suite. We have a unit test called "test/sql/select.py".
+The pg8000 DBAPI uses the Python "select" module. If you run a script
+called "test/sql/query.py", now "test/sql" is part of the import path, and
+pg8000 pulls in "test/sql/select.py" instead. Kaboom ! So we run
+the tests using -m now.
RUNNING ALL TESTS
-----------------
To run all tests:
- $ python test/alltests.py
+ $ python -m alltests
RUNNING INDIVIDUAL TESTS
-------------------------
Any unittest module can be run directly from the module file:
- python test/orm/mapper.py
+ python -m orm.mapper
To run a specific test within the module, specify it as ClassName.methodname:
- python test/orm/mapper.py MapperTest.testget
+ python -m orm.mapper MapperTest.testget
COMMAND LINE OPTIONS
--------------------
Help is available via --help
- $ python test/alltests.py --help
+ $ python -m alltests --help
usage: alltests.py [options] [tests...]
If you'll be running the tests frequently, database aliases can save a lot of
typing. The --dbs option lists the built-in aliases and their matching URLs:
- $ python test/alltests.py --dbs
+ $ python -m alltests --dbs
Available --db options (use --dburi to override)
mysql mysql://scott:tiger@127.0.0.1:3306/test
oracle oracle://scott:tiger@127.0.0.1:1521
To run tests against an aliased database:
- $ python test/alltests.py --db=postgres
+ $ python -m alltests --db=postgres
To customize the URLs with your own users or hostnames, make a simple .ini
file called `test.cfg` at the top level of the SQLAlchemy source distribution
Any log target can be directed to the console with command line options, such
as:
- $ python test/orm/unitofwork.py --log-info=sqlalchemy.orm.mapper \
+ $ python -m orm.unitofwork --log-info=sqlalchemy.orm.mapper \
--log-debug=sqlalchemy.pool --log-info=sqlalchemy.engine
This would log mapper configuration, connection pool checkouts, and SQL
source file can be generated, marking statements that are executed with > and
statements that are missed with !, by running the coverage.py utility with the
"-a" (annotate) option, such as:
-
+
+ # TODO: need to adjust this for -m flag
$ python ./test/testlib/coverage.py -a ./lib/sqlalchemy/sql.py
This will create a new annotated file ./lib/sqlalchemy/sql.py,cover. Pretty
exc = "select nextval('\"%s\".\"%s_%s_seq\"')" % (sch, column.table.name, column.name)
else:
exc = "select nextval('\"%s_%s_seq\"')" % (column.table.name, column.name)
- return self.execute_string(exc.encode(self.dialect.encoding))
+
+ if self.dialect.supports_unicode_statements:
+ return self.execute_string(exc)
+ else:
+ return self.execute_string(exc.encode(self.dialect.encoding))
return super(PGDefaultRunner, self).get_column_default(column)
if table.schema is not None:
schema_where_clause = "n.nspname = :schema"
schemaname = table.schema
+
+ # Py2K
if isinstance(schemaname, str):
schemaname = schemaname.decode(self.encoding)
+ # end Py2K
else:
schema_where_clause = "pg_catalog.pg_table_is_visible(c.oid)"
schemaname = None
s = sql.text(SQL_COLS, bindparams=[sql.bindparam('table_name', type_=sqltypes.Unicode), sql.bindparam('schema', type_=sqltypes.Unicode)], typemap={'attname':sqltypes.Unicode, 'default':sqltypes.Unicode})
tablename = table.name
+ # Py2K
if isinstance(tablename, str):
tablename = tablename.decode(self.encoding)
+ # end Py2K
c = connection.execute(s, table_name=tablename, schema=schemaname)
rows = c.fetchall()
default_paramstyle = 'qmark'
poolclass = pool.SingletonThreadPool
execution_ctx_cls = SQLite_pysqliteExecutionContext
+ description_encoding = None
driver = 'pysqlite'
def __init__(self, **kwargs):
for i, item in enumerate(metadata):
colname = item[0]
- # Py2K
if self.dialect.description_encoding:
colname = colname.decode(self.dialect.description_encoding)
- # end Py2K
if '.' in colname:
# sqlite will in some circumstances prepend table name to colnames, so strip
supports_alter = True
supports_sequences = False
sequences_optional = False
+
+ # Py3K
+ #supports_unicode_statements = True
+ #supports_unicode_binds = True
+ # Py2K
supports_unicode_statements = False
supports_unicode_binds = False
+ # end Py2K
name = 'default'
max_identifier_length = 9999
if label_length and label_length > self.max_identifier_length:
raise exc.ArgumentError("Label length of %d is greater than this dialect's maximum identifier length of %d" % (label_length, self.max_identifier_length))
self.label_length = label_length
- self.description_encoding = getattr(self, 'description_encoding', encoding)
+
+ if not hasattr(self, 'description_encoding'):
+ self.description_encoding = getattr(self, 'description_encoding', encoding)
+
+ # Py3K
+ #self.supports_unicode_statements = True
+ #self.supports_unicode_binds = True
@classmethod
def type_descriptor(cls, typeobj):
import weakref, time, threading
from sqlalchemy import exc, log
-from sqlalchemy import queue as Queue
+from sqlalchemy import queue as sqla_queue
from sqlalchemy.util import thread, threading, pickle, as_interface
proxies = {}
"""
Pool.__init__(self, creator, **params)
- self._pool = Queue.Queue(pool_size)
+ self._pool = sqla_queue.Queue(pool_size)
self._overflow = 0 - pool_size
self._max_overflow = max_overflow
self._timeout = timeout
def do_return_conn(self, conn):
try:
self._pool.put(conn, False)
- except Queue.Full:
+ except sqla_queue.Full:
if self._overflow_lock is None:
self._overflow -= 1
else:
try:
wait = self._max_overflow > -1 and self._overflow >= self._max_overflow
return self._pool.get(wait, self._timeout)
- except Queue.Empty:
+ except sqla_queue.Empty:
if self._max_overflow > -1 and self._overflow >= self._max_overflow:
if not wait:
return self.do_get()
try:
conn = self._pool.get(False)
conn.close()
- except Queue.Empty:
+ except sqla_queue.Empty:
break
self._overflow = 0 - self.size()
sys.path.insert(0, os.path.join(os.getcwd(), 'lib'))
logging.basicConfig()
- testlib.config.configure()
_setup = True
+ testlib.config.configure()
def simple_setup():
"""import testenv; testenv.simple_setup()"""
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
