platypush/tests/test_local.py

78 lines
2.2 KiB
Python
Raw Normal View History

2017-12-22 02:11:56 +01:00
from .context import platypush, config_file, TestTimeoutException
import os
2017-12-22 02:11:56 +01:00
import sys
import logging
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 """
2017-12-22 02:11:56 +01:00
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
2017-12-22 02:11:56 +01:00
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()
2017-12-22 02:11:56 +01:00
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.push(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()
if __name__ == '__main__':
unittest.main()
# vim:sw=4:ts=4:et: