noseplugin.py :  » Database » SQLAlchemy » SQLAlchemy-0.6.0 » lib » sqlalchemy » test » Python Open Source

Home
Python Open Source
1.3.1.2 Python
2.Ajax
3.Aspect Oriented
4.Blog
5.Build
6.Business Application
7.Chart Report
8.Content Management Systems
9.Cryptographic
10.Database
11.Development
12.Editor
13.Email
14.ERP
15.Game 2D 3D
16.GIS
17.GUI
18.IDE
19.Installer
20.IRC
21.Issue Tracker
22.Language Interface
23.Log
24.Math
25.Media Sound Audio
26.Mobile
27.Network
28.Parser
29.PDF
30.Project Management
31.RSS
32.Search
33.Security
34.Template Engines
35.Test
36.UML
37.USB Serial
38.Web Frameworks
39.Web Server
40.Web Services
41.Web Unit
42.Wiki
43.Windows
44.XML
Python Open Source » Database » SQLAlchemy 
SQLAlchemy » SQLAlchemy 0.6.0 » lib » sqlalchemy » test » noseplugin.py
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 import util,log
from sqlalchemy.test import testing,config,requires
from sqlalchemy.test.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):
        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
www.java2java.com | Contact Us
Copyright 2009 - 12 Demo Source and Support. All rights reserved.
All other trademarks are property of their respective owners.