-
Notifications
You must be signed in to change notification settings - Fork 7
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
changes: - introduce a new mite.utils.sleep function (so that tests donʼt sleep for a long time) - change the base class of HandledException to work around a bug in pytest-asyncio (pytest-dev/pytest-asyncio#126) - better async exception handling - fix event loop handling in mite_http - new mite.test module
- Loading branch information
Showing
5 changed files
with
121 additions
and
8 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,101 @@ | ||
from .context import Context | ||
from .config import default_config_loader | ||
from collections import defaultdict | ||
import pytest | ||
import asyncio | ||
import mite.utils as utils | ||
|
||
|
||
class _InterceptHttp: | ||
def __init__(self, http, obj): | ||
http._parent = self | ||
self._obj = obj | ||
object.__setattr__(self, "http", http) | ||
self._old_http = None | ||
|
||
def __setattr__(self, name, val): | ||
if name == "http": | ||
object.__setattr__(self, "_old_http", val) | ||
elif name in ("_obj", "_old_http"): | ||
object.__setattr__(self, name, val) | ||
else: | ||
self._obj.__setattr__(name, val) | ||
|
||
def __getattribute__(self, name): | ||
try: | ||
return object.__getattribute__(self, name) | ||
except AttributeError: | ||
return object.__getattribute__(self, "_obj").__getattribute__(name) | ||
|
||
def __delattr__(self, name): | ||
if name == "http": | ||
object.__delattr__(self, "_old_http") | ||
return | ||
self._obj.__delattr__(name) | ||
|
||
|
||
class _NewHttp: | ||
def __init__(self, requests): | ||
self._requests = requests | ||
|
||
async def get(self, *args, **kwargs): | ||
r = await self._parent._old_http.get(*args, **kwargs) | ||
self._requests['get'].append(r) | ||
return r | ||
|
||
async def post(self, *args, **kwargs): | ||
r = await self._parent._old_http.post(*args, **kwargs) | ||
self._requests['post'].append(r) | ||
return r | ||
|
||
async def delete(self, *args, **kwargs): | ||
r = await self._parent._old_http.delete(*args, **kwargs) | ||
self._requests['delete'].append(r) | ||
return r | ||
|
||
async def put(self, *args, **kwargs): | ||
r = await self._parent._old_http.put(*args, **kwargs) | ||
self._requests['put'].append(r) | ||
return r | ||
|
||
|
||
def http_spy(journey): | ||
async def wrapper(ctx, *args): | ||
requests = defaultdict(list) | ||
# The reason for doing this in terms of this (admittedly complicated) | ||
# dance is the following: we want to be able to say | ||
# "http_spy(journey)". The journey function is already wrapped with a | ||
# decorator (namely mite_http) which, when it is called with a ctx | ||
# argument, will inject a http attribute on that ctx. We want to pass | ||
# our own http argument instead. If we just add it now, it will be | ||
# overwritten by mite_http. We can't "undecorate" the journey, at | ||
# least not without intrusive modifications to the mite codebase. So | ||
# we use the above nasty hack with __getattribute__ and friends. An | ||
# alternative might be to investigate a prope mocking/spying library | ||
# in python... | ||
spy = _InterceptHttp(_NewHttp(requests), ctx) | ||
await journey(spy, *args) | ||
return {'http': requests} | ||
return wrapper | ||
|
||
|
||
async def run_single_journey(journey, datapool=None): | ||
messages = [] | ||
|
||
async def _send(message, **kwargs): | ||
messages.append((message, kwargs)) | ||
|
||
config = default_config_loader() | ||
|
||
ctx = Context(_send, config) | ||
|
||
if datapool is not None: | ||
dpi = datapool.checkout() | ||
result = await journey(ctx, *dpi.data) | ||
else: | ||
result = await journey(ctx) | ||
|
||
if result is None: | ||
result = {} | ||
result['messages'] = messages | ||
return result |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters