Replaced the local fifo test with a test for the HTTP backend

This commit is contained in:
Fabio Manganiello 2018-01-04 18:29:03 +01:00
parent 1581685f62
commit 301c356c33
3 changed files with 75 additions and 84 deletions

View file

@ -1,5 +1,3 @@
backend.local: backend.http:
pusher: True port: 8123
request_fifo: /tmp/test-platypush-requests.fifo
response_fifo: /tmp/test-platypush-responses.fifo

73
tests/test_http.py Normal file
View file

@ -0,0 +1,73 @@
from .context import platypush, config_file, TestTimeoutException
import json
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 = 5
def setUp(self):
logging.basicConfig(level=logging.INFO, stream=sys.stdout)
backends = Config.get_backends()
self.assertTrue('http' in backends)
def test_request_exec_flow(self):
self.start_daemon()
time.sleep(2)
self.send_request()
def start_daemon(self):
def _f():
self.receiver = Daemon(config_file=config_file, requests_to_process=1)
self.receiver.start()
Thread(target=_f).start()
def on_timeout(self, msg):
def _f(): raise TestTimeoutException(msg)
return _f
def send_request(self):
set_timeout(seconds=self.timeout,
on_timeout=self.on_timeout('Receiver response timed out'))
response = requests.post(
u'http://localhost:8123/',
data = {
'msg': json.dumps({
'type': 'request',
'target': Config.get('device_id'),
'action': 'shell.exec',
'args': { 'cmd':'echo ping' }
})
}
)
clear_timeout()
response = Message.build(response.json())
self.assertTrue(isinstance(response, Response))
self.assertEqual(response.output.strip(), 'ping')
self.receiver.stop_app()
if __name__ == '__main__':
unittest.main()
# vim:sw=4:ts=4:et:

View file

@ -1,80 +0,0 @@
from .context import platypush, config_file, TestTimeoutException
import logging
import os
import sys
import time
import unittest
from threading import Thread
from platypush import Daemon
from platypush.config import Config
from platypush.pusher import Pusher
from platypush.utils import set_timeout, clear_timeout
class TestLocal(unittest.TestCase):
""" Tests the full flow on a local backend by executing a command through
the shell.exec plugin and getting the output """
timeout = 5
def setUp(self):
logging.basicConfig(level=logging.INFO, stream=sys.stdout)
backends = Config.get_backends()
self.assertTrue('local' in backends)
try: os.remove(Config.get_backends()['local']['request_fifo'])
except FileNotFoundError as e: pass
try: os.remove(Config.get_backends()['local']['response_fifo'])
except FileNotFoundError as e: pass
def test_local_shell_exec_flow(self):
self.start_sender()
self.start_receiver()
def on_response(self):
def _f(response):
logging.info("Received response: {}".format(response))
clear_timeout()
self.assertEqual(response.output.strip(), 'ping')
return _f
def on_timeout(self, msg):
def _f(): raise TestTimeoutException(msg)
return _f
def start_sender(self):
def _run_sender():
pusher = Pusher(config_file=config_file, backend='local',
on_response=self.on_response())
logging.info('Sending request')
pusher.send_request(target=Config.get('device_id'), action='shell.exec',
cmd='echo ping', timeout=None)
# Start the sender thread and wait for a response
set_timeout(seconds=self.timeout,
on_timeout=self.on_timeout('Receiver response timed out'))
self.sender = Thread(target=_run_sender)
self.sender.start()
def start_receiver(self):
set_timeout(seconds=self.timeout,
on_timeout=self.on_timeout('Sender request timed out'))
self.receiver = Daemon(config_file=config_file, requests_to_process=1)
self.receiver.start()
time.sleep(1)
if __name__ == '__main__':
unittest.main()
# vim:sw=4:ts=4:et: