revil-O/platypush
1
0
Fork 0
forked from platypush/platypush
Code Pull requests Activity

black fixes

Browse source
This commit is contained in:
Fabio Manganiello 2023-03-27 22:54:13 +02:00
parent 2a8a3f4394
commit 4f15758de9
Signed by untrusted user: blacklight
GPG key ID: D90FBA7F76362774
1 changed files with 31 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

37
tests/utils.py
View file

@ -1,7 +1,8 @@
import os import os
import requests
from typing import Optional from typing import Optional
import requests
from platypush.message import Message from platypush.message import Message
from platypush.message.response import Response from platypush.message.response import Response
from platypush.utils import set_timeout, clear_timeout from platypush.utils import set_timeout, clear_timeout
@ -36,12 +37,19 @@ class TimeoutException(RuntimeError):
""" """
Exception raised in case of timeout. Exception raised in case of timeout.
""" """
def __init__(self, msg: str = 'Timeout'): def __init__(self, msg: str = 'Timeout'):
self.msg = msg self.msg = msg
def send_request(action: str, timeout: Optional[float] = None, args: Optional[dict] = None, def send_request(
parse_json: bool = True, authenticate: bool = True, **kwargs): action: str,
timeout: Optional[float] = None,
args: Optional[dict] = None,
parse_json: bool = True,
authenticate: bool = True,
**kwargs
):
if not timeout: if not timeout:
timeout = request_timeout timeout = request_timeout
if not args: if not args:
@ -56,7 +64,8 @@ def send_request(action: str, timeout: Optional[float] = None, args: Optional[di
'type': 'request', 'type': 'request',
'action': action, 'action': action,
'args': args, 'args': args,
}, **kwargs },
**kwargs
) )
clear_timeout() clear_timeout()
@ -71,23 +80,33 @@ def register_user(username: Optional[str] = None, password: Optional[str] = None
username = test_user username = test_user
password = test_pass password = test_pass
set_timeout(seconds=request_timeout, on_timeout=on_timeout('User registration response timed out')) set_timeout(
response = requests.post('{base_url}/register?redirect={base_url}/'.format(base_url=base_url), data={ seconds=request_timeout,
on_timeout=on_timeout('User registration response timed out'),
)
response = requests.post(
'{base_url}/register?redirect={base_url}/'.format(base_url=base_url),
data={
'username': username, 'username': username,
'password': password, 'password': password,
'confirm_password': password, 'confirm_password': password,
}) },
)
clear_timeout() clear_timeout()
return response return response
def on_timeout(msg): def on_timeout(msg):
def _f(): raise TimeoutException(msg) def _f():
raise TimeoutException(msg)
return _f return _f
def parse_response(response): def parse_response(response):
response = Message.build(response.json()) response = Message.build(response.json())
assert isinstance(response, Response), 'Expected Response type, got {}'.format(response.__class__.__name__) assert isinstance(response, Response), 'Expected Response type, got {}'.format(
response.__class__.__name__
)
return response return response