python类wsgi()的实例源码

def test_custom_max_header_line(self):
        self.flags(max_header_line=4096, group='wsgi')  # Default is 16384
        masakari.wsgi.Server("test_custom_max_header_line", None)
        self.assertEqual(CONF.wsgi.max_header_line,
                         eventlet.wsgi.MAX_HEADER_LINE)

def test_start_random_port_with_ipv6(self):
        server = masakari.wsgi.Server("test_random_port", None,
            host="::1", port=0)
        server.start()
        self.assertEqual("::1", server.host)
        self.assertNotEqual(0, server.port)
        server.stop()
        server.wait()

def test_server_pool_waitall(self):
        # test pools waitall method gets called while stopping server
        server = masakari.wsgi.Server("test_server", None,
            host="127.0.0.1")
        server.start()
        with mock.patch.object(server._pool,
                              'waitall') as mock_waitall:
            server.stop()
            server.wait()
            mock_waitall.assert_called_once_with()

def test_reset_pool_size_to_default(self):
        server = masakari.wsgi.Server("test_resize", None,
            host="127.0.0.1", max_url_len=16384)
        server.start()

        # Stopping the server, which in turn sets pool size to 0
        server.stop()
        self.assertEqual(server._pool.size, 0)

        # Resetting pool size to default
        server.reset()
        server.start()
        self.assertEqual(server._pool.size, CONF.wsgi.default_pool_size)

def test_keep_alive(self):
        self.flags(keep_alive=False, group='wsgi')

        # mocking eventlet spawn method to check it is called with
        # configured 'keep_alive' value.
        with mock.patch.object(eventlet,
                               'spawn') as mock_spawn:
            server = masakari.wsgi.Server("test_app", None,
                                      host="127.0.0.1", port=0)
            server.start()
            _, kwargs = mock_spawn.call_args
            self.assertEqual(CONF.wsgi.keep_alive,
                             kwargs['keepalive'])
            server.stop()

def __init__(self, environ, *args, **kwargs):
        if CONF.wsgi.secure_proxy_ssl_header:
            scheme = environ.get(CONF.wsgi.secure_proxy_ssl_header)
            if scheme:
                environ['wsgi.url_scheme'] = scheme
        super(Request, self).__init__(environ, *args, **kwargs)

def __call__(self, environ, start_response):
        r"""Subclasses will probably want to implement __call__ like this:

        @webob.dec.wsgify(RequestClass=Request)
        def __call__(self, req):
          # Any of the following objects work as responses:

          # Option 1: simple string
          res = 'message\n'

          # Option 2: a nicely formatted HTTP exception page
          res = exc.HTTPForbidden(explanation='Nice try')

          # Option 3: a webob Response object (in case you need to play with
          # headers, or you want to be treated like an iterable, or ...)
          res = Response()
          res.app_iter = open('somefile')

          # Option 4: any wsgi app to be run next
          res = self.application

          # Option 5: you can get a Response object for a wsgi app, too, to
          # play with headers etc
          res = req.get_response(self.application)

          # You can then just return your response...
          return res
          # ... or set req.response and return None.
          req.response = res

        See the end of http://pythonpaste.org/webob/modules/dec.html
        for more info.

        """
        raise NotImplementedError(_('You must implement __call__'))

def log_message(self, message):
        raise AttributeError('''\
eventlet.wsgi.server.log_message was deprecated and deleted.
Please use server.log.info instead.''')

def __init__(self, threads=1000, initialize_glance_store=False):
        os.umask(0o27)  # ensure files are created with the correct privileges
        self._logger = logging.getLogger("eventlet.wsgi.server")
        self.threads = threads
        self.children = set()
        self.stale_children = set()
        self.running = True
        self.initialize_glance_store = initialize_glance_store
        self.pgid = os.getpid()
        try:
            os.setpgid(self.pgid, self.pgid)
        except OSError:
            self.pgid = 0

def run_child(self):
        def child_hup(*args):
            """Shuts down child processes, existing requests are handled."""
            signal.signal(signal.SIGHUP, signal.SIG_IGN)
            eventlet.wsgi.is_accepting = False
            self.sock.close()

        pid = os.fork()
        if pid == 0:
            signal.signal(signal.SIGHUP, child_hup)
            signal.signal(signal.SIGTERM, signal.SIG_DFL)
            # ignore the interrupt signal to avoid a race whereby
            # a child worker receives the signal before the parent
            # and is respawned unnecessarily as a result
            signal.signal(signal.SIGINT, signal.SIG_IGN)
            # The child has no need to stash the unwrapped
            # socket, and the reference prevents a clean
            # exit on sighup
            self._sock = None
            self.run_server()
            LOG.info('Child %d exiting normally', os.getpid())
            # self.pool.waitall() is now called in wsgi's server so
            # it's safe to exit here
            sys.exit(0)
        else:
            LOG.info('Started child %s', pid)
            self.children.add(pid)

def _single_run(self, application, sock):
        """Start a WSGI server in a new green thread."""
        LOG.info("Starting single process server")
        eventlet.wsgi.server(sock, application, custom_pool=self.pool,
                             log=self._logger,
                             debug=False,
                             keepalive=CONF.http_keepalive,
                             socket_timeout=self.client_socket_timeout)

def serve_forever(self):
            self.logger = LoggingWrapper()
            eventlet.wsgi.server(self.server, self.handle, self.logger)

def test_relpath_config_not_found(self):
        self.flags(api_paste_config='api-paste.ini', group='wsgi')
        self.assertRaises(
            nova.exception.ConfigNotFound,
            nova.wsgi.Loader,
        )

def test_asbpath_config_not_found(self):
        self.flags(api_paste_config='/etc/nova/api-paste.ini', group='wsgi')
        self.assertRaises(
            nova.exception.ConfigNotFound,
            nova.wsgi.Loader,
        )

def setUp(self):
        super(TestLoaderNormalFilesystem, self).setUp()
        self.config = tempfile.NamedTemporaryFile(mode="w+t")
        self.config.write(self._paste_config.lstrip())
        self.config.seek(0)
        self.config.flush()
        self.loader = nova.wsgi.Loader(self.config.name)

def test_no_app(self):
        server = nova.wsgi.Server("test_app", None)
        self.assertEqual("test_app", server.name)

def test_custom_max_header_line(self):
        self.flags(max_header_line=4096, group='wsgi')  # Default is 16384
        nova.wsgi.Server("test_custom_max_header_line", None)
        self.assertEqual(CONF.wsgi.max_header_line,
                         eventlet.wsgi.MAX_HEADER_LINE)

def test_start_random_port_with_ipv6(self):
        server = nova.wsgi.Server("test_random_port", None,
            host="::1", port=0)
        server.start()
        self.assertEqual("::1", server.host)
        self.assertNotEqual(0, server.port)
        server.stop()
        server.wait()

def test_server_pool_waitall(self):
        # test pools waitall method gets called while stopping server
        server = nova.wsgi.Server("test_server", None,
            host="127.0.0.1")
        server.start()
        with mock.patch.object(server._pool,
                              'waitall') as mock_waitall:
            server.stop()
            server.wait()
            mock_waitall.assert_called_once_with()

def test_reset_pool_size_to_default(self):
        server = nova.wsgi.Server("test_resize", None,
            host="127.0.0.1", max_url_len=16384)
        server.start()

        # Stopping the server, which in turn sets pool size to 0
        server.stop()
        self.assertEqual(server._pool.size, 0)

        # Resetting pool size to default
        server.reset()
        server.start()
        self.assertEqual(server._pool.size, CONF.wsgi.default_pool_size)