Comments (7)
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 theapplication
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.
from pytest-httpserver.
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.
from pytest-httpserver.
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.
from pytest-httpserver.
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.
from pytest-httpserver.
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
from pytest-httpserver.
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
from pytest-httpserver.
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
from pytest-httpserver.
Related Issues (20)
- Matching failure for expect_request HOT 2
- 1.0.4: sphinx fails because directories layout issue HOT 2
- 1.0.4: sphinx warnings `reference target not found` HOT 6
- howto response more than one cookie HOT 2
- remove stale branches HOT 4
- Give a way to choose between listenning to IPV4 or IPV6 HOT 4
- Add client-side to the https code sample HOT 1
- Raise error if any request is received HOT 4
- Considering bringing plugin under pytest-dev HOT 2
- Add all the necessary imports in the code examples in the doc HOT 2
- Allow tests requiring multiple servers at once HOT 1
- turn off (disable) fixture httpserver for some tests HOT 3
- Improve documentation
- Server only listens on IPv4 HOT 16
- propagate assertion errors from custom handlers HOT 2
- Type hints HOT 13
- check_assertions usage HOT 1
- Async support? HOT 1
- MyPy complains when trying to update DEFAULT_MATCHERS HOT 4
Recommend Projects
-
React
A declarative, efficient, and flexible JavaScript library for building user interfaces.
-
Vue.js
๐ Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
-
Typescript
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
-
TensorFlow
An Open Source Machine Learning Framework for Everyone
-
Django
The Web framework for perfectionists with deadlines.
-
Laravel
A PHP framework for web artisans
-
D3
Bring data to life with SVG, Canvas and HTML. ๐๐๐
-
Recommend Topics
-
javascript
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
-
web
Some thing interesting about web. New door for the world.
-
server
A server is a program made to process requests and deliver data to clients.
-
Machine learning
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
-
Visualization
Some thing interesting about visualization, use data art
-
Game
Some thing interesting about game, make everyone happy.
Recommend Org
-
Facebook
We are working to build community through open source technology. NB: members must have two-factor auth.
-
Microsoft
Open source projects and samples from Microsoft.
-
Google
Google โค๏ธ Open Source for everyone.
-
Alibaba
Alibaba Open Source for everyone
-
D3
Data-Driven Documents codes.
-
Tencent
China tencent open source team.
from pytest-httpserver.