python类engine()的实例源码

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

def test_add_sshpubkey(self):
        # key validation happens at the model level; using a short key for brevity
        sshpubkey = 'ssh-rsa AAAAB3NzaC1yc2EAAAABIwAAAo1d01QraTlMVSsbxNrRFi9wrf+M7Q== fakekey'
        cherrypy.engine.publish('ldap-add-key', 'bob', sshpubkey)
        bob = cherrypy.engine.publish('ldap-user-by-uid', 'bob').pop()
        self.assertEqual(sshpubkey in bob['sshPublicKey'], True)

def setUpClass(cls):
        cherrypy.engine.stop()
        cherrypy.engine.passwd = PasswordEnginePlugin(cherrypy.engine)
        cherrypy.engine.passwd.subscribe()
        cherrypy.engine.start()

def tearDownClass(cls):
        cherrypy.engine.passwd.unsubscribe()
        del cherrypy.engine.passwd

def test_plugin(self, mock_random):
        print cherrypy.engine.publish
        hashes = cherrypy.engine.publish('password-hash', 'changeme').pop()
        self.assertIsInstance(hashes, dict)
        self.assertEqual(hashes['userPassword'], '{SSHA}6QBuyak4WbsUzcqUKx0yB74RFUFvDbys')
        self.assertEqual(hashes['sambaNTPassword'], '6597D9FE8469E21D840E2CBFF8D43C8B')

def setUpClass(cls):
        cherrypy.engine.stop()
        cherrypy.engine.token = TokenEnginePlugin(cherrypy.engine, secret="testsecret")
        cherrypy.engine.token.subscribe()
        cherrypy.engine.start()

def tearDownClass(cls):
        cherrypy.engine.token.unsubscribe()
        del cherrypy.engine.token

def test_valid_token(self, mock_epoch):
        valid = cherrypy.engine.publish('token-verify', self.valid_token, self.data).pop()
        self.assertEqual(valid, True)

def test_invalid_token(self, mock_epoch):
        invalid = cherrypy.engine.publish('token-verify', self.invalid_token, self.data).pop()
        self.assertEqual(invalid, False)

def test_invalid_data(self, mock_epoch):
        invalid = cherrypy.engine.publish('token-verify', self.valid_token, 'baddata').pop()
        self.assertEqual(invalid, False)

def test_expired_token(self, mock_epoch=15999.0):
        invalid = cherrypy.engine.publish('token-verify', self.valid_token, self.data).pop()
        self.assertEqual(invalid, False)

def run(self):
        engine = cherrypy.engine

        if hasattr(engine, "signal_handler"):
            engine.signal_handler.subscribe()

        if hasattr(engine, "console_control_handler"):
            engine.console_control_handler.subscribe()

        engine.start()
        engine.block()

def cleanup(self):
        """
        Terminate all connections and clear the pool. Executed when the engine stops.
        """
        self.manager.close_all()
        self.manager.stop()
        self.manager.join()

def start_server():
    """
        Initialize and start the server.
    """
    if hasattr(cherrypy.engine, 'subscribe'):  # CherryPy >= 3.1
        pass
    else:
        raise Exception("This application requires CherryPy >= 3.1 or higher.")
        # cherrypy.engine.on_stop_engine_list.append(_save_data)

    # Some global configuration

    cherrypy.config.update({"server.ssl_certificate": os.path.join(os.path.dirname(__file__), "../cert.pem"),
                            "server.ssl_private_key": os.path.join(os.path.dirname(__file__), "../privkey.pem")})

    # cherrypy.config.update(config=os.path.join(script_dir, "config.ini"))
    app = cherrypy.tree.mount(Server(), '/',
                              {'/':
                                  {
                                      "tools.staticdir.root": os.path.abspath(
                                          os.path.join(os.path.dirname(__file__), "static")),
                                      "tools.decode.on": True,
                                      "tools.trailing_slash.on": True,
                                      "tools.staticdir.on": True,
                                      "tools.staticdir.index": "index.html",
                                      "tools.staticdir.dir": ""
                                  }
                              })

    if hasattr(cherrypy.engine, "signal_handler"):
        cherrypy.engine.signal_handler.subscribe()
    if hasattr(cherrypy.engine, "console_control_handler"):
        cherrypy.engine.console_control_handler.subscribe()
    cherrypy.engine.start()
    cherrypy.engine.block()

def cleanup(self):
        """
        Terminate all connections and clear the pool. Executed when the engine stops.
        """
        self.manager.close_all()
        self.manager.stop()
        self.manager.join()

def run(self):
        wsgi_app_https = ReverseProxied(ProxyFix(wsgi.WSGIPathInfoDispatcher({'/': app})))
        wsgi_app_http = ReverseProxied(ProxyFix(wsgi.WSGIPathInfoDispatcher({'/': adapter_app})))
        cherrypy.server.unsubscribe()
        cherrypy.config.update({'environment': 'production'})

        bind_addr = (self.config['listen'], self.config['port'])

        server_https = wsgi.WSGIServer(bind_addr=bind_addr,
                                       wsgi_app=wsgi_app_https)
        server_https.ssl_adapter = http_helpers.ssl_adapter(self.config['certificate'],
                                                            self.config['private_key'])
        ServerAdapter(cherrypy.engine, server_https).subscribe()
        logger.debug('WSGIServer starting... uid: %s, listen: %s:%s',
                     os.getuid(), bind_addr[0], bind_addr[1])

        for route in http_helpers.list_routes(app):
            logger.debug(route)

        if self.adapter_config['enabled']:
            bind_addr = (self.adapter_config['listen'], self.adapter_config['port'])
            server_adapter = wsgi.WSGIServer(bind_addr=bind_addr,
                                             wsgi_app=wsgi_app_http)
            ServerAdapter(cherrypy.engine, server_adapter).subscribe()
            logger.debug('WSGIServer starting... uid: %s, listen: %s:%s',
                         os.getuid(), bind_addr[0], bind_addr[1])

            for route in http_helpers.list_routes(adapter_app):
                logger.debug(route)

        else:
            logger.debug('Adapter server is disabled')

        try:
            cherrypy.engine.start()
            cherrypy.engine.wait(states.EXITING)
        except KeyboardInterrupt:
            logger.warning('Stopping xivo-ctid-ng: KeyboardInterrupt')
            cherrypy.engine.exit()

def stop(self):
        cherrypy.engine.exit()

def join(self):
        if cherrypy.engine.state == states.EXITING:
            cherrypy.engine.block()

def load(self):
        """Copy stored session data into this session instance."""
        data = self._load()
        # data is either None or a tuple (session_data, expiration_time)
        if data is None or data[1] < self.now():
            if self.debug:
                cherrypy.log('Expired session %r, flushing data.' % self.id,
                             'TOOLS.SESSIONS')
            self._data = {}
        else:
            if self.debug:
                cherrypy.log('Data loaded for session %r.' % self.id,
                             'TOOLS.SESSIONS')
            self._data = data[0]
        self.loaded = True

        # Stick the clean_thread in the class, not the instance.
        # The instances are created and destroyed per-request.
        cls = self.__class__
        if self.clean_freq and not cls.clean_thread:
            # clean_up is an instancemethod and not a classmethod,
            # so that tool config can be accessed inside the method.
            t = cherrypy.process.plugins.Monitor(
                cherrypy.engine, self.clean_up, self.clean_freq * 60,
                name='Session cleanup')
            t.subscribe()
            cls.clean_thread = t
            t.start()
            if self.debug:
                cherrypy.log('Started cleanup thread.', 'TOOLS.SESSIONS')