1
0
Fork 0
platypush/tests/utils.py

113 lines
2.4 KiB
Python
Raw Normal View History

import os
from typing import Optional
2023-03-27 22:54:13 +02:00
import requests
from platypush.message import Message
from platypush.message.response import Response
from platypush.utils import set_timeout, clear_timeout
from . import test_dir
# Default configuration folder for tests
conf_dir = os.path.join(test_dir, 'etc')
# Default configuration file for tests
config_file = os.path.join(conf_dir, 'config_test.yaml')
# Default request timeout in seconds
request_timeout = 10
# Default test user
test_user = 'platypush'
# Default test password
test_pass = 'test'
# Base URL
2021-03-06 16:25:37 +01:00
base_url = None
def set_base_url(url: str):
global base_url
base_url = url
class TimeoutException(RuntimeError):
"""
Exception raised in case of timeout.
"""
2023-03-27 22:54:13 +02:00
def __init__(self, msg: str = 'Timeout'):
self.msg = msg
2023-03-27 22:54:13 +02:00
def send_request(
action: str,
timeout: Optional[float] = None,
args: Optional[dict] = None,
parse_json: bool = True,
authenticate: bool = True,
**kwargs
):
if not timeout:
timeout = request_timeout
if not args:
args = {}
auth = (test_user, test_pass) if authenticate else kwargs.pop('auth', ())
set_timeout(seconds=timeout, on_timeout=on_timeout('Receiver response timed out'))
response = requests.post(
'{}/execute'.format(base_url),
auth=auth,
json={
'type': 'request',
'action': action,
'args': args,
2023-03-27 22:54:13 +02:00
},
**kwargs
)
clear_timeout()
if parse_json:
response = parse_response(response)
return response
def register_user(username: Optional[str] = None, password: Optional[str] = None):
if not username:
username = test_user
password = test_pass
2023-03-27 22:54:13 +02:00
set_timeout(
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,
'password': password,
'confirm_password': password,
},
)
clear_timeout()
return response
def on_timeout(msg):
2023-03-27 22:54:13 +02:00
def _f():
raise TimeoutException(msg)
return _f
def parse_response(response):
response = Message.build(response.json())
2023-03-27 22:54:13 +02:00
assert isinstance(response, Response), 'Expected Response type, got {}'.format(
response.__class__.__name__
)
return response