python类engine()的实例源码

def _update_config(self):
        _cp_config = {
            ## Server Opts ##
            'log.screen': False,
            'log.error_file': "",
            'log.access_file': "",
            'server.socket_host': self._cfg.listen,
            'server.socket_port': self._cfg.listen_port,
            'server.thread_pool': 10,
            'engine.autoreload.on': False,
            'request.show_tracebacks': False,
            'request.show_mismatched_params': False,
            'response.headers.server': APIServer.SERVER_TOKEN,
            ## Custom Tools Opts ##
            'tools.delete_allow_header.on': True,
            'tools.fix_http_content_length.on': True,
            'tools.unescape_response.on': True,
            ## Error Handling Opts ##
            'error_page.default': error_response,
        }
        cherrypy.config.update(_cp_config)
        cherrypy.log.error_log = LOG

def start_handler(self):
        """
        Runs at the end of the request processing by calling
        the opened method of the handler.
        """
        request = cherrypy.request
        if not hasattr(request, 'ws_handler'):
            return

        addr = (request.remote.ip, request.remote.port)
        ws_handler = request.ws_handler
        request.ws_handler = None
        delattr(request, 'ws_handler')

        # By doing this we detach the socket from
        # the CherryPy stack avoiding memory leaks
        detach_connection(request.rfile.rfile)

        cherrypy.engine.publish('handle-websocket', ws_handler, addr)

def start_handler(self):
        """
        Runs at the end of the request processing by calling
        the opened method of the handler.
        """
        request = cherrypy.request
        if not hasattr(request, 'ws_handler'):
            return

        addr = (request.remote.ip, request.remote.port)
        ws_handler = request.ws_handler
        request.ws_handler = None
        delattr(request, 'ws_handler')

        # By doing this we detach the socket from
        # the CherryPy stack avoiding memory leaks
        detach_connection(request.rfile.rfile)

        cherrypy.engine.publish('handle-websocket', ws_handler, addr)

def merge(base, other):
    """Merge one app config (from a dict, file, or filename) into another.

    If the given config is a filename, it will be appended to
    the list of files to monitor for "autoreload" changes.
    """
    if isinstance(other, text_or_bytes):
        cherrypy.engine.autoreload.files.add(other)

    # Load other into base
    for section, value_map in reprconf.as_dict(other).items():
        if not isinstance(value_map, dict):
            raise ValueError(
                'Application config must include section headers, but the '
                "config you tried to merge doesn't have any sections. "
                'Wrap your config in another dict with paths as section '
                "headers, for example: {'/': config}.")
        base.setdefault(section, {}).update(value_map)

def _engine_namespace_handler(k, v):
    """Config handler for the "engine" namespace."""
    engine = cherrypy.engine

    if k == 'SIGHUP':
        engine.subscribe('SIGHUP', v)
    elif k == 'SIGTERM':
        engine.subscribe('SIGTERM', v)
    elif '.' in k:
        plugin, attrname = k.split('.', 1)
        plugin = getattr(engine, plugin)
        if attrname == 'on':
            if v and hasattr(getattr(plugin, 'subscribe', None), '__call__'):
                plugin.subscribe()
                return
            elif (
                (not v) and
                hasattr(getattr(plugin, 'unsubscribe', None), '__call__')
            ):
                plugin.unsubscribe()
                return
        setattr(plugin, attrname, v)
    else:
        setattr(engine, k, v)

def check_site_config_entries_in_app_config(self):
        """Check for mounted Applications that have site-scoped config."""
        for sn, app in iteritems(cherrypy.tree.apps):
            if not isinstance(app, cherrypy.Application):
                continue

            msg = []
            for section, entries in iteritems(app.config):
                if section.startswith('/'):
                    for key, value in iteritems(entries):
                        for n in ('engine.', 'server.', 'tree.', 'checker.'):
                            if key.startswith(n):
                                msg.append('[%s] %s = %s' %
                                           (section, key, value))
            if msg:
                msg.insert(0,
                           'The application mounted at %r contains the '
                           'following config entries, which are only allowed '
                           'in site-wide config. Move them to a [global] '
                           'section and pass them to cherrypy.config.update() '
                           'instead of tree.mount().' % sn)
                warnings.warn(os.linesep.join(msg))

def _populate_known_types(self):
        b = [x for x in vars(builtins).values()
             if type(x) is type(str)]

        def traverse(obj, namespace):
            for name in dir(obj):
                # Hack for 3.2's warning about body_params
                if name == 'body_params':
                    continue
                vtype = type(getattr(obj, name, None))
                if vtype in b:
                    self.known_config_types[namespace + '.' + name] = vtype

        traverse(cherrypy.request, 'request')
        traverse(cherrypy.response, 'response')
        traverse(cherrypy.server, 'server')
        traverse(cherrypy.engine, 'engine')
        traverse(cherrypy.log, 'log')

def run(self):
        while self.running:
            message = s_queue.get()

            if 'timestamp' not in message:
                message['timestamp'] = datetime.datetime.now().isoformat()

            if message['type'] in HISTORY_TYPES:
                cherrypy.engine.publish('add-history', message)
            elif message['type'] == 'command':
                cherrypy.engine.publish('process-command', message['command'], message)

            if message['type'] == 'system_message' and not self.settings['chat']['keys'].get('show_system_msg', True):
                continue

            log.debug("%s", message)
            self.send_message(message, 'chat')
            self.send_message(message, 'gui')
        log.info("Messaging thread stopping")

def __init__(self, host, port, root_folder, **kwargs):
        super(self.__class__, self).__init__()
        self.daemon = True
        self.host = host
        self.port = port
        self.root_folder = root_folder
        self.style_settings = kwargs['style_settings']
        self.modules = kwargs.pop('modules')

        self.root_config = None
        self.css_config = None
        self.gui_root_config = None
        self.gui_css_config = None

        self.rest_config = None

        cherrypy.config.update({'server.socket_port': int(self.port), 'server.socket_host': self.host,
                                'engine.autoreload.on': False})
        self.websocket = WebChatPlugin(cherrypy.engine, self.style_settings)
        self.websocket.subscribe()
        cherrypy.tools.websocket = WebSocketTool()

def update_install():
        ver = version.Version()

        if not core.UPDATE_STATUS or core.UPDATE_STATUS['status'] != u'behind':
            return

        logging.info(u'Running automatic updater.')

        logging.info(u'Currently {} commits behind. Updating to {}.'.format(
                     core.UPDATE_STATUS['behind_count'], core.UPDATE_STATUS['new_hash']))

        core.UPDATING = True

        logging.info(u'Executing update.')
        update = ver.manager.execute_update()
        core.UPDATING = False

        if not update:
            logging.error(u'Update failed.')

        logging.info(u'Update successful, restarting.')
        cherrypy.engine.restart()
        return

def merge(base, other):
    """Merge one app config (from a dict, file, or filename) into another.

    If the given config is a filename, it will be appended to
    the list of files to monitor for "autoreload" changes.
    """
    if isinstance(other, text_or_bytes):
        cherrypy.engine.autoreload.files.add(other)

    # Load other into base
    for section, value_map in reprconf.as_dict(other).items():
        if not isinstance(value_map, dict):
            raise ValueError(
                "Application config must include section headers, but the "
                "config you tried to merge doesn't have any sections. "
                "Wrap your config in another dict with paths as section "
                "headers, for example: {'/': config}.")
        base.setdefault(section, {}).update(value_map)

def _engine_namespace_handler(k, v):
    """Config handler for the "engine" namespace."""
    engine = cherrypy.engine

    if k == 'SIGHUP':
        engine.subscribe('SIGHUP', v)
    elif k == 'SIGTERM':
        engine.subscribe('SIGTERM', v)
    elif "." in k:
        plugin, attrname = k.split(".", 1)
        plugin = getattr(engine, plugin)
        if attrname == 'on':
            if v and hasattr(getattr(plugin, 'subscribe', None), '__call__'):
                plugin.subscribe()
                return
            elif (
                (not v) and
                hasattr(getattr(plugin, 'unsubscribe', None), '__call__')
            ):
                plugin.unsubscribe()
                return
        setattr(plugin, attrname, v)
    else:
        setattr(engine, k, v)

def test_5_Start_Error(self):
        # If test_3 has not been executed, the server won't be stopped,
        # so we'll have to do it.
        if engine.state != engine.states.EXITING:
            engine.exit()

        # If a process errors during start, it should stop the engine
        # and exit with a non-zero exit code.
        p = helper.CPProcess(ssl=(self.scheme.lower() == 'https'),
                             wait=True)
        p.write_conf(
            extra="""starterror: True
test_case_name: "test_5_Start_Error"
"""
        )
        p.start(imports='cherrypy.test._test_states_demo')
        if p.exit_code == 0:
            self.fail("Process failed to return nonzero exit code.")

def check_site_config_entries_in_app_config(self):
        """Check for mounted Applications that have site-scoped config."""
        for sn, app in iteritems(cherrypy.tree.apps):
            if not isinstance(app, cherrypy.Application):
                continue

            msg = []
            for section, entries in iteritems(app.config):
                if section.startswith('/'):
                    for key, value in iteritems(entries):
                        for n in ("engine.", "server.", "tree.", "checker."):
                            if key.startswith(n):
                                msg.append("[%s] %s = %s" %
                                           (section, key, value))
            if msg:
                msg.insert(0,
                           "The application mounted at %r contains the "
                           "following config entries, which are only allowed "
                           "in site-wide config. Move them to a [global] "
                           "section and pass them to cherrypy.config.update() "
                           "instead of tree.mount()." % sn)
                warnings.warn(os.linesep.join(msg))

def _populate_known_types(self):
        b = [x for x in vars(builtins).values()
             if type(x) is type(str)]

        def traverse(obj, namespace):
            for name in dir(obj):
                # Hack for 3.2's warning about body_params
                if name == 'body_params':
                    continue
                vtype = type(getattr(obj, name, None))
                if vtype in b:
                    self.known_config_types[namespace + "." + name] = vtype

        traverse(cherrypy.request, "request")
        traverse(cherrypy.response, "response")
        traverse(cherrypy.server, "server")
        traverse(cherrypy.engine, "engine")
        traverse(cherrypy.log, "log")

def run(self, daemon=False):
        if daemon:
            daemon = cherrypy.process.plugins.Daemonizer(cherrypy.engine)
            daemon.subscribe()
        if self._cfg.pid_file:
            pid = cherrypy.process.plugins.PIDFile(cherrypy.engine, self._cfg.pid_file)
            pid.subscribe()
        cherrypy.engine.start()
        cherrypy.engine.block()

def __init__(self, bus: Bus, settings_module: str = 'settings',
                 wsgi_http_logger: type = HTTPLogger,
                 local_settings: Settings = None) -> None:
        """ CherryPy engine plugin to configure and mount
        the Django application onto the CherryPy server.
        """
        plugins.SimplePlugin.__init__(self, bus)
        self.settings_module = settings_module
        self.wsgi_http_logger = wsgi_http_logger
        self.crawler_settings = local_settings

def start(self) -> None:
        """ When the bus starts, the plugin is also started
        and we load the Django application. We then mount it on
        the CherryPy engine for serving as a WSGI application.
        We let CherryPy serve the application's static files.
        """
        from frontend import settings
        settings.crawler_settings = self.crawler_settings

        os.environ['DJANGO_SETTINGS_MODULE'] = self.settings_module

        from django.core.wsgi import get_wsgi_application

        def corsstaticdir(
                section: str, dir: str, root: str = '', match: str = '',
                content_types: None = None, index: str = '', debug: bool = False
        ) -> bool:
            cherrypy.response.headers['Access-Control-Allow-Origin'] = '*'
            return static.staticdir(section, dir, root, match, content_types, index, debug)

        cherrypy.tree.graft(self.wsgi_http_logger(get_wsgi_application(), self.crawler_settings))

        settings.WORKERS.start_workers(settings.CRAWLER_SETTINGS)

        tool = cherrypy._cptools.HandlerTool(corsstaticdir)
        static_handler = tool.handler(
            section="/",
            dir=os.path.split(settings.STATIC_ROOT)[1],
            root=os.path.abspath(os.path.split(settings.STATIC_ROOT)[0])
        )
        cherrypy.tree.mount(static_handler, settings.STATIC_URL)

        media_handler = tool.handler(
            section="/",
            dir=os.path.split(settings.MEDIA_ROOT)[1],
            root=os.path.abspath(os.path.split(settings.MEDIA_ROOT)[0])
        )
        cherrypy.tree.mount(media_handler, settings.MEDIA_URL)

def reset(self, token=None, username=None):
        if cherrypy.engine.publish('token-verify', token, username).pop():
            cherrypy.session['token'] = token
            cherrypy.session['username'] = username
            return {'ok': True}
        else:
            return {'ok': False}

def change(self, **params):
        engine = cherrypy.engine
        if cherrypy.session.get('auth', False):
            user = cherrypy.session['user']
            oldpasswd = cherrypy.request.params.get('oldpassword')
            newpasswd = cherrypy.request.params.get('newpassword')

            try:
                user.change_password(oldpasswd, newpasswd)

                return {'ok': True}
            except InvalidCredentials:
                return {
                    'ok': False,
                    'error': 'Current password invalid.'
                }
            except UserModelException:
                return {
                    'ok': False,
                    'error': 'Unknown system error.  Contact your Systems Administrator.'
                }

        elif cherrypy.session.get('token', False):
            cherrypy.session['user'] = User(cherrypy.session['username'])
            newpassword = cherrypy.request.params.get('newpassword')

            try:
                cherrypy.session['user'].set_password(newpassword)
                return {'ok': True}
            except UserModelException:
                return {
                    'ok': False,
                    'error': 'Unable to change your password. Try again later.'
                }

def setUpModule():
    cherrypy.tree.mount(Root(), '/')
    cherrypy.engine.start()

def tearDownModule():
    cherrypy.engine.exit()

def setUpClass(cls):
        cls.mockldap = MockLdap(LDAPTestDirectory.directory)
        cherrypy.engine.stop()
        cherrypy.engine.ldap = LDAPEnginePlugin(
            bus=cherrypy.engine,
            uri="ldap://localhost/",
            base_dn="ou=example,o=test",
            bind_dn="cn=admin,ou=example,o=test",
            bind_pw="ldaptest",
            tls=True,
            no_verify=True)
        cherrypy.engine.ldap.subscribe()
        cherrypy.engine.start()

def tearDownClass(cls):
        cherrypy.engine.ldap.unsubscribe()
        del cherrypy.engine.ldap
        del cls.mockldap

def test_bad_base_dn(self):
        with self.assertRaises(cherrypy.process.wspbus.ChannelFailures):
            cherrypy.engine.publish('ldap-auth', 'alice,ou==bad,,', 'alicepw')

def test_auth(self):
        auth = cherrypy.engine.publish('ldap-auth', 'alice', 'alicepw').pop()
        self.assertEqual(auth, True)
        with self.assertRaises(cherrypy.process.wspbus.ChannelFailures):
            cherrypy.engine.publish('ldap-auth', 'alice', 'bobpw')

def test_get_user_by_uid(self):
        alice = cherrypy.engine.publish('ldap-user-by-uid', 'alice').pop()
        self.assertEqual(alice['uid'], 'alice')
        nouser = cherrypy.engine.publish('ldap-user-by-uid', 'nouser').pop()
        self.assertEqual(nouser, {})

def test_set_password(self):
        username = 'bob'
        cherrypy.engine.publish('ldap-set-password', username, self.hashes)
        bob = cherrypy.engine.publish('ldap-user-by-uid', 'bob').pop()
        self.assertEqual(self.hashes['userPassword'] in bob['userPassword'], True)
        self.assertEqual(self.hashes['sambaNTPassword'] in bob['sambaNTPassword'], True)
        self.assertEqual(cherrypy.engine.publish('ldap-auth', 'bob', 'changeme'), [True])

def test_change_password(self):
        cherrypy.engine.publish('ldap-change-password', 'bob', 'bobpw', self.hashes)
        bob = cherrypy.engine.publish('ldap-user-by-uid', 'bob').pop()
        self.assertEqual(self.hashes['userPassword'] in bob['userPassword'], True)
        self.assertEqual(self.hashes['sambaNTPassword'] in bob['sambaNTPassword'], True)
        self.assertEqual(cherrypy.engine.publish('ldap-auth', 'bob', 'changeme'), [True])

def test_change_passwd_bad_creds(self):
        with self.assertRaises(cherrypy.process.wspbus.ChannelFailures):
            cherrypy.engine.publish('ldap-change-password', 'bob', 'alicepw', self.hashes)