82 lines
3.2 KiB
Python
82 lines
3.2 KiB
Python
"""Exercise the real FIFO and Telnet code without an AppDaemon installation."""
|
|
import ast
|
|
from functools import wraps
|
|
from pathlib import Path
|
|
import queue
|
|
import threading
|
|
import types
|
|
import unittest
|
|
|
|
|
|
source = Path(__file__).resolve().parents[1] / 'apps/smartprojector_benq_over_telnet.py'
|
|
tree = ast.parse(source.read_text())
|
|
tree.body = [node for node in tree.body if isinstance(node, (ast.ClassDef, ast.FunctionDef))]
|
|
namespace = dict(SmartSwitch=object, queue=queue, threading=threading, wraps=wraps)
|
|
exec(compile(tree, str(source), 'exec'), namespace)
|
|
SmartProjector = namespace['SmartProjector']
|
|
|
|
|
|
class ProjectorQueueTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.app = SmartProjector()
|
|
self.app._command_queue = queue.Queue()
|
|
self.app._command_stop = threading.Event()
|
|
self.app._command_worker = threading.Thread(target=self.app._process_commands)
|
|
self.app.telnet_ip, self.app.telnet_port = 'projector', 8000
|
|
self.app.log_info = self.app.log_warning = self.app.log_error = lambda *a, **k: None
|
|
self.app._command_worker.start()
|
|
|
|
def tearDown(self):
|
|
self.app._command_stop.set()
|
|
self.app._command_queue.put(None)
|
|
self.app._command_worker.join(2)
|
|
self.assertFalse(self.app._command_worker.is_alive())
|
|
|
|
def test_enqueue_returns_while_network_blocked_and_retries_stay_before_off(self):
|
|
entered, release = threading.Event(), threading.Event()
|
|
sent = []
|
|
|
|
class Session:
|
|
def __init__(self, *args, **kwargs):
|
|
assert kwargs['timeout'] == 2
|
|
def __enter__(self): return self
|
|
def __exit__(self, *args): pass
|
|
def write(self, data): sent.append(data)
|
|
def read_until(self, *args, **kwargs):
|
|
if len(sent) == 1:
|
|
entered.set()
|
|
if not release.wait(2): raise AssertionError('test gate timed out')
|
|
return b''
|
|
return b'*POW=ON#' if sent[-1] == b'*pow=on#' else b'*POW=OFF#'
|
|
|
|
namespace['telnetlib'] = types.SimpleNamespace(Telnet=Session)
|
|
self.assertIsNone(self.app.send_cmd('pow=on'))
|
|
self.assertTrue(entered.wait(2))
|
|
self.assertIsNone(self.app.send_cmd('pow=off'))
|
|
self.assertEqual(sent, [b'*pow=on#'])
|
|
release.set()
|
|
self.app._command_queue.join()
|
|
self.assertEqual(sent, [b'*pow=on#', b'*pow=on#', b'*pow=off#'])
|
|
|
|
def test_failed_operation_does_not_stop_queue(self):
|
|
completed = threading.Event()
|
|
def fail(app): raise RuntimeError('test failure')
|
|
self.app._command_queue.put((fail, (), {}))
|
|
self.app._command_queue.put((lambda app: completed.set(), (), {}))
|
|
self.assertTrue(completed.wait(2))
|
|
|
|
def test_nested_commands_execute_within_current_operation(self):
|
|
order = []
|
|
self.app._send_cmd = lambda cmd, *args: order.append(cmd)
|
|
def compound(app):
|
|
app.send_cmd('appmod=?')
|
|
app.send_cmd('appmod=cinema')
|
|
self.app._command_queue.put((compound, (), {}))
|
|
self.app.send_cmd('pow=off')
|
|
self.app._command_queue.join()
|
|
self.assertEqual(order, ['*appmod=?#', '*appmod=cinema#', '*pow=off#'])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|