python类web()的实例源码

def start(self,
                    addr: str = '',
                    port: int = 0,
                    ssl: SSLContext = None,
                    metrics_url: str = DEFAULT_METRICS_PATH,
                    discovery_agent=None) -> None:
        ''' Start the prometheus metrics HTTP(S) server.

        :param addr: the address to bind the server on. By default this is
          set to an empty string so that the service becomes available on
          all interfaces.

        :param port: The port to bind the server on. The default value is 0
          which will cause the server to bind to an ephemeral port. If you
          want the server to operate on a fixed port then you need to specifiy
          the port.

        :param ssl: a sslContext for use with TLS.

        :param metrics_url: The name of the endpoint route to expose
          prometheus metrics on. Defaults to '/metrics'.

        :param discovery_agent: an agent that can register the metrics
          service with a service discovery mechanism.

        :raises: Exception if the server could not be started.
        '''
        logger.debug(
            'Prometheus metrics server starting on %s:%s%s',
            addr, port, metrics_url)

        if self._svr:
            logger.warning(
                'Prometheus metrics server is already running')
            return

        self._svc = aiohttp.web.Application()
        self._metrics_url = metrics_url
        self._svc.router.add_route(
            GET, metrics_url, self.handle_metrics)
        self._handler = self._svc.make_handler()
        self._https = ssl is not None
        try:
            self._svr = await self.loop.create_server(
                self._handler, addr, port, ssl=ssl)
        except Exception:
            logger.exception('error creating metrics server')
            raise

        logger.debug('Prometheus metrics server started on %s', self.url)

        # register service with service discovery
        if discovery_agent:
            await discovery_agent.register(self)

def make_mocked_request(method, path, headers=None, *,
                        version=HttpVersion(1, 1), closing=False,
                        app=None,
                        reader=sentinel,
                        writer=sentinel,
                        transport=sentinel,
                        payload=sentinel,
                        sslcontext=None,
                        secure_proxy_ssl_header=None):
    """Creates mocked web.Request testing purposes.

    Useful in unit tests, when spinning full web server is overkill or
    specific conditions and errors are hard to trigger.

    """

    if version < HttpVersion(1, 1):
        closing = True

    if headers:
        hdrs = CIMultiDict(headers)
        raw_hdrs = [
            (k.encode('utf-8'), v.encode('utf-8')) for k, v in headers.items()]
    else:
        hdrs = CIMultiDict()
        raw_hdrs = []

    message = RawRequestMessage(method, path, version, hdrs,
                                raw_hdrs, closing, False)
    if app is None:
        app = _create_app_mock()

    if reader is sentinel:
        reader = mock.Mock()

    if writer is sentinel:
        writer = mock.Mock()

    if transport is sentinel:
        transport = _create_transport(sslcontext)

    if payload is sentinel:
        payload = mock.Mock()

    time_service = mock.Mock()
    time_service.time.return_value = 12345
    time_service.strtime.return_value = "Tue, 15 Nov 1994 08:12:31 GMT"

    task = mock.Mock()

    req = Request(message, payload,
                  transport, reader, writer,
                  time_service, task,
                  secure_proxy_ssl_header=secure_proxy_ssl_header)

    match_info = UrlMappingMatchInfo({}, mock.Mock())
    match_info.add_app(app)
    req._match_info = match_info

    return req

def test_start_runserver(tmpworkdir, caplog):
    mktree(tmpworkdir, {
        'app.py': """\
from aiohttp import web

async def hello(request):
    return web.Response(text='<h1>hello world</h1>', content_type='text/html')

async def has_error(request):
    raise ValueError()

def create_app(loop):
    app = web.Application()
    app.router.add_get('/', hello)
    app.router.add_get('/error', has_error)
    return app""",
        'static_dir/foo.js': 'var bar=1;',
    })
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    aux_app, aux_port, _ = runserver(app_path='app.py', static_path='static_dir')
    assert isinstance(aux_app, aiohttp.web.Application)
    assert aux_port == 8001
    start_app = aux_app.on_startup[0]
    stop_app = aux_app.on_shutdown[0]
    loop.run_until_complete(start_app(aux_app))

    async def check_callback(session):
        async with session.get('http://localhost:8000/') as r:
            assert r.status == 200
            assert r.headers['content-type'].startswith('text/html')
            text = await r.text()
            assert '<h1>hello world</h1>' in text
            assert '<script src="http://localhost:8001/livereload.js"></script>' in text

        async with session.get('http://localhost:8000/error') as r:
            assert r.status == 500
            assert 'raise ValueError()' in (await r.text())

    try:
        loop.run_until_complete(check_server_running(loop, check_callback))
    finally:
        loop.run_until_complete(stop_app(aux_app))
    assert (
        'adev.server.dft INFO: pre-check enabled, checking app factory\n'
        'adev.server.dft INFO: Starting aux server at http://localhost:8001 ?\n'
        'adev.server.dft INFO: serving static files from ./static_dir/ at http://localhost:8001/static/\n'
        'adev.server.dft INFO: Starting dev server at http://localhost:8000 ?\n'
    ) in caplog