1. Mike Bayer
  2. sqlalchemy

Commits

Mike Bayer  committed 756aa27

first step of [ticket:1949], remove the setuptools aspect
of the plugin, move it to test/bootstrap

  • Participants
  • Parent commits 40d5a32
  • Branches master

Comments (0)

Files changed (14)

File README.py3k

View file
  • Ignore whitespace
 Running Tests
 -------------
 
-To run the unit tests, ensure Distribute is installed as above,
-and also that at least the ./lib/ and ./test/ directories have been converted
-to Python 3 using the source tool above.   A Python 3 version of Nose
-can be acquired from Bitbucket using Mercurial:
-
-    hg clone http://bitbucket.org/jpellerin/nose3/
-    cd nose3
-    python3 setup.py install
-
-The tests can then be run using the "nosetests3" script installed by the above,
-using the same instructions in README.unittests.
+To run unit tests in Py3k, Nose 1.0 is required, or a development
+version of Nose that supports Python 3.   The tests are run
+using ./sqla_nose.py as described in README.unittests.
 
 Current 3k Issues
 -----------------

File README.unittests

View file
  • Ignore whitespace
     $ easy_install nose
 
 SQLAlchemy implements a nose plugin that must be present when tests are run.
-This plugin is available when SQLAlchemy is installed via setuptools.
+This plugin is invoked when the test runner script provided with
+SQLAlchemy is used.
 
-INSTANT TEST RUNNER
--------------------
-
-A plain vanilla run of all tests using sqlite can be run via setup.py:
-
-    $ python setup.py test
-    
-(NOTE: this command is broken for Python 2.7 with nose 0.11.3, see 
-Nose issue 340.  You will need to use 'nosetests' directly, see below.)
-    
-Setuptools will take care of the rest !   To run nose directly and have
-its full set of options available, read on...
-
-SETUP
------
-
-All that's required is for SQLAlchemy to be installed via setuptools.
-For example, to create a local install in a source distribution directory:
-
-    $ export PYTHONPATH=.
-    $ python setup.py develop -d .
-
-The above will create a setuptools "development" distribution in the local
-path, which allows the Nose plugin to be available when nosetests is run.
-The plugin is enabled using the "with-sqlalchemy=True" configuration
-in setup.cfg.
+**NOTE:** - the nose plugin is no longer installed by setuptools as of 
+version 0.7 !  Please use sqla_nose.py to run tests.
 
 RUNNING ALL TESTS
 -----------------
 To run all tests:
 
-    $ nosetests
+    $ ./sqla_nose.py
 
 (NOTE: if running with Python 2.7 and nose 0.11.3, add "-w test/" to the command.
 Again this is a Nose issue, see Nose issue 342.)
 
 If you're running the tests on Microsoft Windows, then there is an additional
-argument that must be passed to nosetests:
+argument that must be passed to ./sqla_nose.py:
 
-    > nosetests --first-package-wins=True
+    > ./sqla_nose.py --first-package-wins=True
 
 This is required because nose’s importer will normally evict a package from
 sys.modules if it sees a package with the same name in a different location.
 Assuming all tests pass, this is a very unexciting output.  To make it more 
 intersesting:
 
-    $ nosetests -v
-
-ALTERNATE TEST RUNNER
----------------------
-
-The script "sqla_nose.py" is a front-end to Nose which manually associates
-the SQLAlchemy testing plugin with Nose at runtime.   This script can run the 
-tests without any reliance upon setuptools.   In 0.7 we'll be removing the 
-Nose plugin from setup, so this will be the way going forward to run tests:
-
-    $ python sqla_nose.py -v
+    $ ./sqla_nose.py -v
 
 RUNNING INDIVIDUAL TESTS
 -------------------------
 Any directory of test modules can be run at once by specifying the directory
 path:
 
-    $ nosetest test/dialect
+    $ ./sqla_nose.py test/dialect
 
 Any test module can be run directly by specifying its module name:
 
-    $ nosetests test.orm.test_mapper
+    $ ./sqla_nose.py test.orm.test_mapper
 
 To run a specific test within the module, specify it as module:ClassName.methodname:
 
-    $ nosetests test.orm.test_mapper:MapperTest.test_utils
+    $ ./sqla_nose.py test.orm.test_mapper:MapperTest.test_utils
 
 
 COMMAND LINE OPTIONS
 --------------------
 Help is available via --help:
 
-    $ nosetests --help
+    $ ./sqla_nose.py --help
 
 The --help screen is a combination of common nose options and options which 
 the SQLAlchemy nose plugin adds.  The most commonly SQLAlchemy-specific 
 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:
 
-    $ nosetests --dbs
+    $ ./sqla_nose.py --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:
 
-    $ nosetests --db=postgresql
+    $ ./sqla_nose.py --db=postgresql
 
 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:
 
-    $ nosetests test.orm.unitofwork --log-info=sqlalchemy.orm.mapper \
+    $ ./sqla_nose.py test.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
 Coverage is tracked using Nose's coverage plugin.   See the nose 
 documentation for details.  Basic usage is:
 
-    $ nosetests test.sql.test_query --with-coverage
+    $ ./sqla_nose.py test.sql.test_query --with-coverage
 
 BIG COVERAGE TIP !!!  There is an issue where existing .pyc files may
 store the incorrect filepaths, which will break the coverage system.  If

File lib/sqlalchemy/test/__init__.py

View file
  • Ignore whitespace
 
 """
 
-from sqlalchemy_nose import config
+from test.bootstrap import config
 from sqlalchemy.test import testing, engines, requires, profiling, pickleable
 from sqlalchemy.test.schema import Column, Table
 from sqlalchemy.test.testing import \

File lib/sqlalchemy/test/engines.py

View file
  • Ignore whitespace
 import sys, types, weakref
 from collections import deque
-from sqlalchemy_nose import config
+from test.bootstrap import config
 from sqlalchemy.util import function_named, callable
 import re
 import warnings

File lib/sqlalchemy/test/profiling.py

View file
  • Ignore whitespace
 """
 
 import os, sys
-from sqlalchemy_nose import config
 from sqlalchemy.test.util import function_named, gc_collect
 from nose import SkipTest
 

File lib/sqlalchemy/test/testing.py

View file
  • Ignore whitespace
 import warnings
 from cStringIO import StringIO
 
-from sqlalchemy_nose import config
+from test.bootstrap import config
 from sqlalchemy.test import assertsql, util as testutil
 from sqlalchemy.util import function_named, py3k
 from engines import drop_all_tables

File lib/sqlalchemy_nose/__init__.py

  • Ignore whitespace
Empty file removed.

File lib/sqlalchemy_nose/config.py

  • Ignore whitespace
-import optparse, os, sys, re, ConfigParser, time, warnings
-
-# 2to3
-import StringIO
-
-logging = None
-
-__all__ = 'parser', 'configure', 'options',
-
-db = None
-db_label, db_url, db_opts = None, None, {}
-
-options = None
-file_config = None
-
-base_config = """
-[db]
-sqlite=sqlite:///:memory:
-sqlite_file=sqlite:///querytest.db
-postgresql=postgresql://scott:tiger@127.0.0.1:5432/test
-postgres=postgresql://scott:tiger@127.0.0.1:5432/test
-pg8000=postgresql+pg8000://scott:tiger@127.0.0.1:5432/test
-postgresql_jython=postgresql+zxjdbc://scott:tiger@127.0.0.1:5432/test
-mysql_jython=mysql+zxjdbc://scott:tiger@127.0.0.1:5432/test
-mysql=mysql://scott:tiger@127.0.0.1:3306/test
-oracle=oracle://scott:tiger@127.0.0.1:1521
-oracle8=oracle://scott:tiger@127.0.0.1:1521/?use_ansi=0
-mssql=mssql://scott:tiger@SQUAWK\\SQLEXPRESS/test
-firebird=firebird://sysdba:masterkey@localhost//tmp/test.fdb
-maxdb=maxdb://MONA:RED@/maxdb1
-"""
-
-def _log(option, opt_str, value, parser):
-    global logging
-    if not logging:
-        import logging
-        logging.basicConfig()
-
-    if opt_str.endswith('-info'):
-        logging.getLogger(value).setLevel(logging.INFO)
-    elif opt_str.endswith('-debug'):
-        logging.getLogger(value).setLevel(logging.DEBUG)
-
-
-def _list_dbs(*args):
-    print "Available --db options (use --dburi to override)"
-    for macro in sorted(file_config.options('db')):
-        print "%20s\t%s" % (macro, file_config.get('db', macro))
-    sys.exit(0)
-
-def _server_side_cursors(options, opt_str, value, parser):
-    db_opts['server_side_cursors'] = True
-
-def _engine_strategy(options, opt_str, value, parser):
-    if value:
-        db_opts['strategy'] = value
-
-class _ordered_map(object):
-    def __init__(self):
-        self._keys = list()
-        self._data = dict()
-
-    def __setitem__(self, key, value):
-        if key not in self._keys:
-            self._keys.append(key)
-        self._data[key] = value
-
-    def __iter__(self):
-        for key in self._keys:
-            yield self._data[key]
-
-# at one point in refactoring, modules were injecting into the config
-# process.  this could probably just become a list now.
-post_configure = _ordered_map()
-
-def _engine_uri(options, file_config):
-    global db_label, db_url
-    db_label = 'sqlite'
-    if options.dburi:
-        db_url = options.dburi
-        db_label = db_url[:db_url.index(':')]
-    elif options.db:
-        db_label = options.db
-        db_url = None
-
-    if db_url is None:
-        if db_label not in file_config.options('db'):
-            raise RuntimeError(
-                "Unknown engine.  Specify --dbs for known engines.")
-        db_url = file_config.get('db', db_label)
-post_configure['engine_uri'] = _engine_uri
-
-def _require(options, file_config):
-    if not(options.require or
-           (file_config.has_section('require') and
-            file_config.items('require'))):
-        return
-
-    try:
-        import pkg_resources
-    except ImportError:
-        raise RuntimeError("setuptools is required for version requirements")
-
-    cmdline = []
-    for requirement in options.require:
-        pkg_resources.require(requirement)
-        cmdline.append(re.split('\s*(<!>=)', requirement, 1)[0])
-
-    if file_config.has_section('require'):
-        for label, requirement in file_config.items('require'):
-            if not label == db_label or label.startswith('%s.' % db_label):
-                continue
-            seen = [c for c in cmdline if requirement.startswith(c)]
-            if seen:
-                continue
-            pkg_resources.require(requirement)
-post_configure['require'] = _require
-
-def _engine_pool(options, file_config):
-    if options.mockpool:
-        from sqlalchemy import pool
-        db_opts['poolclass'] = pool.AssertionPool
-post_configure['engine_pool'] = _engine_pool
-
-def _create_testing_engine(options, file_config):
-    from sqlalchemy.test import engines, testing
-    global db
-    db = engines.testing_engine(db_url, db_opts)
-    testing.db = db
-post_configure['create_engine'] = _create_testing_engine
-
-def _prep_testing_database(options, file_config):
-    from sqlalchemy.test import engines
-    from sqlalchemy import schema
-
-    # also create alt schemas etc. here?
-    if options.dropfirst:
-        e = engines.utf8_engine()
-        existing = e.table_names()
-        if existing:
-            print "Dropping existing tables in database: " + db_url
-            try:
-                print "Tables: %s" % ', '.join(existing)
-            except:
-                pass
-            print "Abort within 5 seconds..."
-            time.sleep(5)
-            md = schema.MetaData(e, reflect=True)
-            md.drop_all()
-        e.dispose()
-
-post_configure['prep_db'] = _prep_testing_database
-
-def _set_table_options(options, file_config):
-    from sqlalchemy.test import schema
-
-    table_options = schema.table_options
-    for spec in options.tableopts:
-        key, value = spec.split('=')
-        table_options[key] = value
-
-    if options.mysql_engine:
-        table_options['mysql_engine'] = options.mysql_engine
-post_configure['table_options'] = _set_table_options
-
-def _reverse_topological(options, file_config):
-    if options.reversetop:
-        from sqlalchemy.orm import unitofwork, session, mapper, dependency
-        from sqlalchemy import topological
-        from sqlalchemy.test.util import RandomSet
-        topological.set = unitofwork.set = session.set = mapper.set = dependency.set = RandomSet
-post_configure['topological'] = _reverse_topological
-

File lib/sqlalchemy_nose/noseplugin.py

  • Ignore whitespace
-import logging
-import os
-import re
-import sys
-import time
-import warnings
-import ConfigParser
-import StringIO
-
-import nose.case
-from nose.plugins import Plugin
-
-from sqlalchemy_nose import config
-
-from sqlalchemy_nose.config import (
-    _create_testing_engine, _engine_pool, _engine_strategy, _engine_uri, _list_dbs, _log,
-    _prep_testing_database, _require, _reverse_topological, _server_side_cursors,
-    _set_table_options, base_config, db, db_label, db_url, file_config, post_configure)
-
-log = logging.getLogger('nose.plugins.sqlalchemy')
-
-class NoseSQLAlchemy(Plugin):
-    """
-    Handles the setup and extra properties required for testing SQLAlchemy
-    """
-    enabled = True
-    name = 'sqlalchemy'
-    score = 100
-
-    def options(self, parser, env=os.environ):
-        Plugin.options(self, parser, env)
-        opt = parser.add_option
-        opt("--log-info", action="callback", type="string", callback=_log,
-            help="turn on info logging for <LOG> (multiple OK)")
-        opt("--log-debug", action="callback", type="string", callback=_log,
-            help="turn on debug logging for <LOG> (multiple OK)")
-        opt("--require", action="append", dest="require", default=[],
-            help="require a particular driver or module version (multiple OK)")
-        opt("--db", action="store", dest="db", default="sqlite",
-            help="Use prefab database uri")
-        opt('--dbs', action='callback', callback=_list_dbs,
-            help="List available prefab dbs")
-        opt("--dburi", action="store", dest="dburi",
-            help="Database uri (overrides --db)")
-        opt("--dropfirst", action="store_true", dest="dropfirst",
-            help="Drop all tables in the target database first (use with caution on Oracle, "
-            "MS-SQL)")
-        opt("--mockpool", action="store_true", dest="mockpool",
-            help="Use mock pool (asserts only one connection used)")
-        opt("--enginestrategy", action="callback", type="string",
-            callback=_engine_strategy,
-            help="Engine strategy (plain or threadlocal, defaults to plain)")
-        opt("--reversetop", action="store_true", dest="reversetop", default=False,
-            help="Use a random-ordering set implementation in the ORM (helps "
-                  "reveal dependency issues)")
-        opt("--unhashable", action="store_true", dest="unhashable", default=False,
-            help="Disallow SQLAlchemy from performing a hash() on mapped test objects.")
-        opt("--noncomparable", action="store_true", dest="noncomparable", default=False,
-            help="Disallow SQLAlchemy from performing == on mapped test objects.")
-        opt("--truthless", action="store_true", dest="truthless", default=False,
-            help="Disallow SQLAlchemy from truth-evaluating mapped test objects.")
-        opt("--serverside", action="callback", callback=_server_side_cursors,
-            help="Turn on server side cursors for PG")
-        opt("--mysql-engine", action="store", dest="mysql_engine", default=None,
-            help="Use the specified MySQL storage engine for all tables, default is "
-                 "a db-default/InnoDB combo.")
-        opt("--table-option", action="append", dest="tableopts", default=[],
-            help="Add a dialect-specific table option, key=value")
-
-        global file_config
-        file_config = ConfigParser.ConfigParser()
-        file_config.readfp(StringIO.StringIO(base_config))
-        file_config.read(['test.cfg', os.path.expanduser('~/.satest.cfg')])
-        config.file_config = file_config
-        
-    def configure(self, options, conf):
-        Plugin.configure(self, options, conf)
-        self.options = options
-        
-    def begin(self):
-        global testing, requires, util
-        from sqlalchemy.test import testing, requires
-        from sqlalchemy import util
-        
-        testing.db = db
-        testing.requires = requires
-
-        # Lazy setup of other options (post coverage)
-        for fn in post_configure:
-            fn(self.options, file_config)
-        
-    def describeTest(self, test):
-        return ""
-        
-    def wantClass(self, cls):
-        """Return true if you want the main test selector to collect
-        tests from this class, false if you don't, and None if you don't
-        care.
-
-        :Parameters:
-           cls : class
-             The class being examined by the selector
-
-        """
-
-        if not issubclass(cls, testing.TestBase):
-            return False
-        else:
-            if (hasattr(cls, '__whitelist__') and testing.db.name in cls.__whitelist__):
-                return True
-            else:
-                return not self.__should_skip_for(cls)
-    
-    def __should_skip_for(self, cls):
-        if hasattr(cls, '__requires__'):
-            def test_suite(): return 'ok'
-            test_suite.__name__ = cls.__name__
-            for requirement in cls.__requires__:
-                check = getattr(requires, requirement)
-                if check(test_suite)() != 'ok':
-                    # The requirement will perform messaging.
-                    return True
-
-        if cls.__unsupported_on__:
-            spec = testing.db_spec(*cls.__unsupported_on__)
-            if spec(testing.db):
-                print "'%s' unsupported on DB implementation '%s'" % (
-                     cls.__class__.__name__, testing.db.name)
-                return True
-                
-        if getattr(cls, '__only_on__', None):
-            spec = testing.db_spec(*util.to_list(cls.__only_on__))
-            if not spec(testing.db):
-                print "'%s' unsupported on DB implementation '%s'" % (
-                     cls.__class__.__name__, testing.db.name)
-                return True                    
-
-        if getattr(cls, '__skip_if__', False):
-            for c in getattr(cls, '__skip_if__'):
-                if c():
-                    print "'%s' skipped by %s" % (
-                        cls.__class__.__name__, c.__name__)
-                    return True
-                    
-        for rule in getattr(cls, '__excluded_on__', ()):
-            if testing._is_excluded(*rule):
-                print "'%s' unsupported on DB %s version %s" % (
-                    cls.__class__.__name__, testing.db.name,
-                    _server_version())
-                return True
-        return False
-
-    def beforeTest(self, test):
-        testing.resetwarnings()
-
-    def afterTest(self, test):
-        testing.resetwarnings()
-        
-    def afterContext(self):
-        testing.global_cleanup_assertions()
-        
-    #def handleError(self, test, err):
-        #pass
-
-    #def finalize(self, result=None):
-        #pass

File setup.py

View file
  • Ignore whitespace
 
 def find_packages(dir_):
     packages = []
-    for pkg in ['sqlalchemy', 'sqlalchemy_nose']:
+    for pkg in ['sqlalchemy']:
         for _dir, subdirectories, files in os.walk(os.path.join(dir_, pkg)):
             if '__init__.py' in files:
                 lib, fragment = _dir.split(os.sep, 1)
 
       tests_require = ['nose >= 0.11'],
       test_suite = "nose.collector",
-      entry_points = {
-          'nose.plugins.0.10': [
-              'sqlalchemy = sqlalchemy_nose.noseplugin:NoseSQLAlchemy',
-              ]
-          },
       
       long_description = """\
 SQLAlchemy is:

File sqla_nose.py

View file
  • Ignore whitespace
 """
 import sys
 
-try:
-    from sqlalchemy_nose.noseplugin import NoseSQLAlchemy
-except ImportError:
-    from os import path
-    sys.path.append(path.join(path.dirname(path.abspath(__file__)), 'lib'))
-    from sqlalchemy_nose.noseplugin import NoseSQLAlchemy
+from os import path
+for pth in ['.', './lib']:
+    sys.path.insert(0, path.join(path.dirname(path.abspath(__file__)), pth))
+
+from test.bootstrap.noseplugin import NoseSQLAlchemy
 
 import nose
 

File test/bootstrap/__init__.py

  • Ignore whitespace
Empty file added.

File test/bootstrap/config.py

View file
  • Ignore whitespace
+import optparse, os, sys, re, ConfigParser, time, warnings
+
+# 2to3
+import StringIO
+
+logging = None
+
+__all__ = 'parser', 'configure', 'options',
+
+db = None
+db_label, db_url, db_opts = None, None, {}
+
+options = None
+file_config = None
+
+base_config = """
+[db]
+sqlite=sqlite:///:memory:
+sqlite_file=sqlite:///querytest.db
+postgresql=postgresql://scott:tiger@127.0.0.1:5432/test
+postgres=postgresql://scott:tiger@127.0.0.1:5432/test
+pg8000=postgresql+pg8000://scott:tiger@127.0.0.1:5432/test
+postgresql_jython=postgresql+zxjdbc://scott:tiger@127.0.0.1:5432/test
+mysql_jython=mysql+zxjdbc://scott:tiger@127.0.0.1:5432/test
+mysql=mysql://scott:tiger@127.0.0.1:3306/test
+oracle=oracle://scott:tiger@127.0.0.1:1521
+oracle8=oracle://scott:tiger@127.0.0.1:1521/?use_ansi=0
+mssql=mssql://scott:tiger@SQUAWK\\SQLEXPRESS/test
+firebird=firebird://sysdba:masterkey@localhost//tmp/test.fdb
+maxdb=maxdb://MONA:RED@/maxdb1
+"""
+
+def _log(option, opt_str, value, parser):
+    global logging
+    if not logging:
+        import logging
+        logging.basicConfig()
+
+    if opt_str.endswith('-info'):
+        logging.getLogger(value).setLevel(logging.INFO)
+    elif opt_str.endswith('-debug'):
+        logging.getLogger(value).setLevel(logging.DEBUG)
+
+
+def _list_dbs(*args):
+    print "Available --db options (use --dburi to override)"
+    for macro in sorted(file_config.options('db')):
+        print "%20s\t%s" % (macro, file_config.get('db', macro))
+    sys.exit(0)
+
+def _server_side_cursors(options, opt_str, value, parser):
+    db_opts['server_side_cursors'] = True
+
+def _engine_strategy(options, opt_str, value, parser):
+    if value:
+        db_opts['strategy'] = value
+
+class _ordered_map(object):
+    def __init__(self):
+        self._keys = list()
+        self._data = dict()
+
+    def __setitem__(self, key, value):
+        if key not in self._keys:
+            self._keys.append(key)
+        self._data[key] = value
+
+    def __iter__(self):
+        for key in self._keys:
+            yield self._data[key]
+
+# at one point in refactoring, modules were injecting into the config
+# process.  this could probably just become a list now.
+post_configure = _ordered_map()
+
+def _engine_uri(options, file_config):
+    global db_label, db_url
+    db_label = 'sqlite'
+    if options.dburi:
+        db_url = options.dburi
+        db_label = db_url[:db_url.index(':')]
+    elif options.db:
+        db_label = options.db
+        db_url = None
+
+    if db_url is None:
+        if db_label not in file_config.options('db'):
+            raise RuntimeError(
+                "Unknown engine.  Specify --dbs for known engines.")
+        db_url = file_config.get('db', db_label)
+post_configure['engine_uri'] = _engine_uri
+
+def _require(options, file_config):
+    if not(options.require or
+           (file_config.has_section('require') and
+            file_config.items('require'))):
+        return
+
+    try:
+        import pkg_resources
+    except ImportError:
+        raise RuntimeError("setuptools is required for version requirements")
+
+    cmdline = []
+    for requirement in options.require:
+        pkg_resources.require(requirement)
+        cmdline.append(re.split('\s*(<!>=)', requirement, 1)[0])
+
+    if file_config.has_section('require'):
+        for label, requirement in file_config.items('require'):
+            if not label == db_label or label.startswith('%s.' % db_label):
+                continue
+            seen = [c for c in cmdline if requirement.startswith(c)]
+            if seen:
+                continue
+            pkg_resources.require(requirement)
+post_configure['require'] = _require
+
+def _engine_pool(options, file_config):
+    if options.mockpool:
+        from sqlalchemy import pool
+        db_opts['poolclass'] = pool.AssertionPool
+post_configure['engine_pool'] = _engine_pool
+
+def _create_testing_engine(options, file_config):
+    from sqlalchemy.test import engines, testing
+    global db
+    db = engines.testing_engine(db_url, db_opts)
+    testing.db = db
+post_configure['create_engine'] = _create_testing_engine
+
+def _prep_testing_database(options, file_config):
+    from sqlalchemy.test import engines
+    from sqlalchemy import schema
+
+    # also create alt schemas etc. here?
+    if options.dropfirst:
+        e = engines.utf8_engine()
+        existing = e.table_names()
+        if existing:
+            print "Dropping existing tables in database: " + db_url
+            try:
+                print "Tables: %s" % ', '.join(existing)
+            except:
+                pass
+            print "Abort within 5 seconds..."
+            time.sleep(5)
+            md = schema.MetaData(e, reflect=True)
+            md.drop_all()
+        e.dispose()
+
+post_configure['prep_db'] = _prep_testing_database
+
+def _set_table_options(options, file_config):
+    from sqlalchemy.test import schema
+
+    table_options = schema.table_options
+    for spec in options.tableopts:
+        key, value = spec.split('=')
+        table_options[key] = value
+
+    if options.mysql_engine:
+        table_options['mysql_engine'] = options.mysql_engine
+post_configure['table_options'] = _set_table_options
+
+def _reverse_topological(options, file_config):
+    if options.reversetop:
+        from sqlalchemy.orm import unitofwork, session, mapper, dependency
+        from sqlalchemy import topological
+        from sqlalchemy.test.util import RandomSet
+        topological.set = unitofwork.set = session.set = mapper.set = dependency.set = RandomSet
+post_configure['topological'] = _reverse_topological
+

File test/bootstrap/noseplugin.py

View file
  • Ignore whitespace
+import logging
+import os
+import re
+import sys
+import time
+import warnings
+import ConfigParser
+import StringIO
+
+import nose.case
+from nose.plugins import Plugin
+
+from test.bootstrap import config
+
+from test.bootstrap.config import (
+    _create_testing_engine, _engine_pool, _engine_strategy, _engine_uri, _list_dbs, _log,
+    _prep_testing_database, _require, _reverse_topological, _server_side_cursors,
+    _set_table_options, base_config, db, db_label, db_url, file_config, post_configure)
+
+log = logging.getLogger('nose.plugins.sqlalchemy')
+
+class NoseSQLAlchemy(Plugin):
+    """
+    Handles the setup and extra properties required for testing SQLAlchemy
+    """
+    enabled = True
+    name = 'sqlalchemy'
+    score = 100
+
+    def options(self, parser, env=os.environ):
+        Plugin.options(self, parser, env)
+        opt = parser.add_option
+        opt("--log-info", action="callback", type="string", callback=_log,
+            help="turn on info logging for <LOG> (multiple OK)")
+        opt("--log-debug", action="callback", type="string", callback=_log,
+            help="turn on debug logging for <LOG> (multiple OK)")
+        opt("--require", action="append", dest="require", default=[],
+            help="require a particular driver or module version (multiple OK)")
+        opt("--db", action="store", dest="db", default="sqlite",
+            help="Use prefab database uri")
+        opt('--dbs', action='callback', callback=_list_dbs,
+            help="List available prefab dbs")
+        opt("--dburi", action="store", dest="dburi",
+            help="Database uri (overrides --db)")
+        opt("--dropfirst", action="store_true", dest="dropfirst",
+            help="Drop all tables in the target database first (use with caution on Oracle, "
+            "MS-SQL)")
+        opt("--mockpool", action="store_true", dest="mockpool",
+            help="Use mock pool (asserts only one connection used)")
+        opt("--enginestrategy", action="callback", type="string",
+            callback=_engine_strategy,
+            help="Engine strategy (plain or threadlocal, defaults to plain)")
+        opt("--reversetop", action="store_true", dest="reversetop", default=False,
+            help="Use a random-ordering set implementation in the ORM (helps "
+                  "reveal dependency issues)")
+        opt("--unhashable", action="store_true", dest="unhashable", default=False,
+            help="Disallow SQLAlchemy from performing a hash() on mapped test objects.")
+        opt("--noncomparable", action="store_true", dest="noncomparable", default=False,
+            help="Disallow SQLAlchemy from performing == on mapped test objects.")
+        opt("--truthless", action="store_true", dest="truthless", default=False,
+            help="Disallow SQLAlchemy from truth-evaluating mapped test objects.")
+        opt("--serverside", action="callback", callback=_server_side_cursors,
+            help="Turn on server side cursors for PG")
+        opt("--mysql-engine", action="store", dest="mysql_engine", default=None,
+            help="Use the specified MySQL storage engine for all tables, default is "
+                 "a db-default/InnoDB combo.")
+        opt("--table-option", action="append", dest="tableopts", default=[],
+            help="Add a dialect-specific table option, key=value")
+
+        global file_config
+        file_config = ConfigParser.ConfigParser()
+        file_config.readfp(StringIO.StringIO(base_config))
+        file_config.read(['test.cfg', os.path.expanduser('~/.satest.cfg')])
+        config.file_config = file_config
+        
+    def configure(self, options, conf):
+        Plugin.configure(self, options, conf)
+        self.options = options
+        
+    def begin(self):
+        global testing, requires, util
+        from sqlalchemy.test import testing, requires
+        from sqlalchemy import util
+        
+        testing.db = db
+        testing.requires = requires
+
+        # Lazy setup of other options (post coverage)
+        for fn in post_configure:
+            fn(self.options, file_config)
+        
+    def describeTest(self, test):
+        return ""
+        
+    def wantClass(self, cls):
+        """Return true if you want the main test selector to collect
+        tests from this class, false if you don't, and None if you don't
+        care.
+
+        :Parameters:
+           cls : class
+             The class being examined by the selector
+
+        """
+
+        if not issubclass(cls, testing.TestBase):
+            return False
+        else:
+            if (hasattr(cls, '__whitelist__') and testing.db.name in cls.__whitelist__):
+                return True
+            else:
+                return not self.__should_skip_for(cls)
+    
+    def __should_skip_for(self, cls):
+        if hasattr(cls, '__requires__'):
+            def test_suite(): return 'ok'
+            test_suite.__name__ = cls.__name__
+            for requirement in cls.__requires__:
+                check = getattr(requires, requirement)
+                if check(test_suite)() != 'ok':
+                    # The requirement will perform messaging.
+                    return True
+
+        if cls.__unsupported_on__:
+            spec = testing.db_spec(*cls.__unsupported_on__)
+            if spec(testing.db):
+                print "'%s' unsupported on DB implementation '%s'" % (
+                     cls.__class__.__name__, testing.db.name)
+                return True
+                
+        if getattr(cls, '__only_on__', None):
+            spec = testing.db_spec(*util.to_list(cls.__only_on__))
+            if not spec(testing.db):
+                print "'%s' unsupported on DB implementation '%s'" % (
+                     cls.__class__.__name__, testing.db.name)
+                return True                    
+
+        if getattr(cls, '__skip_if__', False):
+            for c in getattr(cls, '__skip_if__'):
+                if c():
+                    print "'%s' skipped by %s" % (
+                        cls.__class__.__name__, c.__name__)
+                    return True
+                    
+        for rule in getattr(cls, '__excluded_on__', ()):
+            if testing._is_excluded(*rule):
+                print "'%s' unsupported on DB %s version %s" % (
+                    cls.__class__.__name__, testing.db.name,
+                    _server_version())
+                return True
+        return False
+
+    def beforeTest(self, test):
+        testing.resetwarnings()
+
+    def afterTest(self, test):
+        testing.resetwarnings()
+        
+    def afterContext(self):
+        testing.global_cleanup_assertions()
+        
+    #def handleError(self, test, err):
+        #pass
+
+    #def finalize(self, result=None):
+        #pass