Extension for delayed matching and response

[matez0]

Hi,

I have a system with 2 independent interfaces: A <----> System under test (SUT) <----> B
A typical communication sequence looks like:
A sends request to SUT that
triggers that SUT sends request to B;
B responds that
triggers that SUT responds.

I would like to use behave to test this communication sequence.
Currently, using pytest_httpserver, this looks like:

Then SUT sends some request to B
When B responds something
When A sends some request to SUT
Then SUT responds  something

That looks a bit confusing, isn't it?

Would an extension of HTTPServer be possible (and worth to do) which:

• accepts any incoming request;
• matches the request only when calling, e.g. expect_request(...) -> RequestHandler;
• responds only when calling, e.g. request_handler.respond_with_json({})

So, the dispatch would be delayed until the expect_request call and the response would be delayed until the respond_with_* call.

Usage would look like:

with BlockingHttpServer() as httpserver:
    endpoint = '/reserve-table'
    sut_response_handler = send_request_to_sut_async(endpoint)
    
    request_handler = httpserver.expect_request(endpoint)  # here we can get assertion error
    
    request_handler.respond_with_json({})
    
    sut_response_handler.expect_response('valagit')  # here we can get assertion error

[csernazs]

Hi,

"valagit" 😋

This seems a good idea, and it makes sense, however I'm not sure this library can help you. pytest-httpserver pytest plugin is designed to run the httpserver in a separate thread and the test subject (the client) running in the main thread.

Your case is the opposite: the client running in a separate thread and the server running in the main thread, so it could raise exceptions.

I think there are two options still:

• keeping the httpserver running in a thread and adding syncrhonization to it. I think Queue object from the standard library would be a good choice so you could override dispatch and it would pull one item from this queue and use that handler object from it. You could "feed" this queue from the test. There could be a separate queue filling with the test results so you could raise assertionerror there.
• you could keep the server in the mainthread so the new connection from the client won't be accept()-ed by the server automatically , it will be accepted when you call the dispatch() method on it manually. The entry point for a request is the application method in the HTTPServer so you would need to extract some functionality from werkzeug (and possibly basehttpserver) to implement a code which accept()s the connection and creates the request object and then passes it to the application method. In this way you would have the server running in the main thread, so no synchronization would be needed.

Both of these seems to be a hack for me on this library. I think I can help you creating a POC for the first one.

[csernazs]

I have pushed a very basic and ugly POC to the repo (see above), so you can get a glimpse into my idea.
I think it is a big hack and honestly I think implementing what you wanted from scratch would be better looking.

[matez0]

Hi, :)

Thank you for your work. It is almost there.

The request handler and most of the code and the fine documentation could be reusable for this special HTTP test server,
so I would not give up yet. This could be an "Integration with behave" section.

If I understand well, the PoC HTTPServer's serving thread does not block until no response was given by the test execution thread.

I suppose that each incoming request is served by a newly created thread which calls the application method.
From the test execution thread a HTTPServer thread is launched which creates these new threads.
Also there is a client thread (SUT) whose actions are triggered from the test execution thread.
The test execution thread calls the request handler's respond_with_ method which should trigger the response.

The dispatch in the PoC may block on getting the request handler from the queue. That is OK.
Then we should also block on calling respond waiting for the request handler's respond_with_ method is called.
For that, the RequestHandler's respond and respond_with_ methods have to be extended with synchronization mechanism.

[csernazs]

I suppose that each incoming request is served by a newly created thread which calls the application method.

This is not true. httpserver runs in its own thread, but the serving of the requests happens in a single thread. That means that it can work with one single http connection.
This is controlled by the make_server method in werkzeug, which is called from the start method of the HTTPServer of pytest-httpserver.

Then we should also block on calling respond waiting for the request handler's respond_with_ method is called.
For that, the RequestHandler's respond and respond_with_ methods have to be extended with synchronization mechanism.

This cannot be done easily as the API is different. It accepts the RequestHandler object, which defines the response, not the request. The RequestHandler contains the RequestMatcher object, which is used to match the request. So you cannot specify the RequestMatcher alone to the HTTPServer. You could specify a RequestHandler object with an incomplete RequestMatcher (eg. a None value) but that would mean you know the response before the request. If I understood correctly you would like to do the opposite: first you have the request definition in behave (where you can create the RequestMatcher object) and then you have the response definition (where you can create the RequestHandler object with the RequestMatcher created in the previous step). I think you should collect both RequestsHandler and RequestMatcher in behave once the request and response definitions are available then pass them in the form of RequestMatcher to the httpserver.

On the other hand, in case the API would be in the order you need (eg. accepting RequestMatcher which contains the RequestHandler), the server could still do nothing (from the client point of view) when the request is matched but no response is available, so there's no need to add synchronization (eg. blocking) here. This would make the API better as you could raise AssertionError when the request cannot be matched, but the feature description in behave would still contain the response definition so you could do that there.

I'm not saying that it is impossible but this would again require some clever trick to enforce it to the API which is currently working differently.

[matez0]

Hi,

This is my PoC which contains the logic needed for behave tests.
It is extendable to handle requests in multiple threads which allows testing requests with non deterministic order.
The characteristic of behave test steps is that the assertion has to be done at the end of the step, not later after some steps.

from contextlib import contextmanager
from multiprocessing import Pool
from urllib.parse import urlparse

from queue import Queue, Empty

import requests
import pytest

from werkzeug.wrappers import Request, Response

from pytest_httpserver import HTTPServer
from pytest_httpserver.httpserver import \
    UNDEFINED, URIPattern, HeaderValueMatcher, METHOD_ALL, QueryMatcher, RequestHandler

from typing import Any
from typing import Callable
from typing import Iterable
from typing import List
from typing import Mapping
from typing import Optional
from typing import Pattern
from typing import Union


class BRequestHandler(RequestHandler):
    def __init__(self, request, response_queue):
        super().__init__(None)
        self.request = request
        self.response_queue = response_queue

    def respond_with_data(self, *args, **kwargs):
        self._respond_via_queue(super().respond_with_data, *args, **kwargs)

    def respond_with_response(self, *args, **kwargs):
        self._respond_via_queue(super().respond_with_response, *args, **kwargs)

    def _respond_via_queue(self, repond_method, *args, **kwargs):
        repond_method(*args, **kwargs)
        self.response_queue.put_nowait(self.respond(self.request))


class BHttpServer(HTTPServer):
    def __init__(self, *args, timeout=30, **kwargs):
        super().__init__(*args, **kwargs)
        self.timeout = timeout
        self.request_queue = Queue()
        self.request_handlers = Queue()

    def assert_request(
        self,
        uri: Union[str, URIPattern, Pattern[str]],
        method: str = METHOD_ALL,
        data: Union[str, bytes, None] = None,
        data_encoding: str = "utf-8",
        headers: Optional[Mapping[str, str]] = None,
        query_string: Union[None, QueryMatcher, str, bytes, Mapping] = None,
        header_value_matcher: Optional[HeaderValueMatcher] = None,
        json: Any = UNDEFINED,
        timeout: int = 30
    ) -> BRequestHandler:

        matcher = self.create_matcher(
            uri,
            method=method.upper(),
            data=data,
            data_encoding=data_encoding,
            headers=headers,
            query_string=query_string,
            header_value_matcher=header_value_matcher,
            json=json,
        )

        try:
            request = self.request_queue.get(timeout=timeout)
        except Empty:
            raise AssertionError(f'Waiting for request {matcher} timed out')

        diff = matcher.difference(request)

        request_handler = BRequestHandler(request, Queue())

        self.request_handlers.put_nowait(request_handler)

        if diff:
            request_handler.respond_with_response(self.respond_nohandler(request))
            raise AssertionError(f'Request {matcher} does not match: {diff}')

        return request_handler

    def dispatch(self, request: Request) -> Response:
        self.request_queue.put_nowait(request)

        try:
            request_handler = self.request_handlers.get(timeout=self.timeout)
        except Empty:
            return self.respond_nohandler(request)

        try:
            return request_handler.response_queue.get(timeout=self.timeout)
        except Empty:
            assertion = AssertionError(f"No response for requesr: {request_handler.request}")
            self.add_assertion(assertion)
            raise assertion


@pytest.fixture
def httpserver():
    server = BHttpServer(timeout=5)
    server.start()

    yield server

    server.clear()
    if server.is_running():
        server.stop()


def test_blocking_http_server_behave_workflow(httpserver: BHttpServer):
    request = dict(
        method='GET',
        url=httpserver.url_for('/my/path'),
    )

    with when_a_request_is_being_sent_to_the_server(request) as server_connection:

        client_connection = then_the_server_gets_the_request(httpserver, request)

        response = {"foo": "bar"}

        when_the_server_responds_to(client_connection, response)

        then_the_response_is_got_from(server_connection, response)


def test_blocking_http_server_raises_assertion_error_when_request_does_not_match(httpserver: BHttpServer):
    request = dict(
        method='GET',
        url=httpserver.url_for('/my/path'),
    )

    with when_a_request_is_being_sent_to_the_server(request):

        with pytest.raises(AssertionError) as exc:
            httpserver.assert_request(uri='/not/my/path/')

        assert '/not/my/path/' in str(exc)
        assert 'does not match' in str(exc)


def test_blocking_http_server_raises_assertion_error_when_request_was_not_sent(httpserver: BHttpServer):
    with pytest.raises(AssertionError) as exc:
        httpserver.assert_request(uri='/my/path/', timeout=1)

    assert '/my/path/' in str(exc)
    assert 'timed out' in str(exc)


def test_blocking_http_server_ignores_when_request_is_not_asserted(httpserver: BHttpServer):
    request = dict(
        method='GET',
        url=httpserver.url_for('/my/path'),
    )
    httpserver.timeout = 1

    with when_a_request_is_being_sent_to_the_server(request) as server_connection:

        assert server_connection.get(timeout=9).text == 'No handler found for this request'


@contextmanager
def when_a_request_is_being_sent_to_the_server(request):
    with Pool(1) as pool:
        yield pool.apply_async(requests.request, kwds=request)


def then_the_server_gets_the_request(server, request):
    _request = dict(request)
    del _request['url']
    _request['uri'] = get_uri(request['url'])
    return server.assert_request(**_request)


def get_uri(url):
    url = urlparse(url)
    return '?'.join(item for item in [url.path, url.query] if item)


def when_the_server_responds_to(client_connection, response):
    client_connection.respond_with_json(response)


def then_the_response_is_got_from(server_connection, response):
    assert server_connection.get(timeout=9).json() == response

[csernazs]

Hi,

I'm trying to catch up with this. I started to look at your code but first I need to look at mine and read this issue again as a few months have passed since then.

Thanks for the code!

Zsolt

[matez0]

Hi Zsolt,

Thank you very much for checking the code.
I have already used it in production with a reference link to my comment.

In case of the contribution is acceptable, to make the process faster, I created a pull request with slight changes and added documentation.

BR,
Zoltán