Source code for pulsar.apps.test.wsgi

"""Classes for testing WSGI servers using the HttpClient
"""
from asyncio import Transport

from pulsar.apps import http
from pulsar.apps.wsgi import HttpServerResponse
from pulsar.utils.httpurl import http_parser


__all__ = ['HttpTestClient']


class DummyTransport(Transport):

    @property
    def transport(self):
        return self

    def close(self):
        pass

    def abort(self):
        pass


class DummyConnection(DummyTransport):
    """A class simulating a :class:`pulsar.Transport` to a :attr:`connection`

    .. attribute:: client

        The :class:`pulsar.Client` using this :class:`DummyTransport`

    .. attribute:: connection

        The *server* connection for this :attr:`client`
    """
    def __init__(self, producer, address):
        super().__init__()
        self.address = address
        self._producer = producer
        self._processed = 0
        self._current_consumer = None

    def __enter__(self):
        return self

    def __exit__(self, type, value, traceback):
        pass

    @property
    def _loop(self):
        return self._producer._loop

    def detach(self, discard=True):
        pass

    def current_consumer(self):
        consumer = http.HttpResponse(self._loop)
        consumer._connection = self
        consumer.server_side = None

        if self._producer.wsgi:
            server_side = HttpServerResponse(
                self._producer.wsgi, self._producer.test.cfg, loop=self._loop
            )
            server_side._connection = DummyServerConnection(
                server_side, consumer, self.address
            )
            consumer.server_side = server_side

        else:
            consumer.server_side = None
            consumer.message = b''
            consumer.in_parser = http_parser(kind=0)

        consumer.connection_made(self)
        self._current_consumer = consumer
        return consumer

    def write(self, chunk):
        consumer = self._current_consumer
        server_side = consumer.server_side
        if server_side:
            server_side.data_received(chunk)
        else:
            consumer.message += chunk
            assert consumer.in_parser.execute(chunk, len(chunk)) == len(chunk)
            if consumer.in_parser.is_message_complete():
                consumer.finished()


class DummyServerConnection(DummyTransport):

    def __init__(self, server, response, address):
        super().__init__({'sockname': ('127.0.0.1', 1234)})
        self.address = address
        self._current_consumer = server
        self.response = response

    def write(self, chunk):
        self.response.data_received(chunk)


class DummyConnectionPool:
    """A class for simulating a client connection with a server
    """
    def __init__(self, connector, **kwargs):
        self.connector = connector

    async def connect(self):
        return await self.connector()


[docs]class HttpTestClient(http.HttpClient):
    """A test client for http requests to a WSGI server handlers.

    .. attribute:: wsgi

        The WSGI server handler to test
    """
    client_version = 'Pulsar-Http-Test-Client'
    connection_pool = DummyConnectionPool

    def __init__(self, test=None, wsgi=None, **kwargs):
        self.test = test
        self.wsgi = wsgi
        super().__init__(**kwargs)

    async def create_connection(self, address, ssl=None):
        return DummyConnection(self, address)