platypush/tests/utils.py

import os
from typing import Optional

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
base_url = None


def set_base_url(url: str):
    global base_url
    base_url = url


class TimeoutException(RuntimeError):
    """
    Exception raised in case of timeout.
    """

    def __init__(self, msg: str = 'Timeout'):
        self.msg = msg


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(
        f'{base_url}/execute',
        auth=auth,
        timeout=10,
        json={
            'type': 'request',
            'action': action,
            'args': args,
        },
        **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

    set_timeout(
        seconds=request_timeout,
        on_timeout=on_timeout('User registration response timed out'),
    )
    response = requests.post(
        f'{base_url}/auth?type=register&redirect={base_url}/',
        data={
            'username': username,
            'password': password,
            'confirm_password': password,
        },
    )

    clear_timeout()
    return response


def on_timeout(msg):
    def _f():
        raise TimeoutException(msg)

    return _f


def parse_response(response):
    response = Message.build(response.json())
    assert isinstance(
        response, Response
    ), f'Expected Response type, got {response.__class__.__name__}'
    return response