def begin(self):
        """Called before any tests are collected or run

        Loads the application, and in turn its configuration.

        """
        global pylonsapp
        path = os.getcwd()
        sys.path.insert(0, path)
        pkg_resources.working_set.add_entry(path)
        self.app = pylonsapp = loadapp('config:' + self.config_file,
                                       relative_to=path)

        # Setup the config and app_globals, only works if we can get
        # to the config object
        conf = getattr(pylonsapp, 'config')
        if conf:
            pylons.config._push_object(conf)

            if 'pylons.app_globals' in conf:
                pylons.app_globals._push_object(conf['pylons.app_globals'])

        # Initialize a translator for tests that utilize i18n
        translator = _get_translator(pylons.config.get('lang'))
        pylons.translator._push_object(translator)

 def setup_app_env(self, environ, start_response):
     """Setup and register all the Pylons objects with the registry
     
     After creating all the global objects for use in the request,
     :meth:`~PylonsApp.register_globals` is called to register them
     in the environment.
     
     """
     if self.log_debug:
         log.debug("Setting up Pylons stacked object globals")
     
     
     # Setup the basic pylons global objects
     req_options = self.request_options
     req = Request(environ, charset=req_options['charset'],
                   unicode_errors=req_options['errors'],
                   decode_param_names=req_options['decode_param_names'])
     req.language = req_options['language']
     req.config = self.config
     req.link, req.route_dict = environ['wsgiorg.routing_args']
     
     response = Response(
         content_type=self.response_options['content_type'],
         charset=self.response_options['charset'])
     response.headers.update(self.response_options['headers'])
     
     # Store a copy of the request/response in environ for faster access
     pylons_obj = PylonsContext()
     pylons_obj.config = self.config
     pylons_obj.request = req
     pylons_obj.response = response
     pylons_obj.app_globals = self.globals
     pylons_obj.h = self.helpers
     
     if 'routes.url' in environ:
         pylons_obj.url = environ['routes.url']
             
     environ['pylons.pylons'] = pylons_obj
     
     environ['pylons.environ_config'] = self.environ_config
     
     # Setup the translator object
     lang = self.config['lang']
     pylons_obj.translator = _get_translator(lang, pylons_config=self.config)
     
     if self.config['pylons.strict_tmpl_context']:
         tmpl_context = ContextObj()
     else:
         tmpl_context = AttribSafeContextObj()
     pylons_obj.tmpl_context = req.tmpl_context = tmpl_context
     
     if self._session_key in environ:
         pylons_obj.session = req.session = environ[self._session_key]
     if self._cache_key in environ:
         pylons_obj.cache = environ[self._cache_key]
     
     # Load the globals with the registry if around
     if 'paste.registry' in environ:
         self.register_globals(environ)

def setup_py_trans():
    global lang_setup
    import pylons
    from pylons.i18n.translation import _get_translator
    root = os.path.join(test_root, 'sample_controllers')
    lang_setup = {'pylons.paths': {'root': root}, 'pylons.package': 'sample_controllers'}
    sys.path.append(test_root)
    pylons.translator._push_object(_get_translator(None, pylons_config=lang_setup))

def init_stack(config=None):
    if not config:
        config = pylons.test.pylonsapp.config
    url._push_object(URLGenerator(config['routes.map'], environ))
    pylons.app_globals._push_object(config['pylons.app_globals'])
    pylons.config._push_object(config)
    pylons.tmpl_context._push_object(ContextObj())
    # Initialize a translator for tests that utilize i18n
    translator = _get_translator(pylons.config.get('lang'))
    pylons.translator._push_object(translator)

 def __init__(self, *args, **kwargs):
     wsgiapp = loadapp('config:test.ini', relative_to=conf_dir)
     config = wsgiapp.config
     pylons.app_globals._push_object(config['pylons.app_globals'])
     pylons.config._push_object(config)
     
     # Initialize a translator for tests that utilize i18n
     translator = _get_translator(pylons.config.get('lang'))
     pylons.translator._push_object(translator)
     
     url._push_object(URLGenerator(config['routes.map'], environ))
     self.app = TestApp(wsgiapp)
     TestCase.__init__(self, *args, **kwargs)

    def begin(self):
        """Called before any tests are collected or run

        Loads the application, and in turn its configuration.

        """
        global pylonsapp
        path = os.getcwd()
        sys.path.insert(0, path)
        pkg_resources.working_set.add_entry(path)
        self.app = pylonsapp = loadapp('config:' + self.config_file,
                                       relative_to=path)

        # Initialize a translator for tests that utilize i18n
        translator = _get_translator(pylons.config.get('lang'))
        pylons.translator._push_object(translator)

def pytest_configure():
    path = os.getcwd()
    sys.path.insert(0, path)
    pkg_resources.working_set.add_entry(path)
    pylons.test.pylonsapp = loadapp('config:test.ini', relative_to=path)

    # Setup the config and app_globals, only works if we can get
    # to the config object
    conf = getattr(pylons.test.pylonsapp, 'config')
    if conf:
        pylons.config._push_object(conf)

        if 'pylons.app_globals' in conf:
            pylons.app_globals._push_object(conf['pylons.app_globals'])

    # Initialize a translator for tests that utilize i18n
    translator = _get_translator(pylons.config.get('lang'))
    pylons.translator._push_object(translator)

    return pylons.test.pylonsapp

def grok_app(conf, packages=None):
    if packages is None:
        packages = []

    items = [request, tmpl_context, response, session, url]
    translator._push_object(_get_translator(None))
    config._push_object(conf)

    for item in items:
        item._push_object(object())

    zope_config = zope.configuration.config.ConfigurationMachine()
    do_grok('grokcore.component', zope_config)
    for package in packages:
        do_grok(package, zope_config)
    zope_config.execute_actions()

    translator._pop_object()
    config._pop_object()
    for item in items:
        item._pop_object()

    def run_harvester(self, *args, **kwds):
        from pylons.i18n.translation import _get_translator
        import pylons

        pylons.translator._push_object(_get_translator(pylons.config.get("lang")))

        from ckan.model import HarvestingJob
        from ckan.controllers.harvesting import HarvestingJobController

        jobs = HarvestingJob.filter(status=u"New").all()
        jobs_len = len(jobs)
        jobs_count = 0
        if jobs_len:
            print "Running %s harvesting jobs..." % jobs_len
        else:
            print "There are no new harvesting jobs."
        print ""
        for job in jobs:
            jobs_count += 1
            print "Running job %s/%s: %s" % (jobs_count, jobs_len, job.id)
            self.print_harvesting_job(job)
            job_controller = HarvestingJobController(job)
            job_controller.harvest_documents()

def pytest_sessionstart():
    # setup resources before any test is executed
    pylonsapp = None
    pylons.test.pylonsapp = pylonsapp
    path = os.getcwd()
    sys.path.insert(0, path)
    pkg_resources.working_set.add_entry(path)
    config_file = py.test.config.inicfg.get("test_ini")
    pylonsapp = pylons.test.pylonsapp = loadapp('config:' + config_file,
                                                relative_to=path)

    # Setup the config and app_globals, only works if we can get
    # to the config object
    conf = getattr(pylonsapp, 'config')
    if conf:
        pylons.config._push_object(conf)

        if 'pylons.app_globals' in conf:
            pylons.app_globals._push_object(conf['pylons.app_globals'])

    # Initialize a translator for tests that utilize i18n
    translator = _get_translator(pylons.config.get('lang'))
    pylons.translator._push_object(translator)

    ):
        from Captcha import Base

from pylons import app_globals as g
from r2.config.middleware import RedditApp

# unfortunately, because of the deep intertwinded dependency we have in the
# orm with app_globals, we unfortunately have to do some pylons-setup
# at import time
baseplate.events.EventQueue = Queue.Queue
wsgiapp = loadapp('config:test.ini', relative_to=conf_dir)
pylons.app_globals._push_object(wsgiapp.config['pylons.app_globals'])
pylons.config._push_object(wsgiapp.config)

# Initialize a translator for tests that utilize i18n
translator = _get_translator(pylons.config.get('lang'))
pylons.translator._push_object(translator)

url._push_object(URLGenerator(pylons.config['routes.map'], {}))


def diff_dicts(d, expected, prefix=None):
    """Given 2 dicts, return a summary of their differences

    :param dict d: dict to match from ("got")
    :param dict expected: dict to match against ("want")
    :param prefix: key prefix (used for recursion)
    :type prefix: list or None
    :rtype: dict
    :returns: mapping of flattened keys to 2-ples of (got, want)
    """

def importdomains(orgid, filename, skipfirst):
    "Import domains"
    logger = importdomains.get_logger()
    results = dict(rows=[], global_error=[])
    keys = tuple(DOMAINFIELDS + DAFIELDS + DSFIELDS + ASFIELDS)
    translator = _get_translator(None)
    pylons.translator._push_object(translator)
    try:
        with open(filename, 'rU') as handle:
            dialect = csv.Sniffer().sniff(handle.read(1024))
            handle.seek(0)
            rows = csv.DictReader(handle, fieldnames=keys, dialect=dialect)
            query = Session.query(Group).filter(Group.id == orgid)
            org = query.one()
            logger.info("Importing domains from file: %s for: %s" %
                        (filename, org.name))
            try:
                count = 1
                for row in rows:
                    if skipfirst and (count == 1):
                        count += 1
                        continue
                    result = dict(id=None,
                                name=row['name'],
                                imported=False,
                                error=None)
                    try:
                        session_dict = {}
                        token = make_csrf(session_dict)
                        fields = getkeys(row)
                        post_data = dict2mdict(fields)
                        post_data.add('csrf_token', token)
                        form = AddDomainForm(post_data,
                                            csrf_context=session_dict)
                        form.organizations.query = query
                        if form.validate():
                            # insert to db
                            domain = Domain()
                            for field in form:
                                if field.name != 'csrf_token':
                                    setattr(domain, field.name, field.data)
                            domain.organizations.append(org)
                            Session.add(domain)
                            Session.commit()
                            result['id'] = domain.id
                            result['imported'] = True
                            logger.info("Imported domain: %s" % row['name'])
                            # process other data
                            process_aux(row, domain.id)
                        else:
                            logger.info("Import failed domain: %s" %
                                        row['name'])
                            if isinstance(form.errors, dict):
                                errors = []
                                for field in form.errors:
                                    themsg = u'%s: %s' % (field,
                                                unicode(form.errors[field][0]))
                                    errors.append(themsg)
                                result['error'] = u', '.join(errors)
                            else:
                                result['error'] = form.errors
                    except TypeError, err:
                        logger.info("Import failed domain: %s" % row['name'])
                        result['error'] = str(err)
                    except IntegrityError, err:
                        Session.rollback()
                        logger.info("Import failed domain: %s" % row['name'])
                        result['error'] = 'Domain already exists'
                    finally:
                        count += 1
                        results['rows'].append(result)

    with patch.object(sys, "path", [x for x in sys.path if not x.endswith("r2/r2/lib")]):
        from Captcha import Base

from pylons import app_globals as g
from r2.config.middleware import RedditApp

# unfortunately, because of the deep intertwinded dependency we have in the
# orm with app_globals, we unfortunately have to do some pylons-setup
# at import time
baseplate.events.EventQueue = Queue.Queue
wsgiapp = loadapp("config:test.ini", relative_to=conf_dir)
pylons.app_globals._push_object(wsgiapp.config["pylons.app_globals"])
pylons.config._push_object(wsgiapp.config)

# Initialize a translator for tests that utilize i18n
translator = _get_translator(pylons.config.get("lang"))
pylons.translator._push_object(translator)

url._push_object(URLGenerator(pylons.config["routes.map"], {}))


class MockAmqp(object):
    """An amqp replacement, suitable for unit tests.

    Besides providing a mock `queue` for storing all received events, this
    class provides a set of handy assert-style functions for checking what
    was previously queued.
    """

    def __init__(self, test_cls):
        self.queue = defaultdict(list)

def setup_app(command, conf, vars):
    load_environment(conf.global_conf, conf.local_conf)
    
    # http://pylonshq.com/project/pylonshq/ticket/509
    from pylons.i18n.translation import _get_translator
    import pylons
    pylons.translator._push_object(_get_translator(pylons.config.get('lang')))

    from wurdig.model import meta
    meta.metadata.bind = meta.engine
    
    users = UsersFromDatabase(model)
    
    filename = os.path.split(conf.filename)[-1]
    if filename == 'test.ini':
        # Permanently drop any existing tables
        meta.metadata.drop_all(checkfirst=True)
        
    # Create the tables if they aren't there already
    meta.metadata.create_all(checkfirst=True)
    
    users.role_create("admin")
    users.user_create("admin", password=md5("admin"))
    users.user_add_role("admin", role="admin")
    
    setting_site_title = model.Setting()
    setting_site_title.key = u'site_title'
    setting_site_title.value = u'My Wurdig Blog'
    setting_site_title.description = _(u'Site Title?')
    setting_site_title.help = _(u'What is the title of this '
                                'site (ex. Jason Leveille\'s Blog)?')
    meta.Session.add(setting_site_title)
    meta.Session.flush()
    
    setting_site_tagline = model.Setting()
    setting_site_tagline.key = u'site_tagline'
    setting_site_tagline.value = u'Just Another Wurdig Blog'
    setting_site_tagline.description = _(u'Site Tagline?')
    setting_site_tagline.help = _(u'What is the tagline of this site?')
    meta.Session.add(setting_site_tagline)
    meta.Session.flush()
    
    setting_display_tagline = model.Setting()
    setting_display_tagline.key = u'display_tagline'
    setting_display_tagline.value = u'true'
    setting_display_tagline.description = _(u'Display Site Tagline?')
    setting_display_tagline.type = u'b'
    setting_display_tagline.help = _(u'Should the tagline be shown for your site?')
    meta.Session.add(setting_display_tagline)
    meta.Session.flush()
    
    setting_admin_email = model.Setting()
    setting_admin_email.key = u'admin_email'
    setting_admin_email.value = u''
    setting_admin_email.description = _(u'Administrator Email')
    setting_admin_email.help = _(u'What is the administrator email for this site?  '
                                 'This will be used for notifications, to send email, etc?')
    meta.Session.add(setting_admin_email)
    meta.Session.flush()
    
    setting_display_admin_email = model.Setting()
    setting_display_admin_email.key = u'display_admin_email'
    setting_display_admin_email.value = u'false'
    setting_display_admin_email.description = _(u'Display Admin Email?')
    setting_display_admin_email.type = u'b'
    setting_display_admin_email.help = _(u'Should the administrator email be '
                                         'display (obfuscated) publicly?')
    meta.Session.add(setting_display_admin_email)
    meta.Session.flush()
    
    setting_spamword = model.Setting()
    setting_spamword.key = u'spamword'
    setting_spamword.value = u'wurdig'
    setting_spamword.description = _(u'Your Spam Deterrent Word?')
    setting_spamword.help = _(u'If you enable Akismet this spamword '
                              'becomes irrelevant.  Otherwise, what is '
                              'your spam deterrent word (used in comments)?')
    meta.Session.add(setting_spamword)
    meta.Session.flush()
    
    setting_enable_googlesearch = model.Setting()
    setting_enable_googlesearch.key = u'enable_googlesearch'
    setting_enable_googlesearch.value = u'false'
    setting_enable_googlesearch.description = _(u'Enable Google Search?')
    setting_enable_googlesearch.type = u'b'
    setting_enable_googlesearch.help = _(u'Should Google Search be enabled?  '
                                         'If so, you\'ll need a Google Search '
                                         'Key (http://www.google.com/sitesearch/).')
    meta.Session.add(setting_enable_googlesearch)
    meta.Session.flush()
    
    setting_googlesearch_key = model.Setting()
    setting_googlesearch_key.key = u'googlesearch_key'
    setting_googlesearch_key.value = u''
    setting_googlesearch_key.description = _(u'Your Google Search Key?')
    setting_googlesearch_key.help = _(u'What is your Google Search Key '
                                      '(http://www.google.com/sitesearch/)?  '
                                      'Only relevant if you have Google search enabled.')
    meta.Session.add(setting_googlesearch_key)
    meta.Session.flush()
#.........这里部分代码省略.........

def load_config(filename):
    conf = appconfig("config:" + os.path.abspath(filename) + "#content")
    config = load_environment(conf.global_conf, conf.local_conf)
    pylons.config.update(config)
    translator = _get_translator(pylons.config.get("lang"))
    pylons.translator._push_object(translator)

def importaccounts(domid, filename, skipfirst, userid):
    "import accounts"
    logger = importaccounts.get_logger()
    results = dict(rows=[], global_error=[])
    keys = tuple(ACCOUNTFIELDS + ADDRESSFIELDS)
    translator = _get_translator(None)
    pylons.translator._push_object(translator)
    try:
        with open(filename, 'rU') as handle:
            dialect = csv.Sniffer().sniff(handle.read(1024))
            handle.seek(0)
            rows = csv.DictReader(handle, fieldnames=keys, dialect=dialect)
            query = Session.query(Domain).filter(Domain.id == domid)
            domain = query.one()
            requester = Session.query(User).get(userid)
            logger.info("Importing accounts from file: %s for: %s" %
                        (filename, domain.name))
            try:
                count = 1
                for row in rows:
                    if skipfirst and (count == 1):
                        count += 1
                        continue
                    result = dict(id=None,
                                username=row['username'],
                                imported=False,
                                error=None)
                    try:
                        session_dict = {}
                        dummy = AddUserForm(dict2mdict({}),
                                csrf_context=session_dict)
                        fields = getkeys(row)
                        post_data = dict2mdict(fields)
                        post_data.add('password2', row['password1'])
                        post_data.add('domains', domid)
                        post_data.add('csrf_token',
                                    dummy.csrf_token.current_token)
                        form = AddUserForm(post_data,
                                csrf_context=session_dict)
                        form.domains.query = query
                        if form.validate():
                            #db insert
                            if domain.name != form.email.data.split('@')[1]:
                                raise TypeError(
                                    'Cannot import: %s into domain: %s' %
                                    (form.email.data, domain.name)
                                    )
                            user = User(form.username.data, form.email.data)
                            for attr in ['firstname', 'lastname', 'email',
                                'active', 'account_type', 'send_report',
                                'spam_checks', 'low_score', 'high_score',
                                'timezone']:
                                setattr(user, attr, getattr(form, attr).data)
                            user.local = True
                            user.set_password(form.password1.data)
                            if user.is_peleb:
                                user.domains = [domain]
                            Session.add(user)
                            Session.commit()
                            result['id'] = user.id
                            result['imported'] = True
                            logger.info("Imported account: %s" %
                                        row['username'])
                            #address add
                            add_address(row, user, requester)
                        else:
                            logger.info("Import failed account: %s" %
                                        row['username'])
                            if isinstance(form.errors, dict):
                                errors = []
                                for field in form.errors:
                                    themsg = u'%s: %s' % (field,
                                            unicode(form.errors[field][0]))
                                    errors.append(themsg)
                                result['error'] = u', '.join(errors)
                            else:
                                result['error'] = form.errors
                    except TypeError, err:
                        logger.info("Import failed account: %s" %
                                    row['username'])
                        result['error'] = str(err)
                    except IntegrityError, err:
                        Session.rollback()
                        logger.info("Import failed account: %s" %
                                    row['username'])
                        result['error'] = 'Account already exists'
                    finally:
                        count += 1
                        results['rows'].append(result)

def load_config(filename):
    conf = appconfig('config:' + os.path.abspath(filename) + '#content')
    load_environment(conf.global_conf, conf.local_conf)
    translator = _get_translator(pylons.config.get('lang'))
    pylons.translator._push_object(translator)

 def testSetUp(self):
     config = pylons.test.pylonsapp.config
     translator = _get_translator(config.get('lang'), pylons_config=config)
     pylons.translator._push_object(translator)
     url._push_object(URLGenerator(pylons.test.pylonsapp.config['routes.map'], environ))