platypush/tests/test_http.py

134 lines
4.8 KiB
Python
Raw Normal View History

import os
from .context import config_file, TestTimeoutException
import logging
import requests
import sys
import time
import unittest
from threading import Thread
from platypush import Daemon
from platypush.config import Config
from platypush.message import Message
from platypush.message.response import Response
from platypush.utils import set_timeout, clear_timeout
class TestHttp(unittest.TestCase):
""" Tests the full flow of a request/response on the HTTP backend.
Runs a remote command over HTTP via shell.exec plugin and gets the output """
timeout = 10
sleep_secs = 10
db_file = '/tmp/platypush-tests.db'
test_user = 'platypush'
test_pass = 'test'
base_url = 'http://localhost:8123'
expected_registration_redirect = '{base_url}/register?redirect={base_url}/execute'.format(base_url=base_url)
expected_login_redirect = '{base_url}/login?redirect={base_url}/execute'.format(base_url=base_url)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.app = None
def setUp(self):
logging.basicConfig(level=logging.INFO, stream=sys.stdout)
backends = Config.get_backends()
self.assertTrue('http' in backends, 'Missing HTTP server configuration')
self.start_daemon()
logging.info('Sleeping {} seconds while waiting for the daemon to start up'.format(self.sleep_secs))
time.sleep(self.sleep_secs)
def test_http_flow(self):
# An /execute request performed before any user is registered should redirect to the registration page.
response = self.send_request()
self.assertEqual(self.expected_registration_redirect, response.url,
'No users registered, but the application did not redirect us to the registration page')
# Emulate a first user registration through form and get the session_token.
response = self.register_user()
self.assertGreater(len(response.history), 0, 'Redirect missing from the history')
self.assertTrue('session_token' in response.history[0].cookies, 'No session_token returned upon registration')
self.assertEqual('{base_url}/'.format(base_url=self.base_url), response.url,
'The registration form did not redirect to the main panel')
# After a first user has been registered any unauthenticated call to /execute should redirect to /login.
response = self.send_request()
self.assertEqual(self.expected_login_redirect, response.url,
'An unauthenticated request after user registration should result in a login redirect')
# A request authenticated with user/pass should succeed.
response = self.parse_response(self.send_request(auth=(self.test_user, self.test_pass)))
self.assertEqual(response.__class__, Response, 'The request did not return a proper Response object')
self.assertEqual(response.output.strip(), 'ping', 'The request did not return the expected output')
# A request with the wrong user/pass should fail.
response = self.send_request(auth=('wrong', 'wrong'))
self.assertEqual(self.expected_login_redirect, response.url,
'A request with wrong credentials should fail')
def start_daemon(self):
def _f():
self.app = Daemon(config_file=config_file)
self.app.start()
Thread(target=_f).start()
@staticmethod
def on_timeout(msg):
def _f(): raise TestTimeoutException(msg)
return _f
def send_request(self, **kwargs):
set_timeout(seconds=self.timeout,
on_timeout=self.on_timeout('Receiver response timed out'))
response = requests.post(
'{}/execute'.format(self.base_url),
json={
2018-01-29 16:34:00 +01:00
'type': 'request',
'target': Config.get('device_id'),
'action': 'shell.exec',
'args': {'cmd': 'echo ping'}
}, **kwargs
)
clear_timeout()
return response
def register_user(self):
set_timeout(seconds=self.timeout,
on_timeout=self.on_timeout('User registration response timed out'))
response = requests.post('{base_url}/register?redirect={base_url}/'.format(base_url=self.base_url), data={
'username': self.test_user,
'password': self.test_pass,
'confirm_password': self.test_pass,
})
clear_timeout()
return response
@staticmethod
def parse_response(response):
return Message.build(response.json())
def tearDown(self):
if self.app:
self.app.stop_app()
if os.path.isfile(self.db_file):
os.unlink(self.db_file)
if __name__ == '__main__':
unittest.main()
# vim:sw=4:ts=4:et: