platypush/tests/test_local.py

70 lines
1.9 KiB
Python
Raw Normal View History

2017-12-22 02:11:56 +01:00
from .context import platypush, config_file, TestTimeoutException
import os
import sys
import logging
import unittest
import threading
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):
timeout = 5
def setUp(self):
logging.basicConfig(level=logging.INFO, stream=sys.stdout)
backends = Config.get_backends()
self.assertTrue('local' in backends)
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))
self.assertEqual(response.output.strip(), 'ping')
# os._exit(0)
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: