From d6cf6cf6bfef6bc36f03fb29725bf53df7055413 Mon Sep 17 00:00:00 2001 From: Ken Roberts Date: Sun, 28 Apr 2019 00:03:51 -0700 Subject: [PATCH] PJLink2 update V03 --- openlp/core/projectors/pjlink.py | 18 +- .../test_projector_pjlink_base_02.py | 682 ++++++++++++++++-- .../test_projector_pjlink_cmd_routing.py | 7 +- 3 files changed, 653 insertions(+), 54 deletions(-) diff --git a/openlp/core/projectors/pjlink.py b/openlp/core/projectors/pjlink.py index d2297ed52..f4eed7eaf 100644 --- a/openlp/core/projectors/pjlink.py +++ b/openlp/core/projectors/pjlink.py @@ -48,6 +48,7 @@ Website: http://pjlink.jbmia.or.jp/english/dl_class2.html """ import logging from codecs import decode +from copy import copy from PyQt5 import QtCore, QtNetwork @@ -276,6 +277,7 @@ class PJLink(QtNetwork.QTcpSocket): self.model_lamp = None # RLMP self.mute = None # AVMT self.other_info = None # INFO + self.pjlink_class = copy(PJLINK_CLASS) self.pjlink_name = None # NAME self.power = S_OFF # POWR self.serial_no = None # SNUM @@ -629,11 +631,14 @@ class PJLink(QtNetwork.QTcpSocket): :param salt: Optional salt for md5 hash initial authentication :param priority: Option to send packet now rather than queue it up """ - if QSOCKET_STATE[self.state()] != S_CONNECTED: + if QSOCKET_STATE[self.state()] != QSOCKET_STATE[S_CONNECTED]: log.warning('({ip}) send_command(): Not connected - returning'.format(ip=self.entry.name)) return self.reset_information() if cmd not in PJLINK_VALID_CMD: log.error('({ip}) send_command(): Invalid command requested - ignoring.'.format(ip=self.entry.name)) + if self.priority_queue or self.send_queue: + # Just in case there's already something to send + return self._send_command() return log.debug('({ip}) send_command(): Building cmd="{command}" opts="{data}"{salt}'.format(ip=self.entry.name, command=cmd, @@ -649,9 +654,9 @@ class PJLink(QtNetwork.QTcpSocket): options=opts, suffix=PJLINK_SUFFIX) if out in self.priority_queue: - log.debug('({ip}) send_command(): Already in priority queue - skipping'.format(ip=self.entry.name)) + log.warning('({ip}) send_command(): Already in priority queue - skipping'.format(ip=self.entry.name)) elif out in self.send_queue: - log.debug('({ip}) send_command(): Already in normal queue - skipping'.format(ip=self.entry.name)) + log.warning('({ip}) send_command(): Already in normal queue - skipping'.format(ip=self.entry.name)) else: if priority: log.debug('({ip}) send_command(): Adding to priority queue'.format(ip=self.entry.name)) @@ -672,7 +677,8 @@ class PJLink(QtNetwork.QTcpSocket): :param utf8: Send as UTF-8 string otherwise send as ASCII string """ if not data and not self.priority_queue and not self.send_queue: - log.debug('({ip}) _send_command(): Nothing to send - returning'.format(ip=self.entry.name)) + log.warning('({ip}) _send_command(): Nothing to send - returning'.format(ip=self.entry.name)) + self.send_busy = False return log.debug('({ip}) _send_command(data="{data}")'.format(ip=self.entry.name, data=data.strip() if data else data)) @@ -684,7 +690,7 @@ class PJLink(QtNetwork.QTcpSocket): log.debug('({ip}) _send_command(): Connection status: {data}'.format(ip=self.entry.name, data=conn_state)) if QSOCKET_STATE[self.state()] != S_CONNECTED: - log.debug('({ip}) _send_command() Not connected - abort'.format(ip=self.entry.name)) + log.warning('({ip}) _send_command() Not connected - abort'.format(ip=self.entry.name)) self.send_busy = False return self.disconnect_from_host() if data and data not in self.priority_queue: @@ -707,7 +713,7 @@ class PJLink(QtNetwork.QTcpSocket): log.debug('({ip}) _send_command(): Getting normal queued packet'.format(ip=self.entry.name)) else: # No data to send - log.debug('({ip}) _send_command(): No data to send'.format(ip=self.entry.name)) + log.warning('({ip}) _send_command(): No data to send'.format(ip=self.entry.name)) self.send_busy = False return self.send_busy = True diff --git a/tests/openlp_core/projectors/test_projector_pjlink_base_02.py b/tests/openlp_core/projectors/test_projector_pjlink_base_02.py index 7183389e9..e33c4a6fc 100644 --- a/tests/openlp_core/projectors/test_projector_pjlink_base_02.py +++ b/tests/openlp_core/projectors/test_projector_pjlink_base_02.py @@ -22,11 +22,12 @@ """ Package to test the openlp.core.projectors.pjlink base package. """ -from unittest import TestCase, skip +from unittest import TestCase from unittest.mock import call, patch import openlp.core.projectors.pjlink -from openlp.core.projectors.constants import S_NOT_CONNECTED +from openlp.core.projectors.constants import E_NETWORK, PJLINK_PREFIX, PJLINK_SUFFIX, QSOCKET_STATE, \ + S_CONNECTED, S_NOT_CONNECTED from openlp.core.projectors.db import Projector from openlp.core.projectors.pjlink import PJLink from tests.resources.projector.data import TEST1_DATA @@ -36,67 +37,658 @@ class TestPJLinkBase(TestCase): """ Tests for the PJLink module """ - @skip('Needs update to new setup') + def setUp(self): + """ + Initialize test state(s) + """ + # Default PJLink instance for tests + self.pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) + + def tearDown(self): + """ + Cleanup test state(s) + """ + del(self.pjlink) + + # ------------ Test PJLink._underscore_send_command ---------- + @patch.object(openlp.core.projectors.pjlink.PJLink, 'change_status') + @patch.object(openlp.core.projectors.pjlink.PJLink, 'write') + @patch.object(openlp.core.projectors.pjlink.PJLink, 'disconnect_from_host') + @patch.object(openlp.core.projectors.pjlink.PJLink, 'state') + @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information') + @patch.object(openlp.core.projectors.pjlink, 'log') + def test_local_send_command_network_error(self, mock_log, mock_reset, mock_state, mock_disconnect, mock_write, + mock_change_status): + """ + Test _underscore_send_command when possible network error occured + """ + # GIVEN: Test object + test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX, + clss=self.pjlink.pjlink_class, + suff=PJLINK_SUFFIX) + log_error_calls = [] + log_warning_calls = [call('({ip}) _send_command(): -1 received - ' + 'disconnecting from host'.format(ip=self.pjlink.name))] + log_debug_calls = [call('({ip}) _send_command(data="None")'.format(ip=self.pjlink.name)), + call('({ip}) _send_command(): priority_queue: []'.format(ip=self.pjlink.name)), + call("({ip}) _send_command(): send_queue: ['{data}\\r']".format(ip=self.pjlink.name, + data=test_command.strip())), + call('({ip}) _send_command(): Connection status: S_CONNECTED'.format(ip=self.pjlink.name)), + call('({ip}) _send_command(): Getting normal queued packet'.format(ip=self.pjlink.name)), + call('({ip}) _send_command(): Sending "{data}"'.format(ip=self.pjlink.name, + data=test_command.strip())) + ] + + mock_state.return_value = QSOCKET_STATE[S_CONNECTED] + mock_write.return_value = -1 + self.pjlink.send_queue = [test_command] + self.pjlink.priority_queue = [] + + # WHEN: _send_command called with no data and queue's emtpy + # Patch some attributes here since they are not available until after instantiation + with patch.object(self.pjlink, 'socket_timer') as mock_timer, \ + patch.object(self.pjlink, 'waitForBytesWritten') as mock_waitBytes: + mock_waitBytes.return_value = True + self.pjlink._send_command() + + # THEN: + mock_log.error.assert_has_calls(log_error_calls) + mock_log.warning.assert_has_calls(log_warning_calls) + mock_log.debug.assert_has_calls(log_debug_calls) + mock_change_status.called_with(E_NETWORK, 'Error while sending data to projector') + assert (not self.pjlink.send_queue), 'Send queue should be empty' + assert (not self.pjlink.priority_queue), 'Priority queue should be empty' + assert mock_timer.start.called, 'Timer should have been called' + assert (not mock_reset.called), 'reset_information() should not should have been called' + assert mock_disconnect.called, 'disconnect_from_host() should have been called' + assert self.pjlink.send_busy, 'send_busy should be True' + + @patch.object(openlp.core.projectors.pjlink.PJLink, 'state') + @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information') + @patch.object(openlp.core.projectors.pjlink, 'log') + def test_local_send_command_no_data(self, mock_log, mock_reset, mock_state): + """ + Test _underscore_send_command with no data to send + """ + # GIVEN: Test object + log_error_calls = [] + log_warning_calls = [call('({ip}) _send_command(): Nothing to send - returning'.format(ip=self.pjlink.name))] + log_debug_calls = [] + mock_state.return_value = S_CONNECTED + self.pjlink.send_queue = [] + self.pjlink.priority_queue = [] + + # WHEN: _send_command called with no data and queue's emtpy + # Patch some attributes here since they are not available until after instantiation + with patch.object(self.pjlink, 'socket_timer') as mock_timer: + self.pjlink._send_command() + + # THEN: + mock_log.error.assert_has_calls(log_error_calls) + mock_log.warning.assert_has_calls(log_warning_calls) + mock_log.debug.assert_has_calls(log_debug_calls) + assert (not self.pjlink.send_queue), 'Send queue should be empty' + assert (not self.pjlink.priority_queue), 'Priority queue should be empty' + assert (not mock_timer.called), 'Timer should not have been called' + assert (not mock_reset.called), 'reset_information() should not have been called' + + @patch.object(openlp.core.projectors.pjlink.PJLink, 'state') + @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information') + @patch.object(openlp.core.projectors.pjlink, 'log') + def test_local_send_command_no_data_queue_check(self, mock_log, mock_reset, mock_state): + """ + Test _underscore_send_command last queue length check + """ + # GIVEN: Test object + log_error_calls = [] + log_warning_calls = [call('({ip}) _send_command(): No data to send'.format(ip=self.pjlink.name))] + log_debug_calls = [] + mock_state.return_value = QSOCKET_STATE[S_CONNECTED] + self.pjlink.priority_queue = [] + + # WHEN: _send_command called with no data and queue's emtpy + # Patch some attributes here since they are not available until after instantiation + with patch.object(self.pjlink, 'socket_timer') as mock_timer, \ + patch.object(self.pjlink, 'send_queue') as mock_queue: + # Unlikely case of send_queue not really empty, but len(send_queue) returns 0 + mock_queue.return_value = ['test'] + mock_queue.__len__.return_value = 0 + self.pjlink._send_command(data=None) + + # THEN: + mock_log.error.assert_has_calls(log_error_calls) + mock_log.warning.assert_has_calls(log_warning_calls) + mock_log.debug.assert_has_calls(log_debug_calls) + assert (not self.pjlink.priority_queue), 'Priority queue should be empty' + assert (not mock_timer.called), 'Timer should not have been called' + assert (not mock_reset.called), 'reset_information() should not have been called' + + @patch.object(openlp.core.projectors.pjlink.PJLink, 'write') + @patch.object(openlp.core.projectors.pjlink.PJLink, 'disconnect_from_host') + @patch.object(openlp.core.projectors.pjlink.PJLink, 'state') + @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information') + @patch.object(openlp.core.projectors.pjlink, 'log') + def test_local_send_command_normal_send(self, mock_log, mock_reset, mock_state, mock_disconnect, mock_write): + """ + Test _underscore_send_command using normal queue + """ + # GIVEN: Test object + test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX, + clss=self.pjlink.pjlink_class, + suff=PJLINK_SUFFIX) + log_error_calls = [] + log_warning_calls = [] + log_debug_calls = [call('({ip}) _send_command(data="None")'.format(ip=self.pjlink.name)), + call('({ip}) _send_command(): priority_queue: []'.format(ip=self.pjlink.name)), + call("({ip}) _send_command(): send_queue: ['{data}\\r']".format(ip=self.pjlink.name, + data=test_command.strip())), + call('({ip}) _send_command(): Connection status: S_CONNECTED'.format(ip=self.pjlink.name)), + call('({ip}) _send_command(): Getting normal queued packet'.format(ip=self.pjlink.name)), + call('({ip}) _send_command(): Sending "{data}"'.format(ip=self.pjlink.name, + data=test_command.strip())) + ] + + mock_state.return_value = QSOCKET_STATE[S_CONNECTED] + mock_write.return_value = len(test_command) + self.pjlink.send_queue = [test_command] + self.pjlink.priority_queue = [] + + # WHEN: _send_command called with no data and queue's emtpy + # Patch some attributes here since they are not available until after instantiation + with patch.object(self.pjlink, 'socket_timer') as mock_timer, \ + patch.object(self.pjlink, 'waitForBytesWritten') as mock_waitBytes: + mock_waitBytes.return_value = True + self.pjlink._send_command() + + # THEN: + mock_log.error.assert_has_calls(log_error_calls) + mock_log.warning.assert_has_calls(log_warning_calls) + mock_log.debug.assert_has_calls(log_debug_calls) + assert (not self.pjlink.send_queue), 'Send queue should be empty' + assert (not self.pjlink.priority_queue), 'Priority queue should be empty' + assert mock_timer.start.called, 'Timer should have been called' + assert (not mock_reset.called), 'reset_information() should not have been called' + assert (not mock_disconnect.called), 'disconnect_from_host() should not have been called' + assert self.pjlink.send_busy, 'send_busy flag should be True' + + @patch.object(openlp.core.projectors.pjlink.PJLink, 'disconnect_from_host') + @patch.object(openlp.core.projectors.pjlink.PJLink, 'state') + @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information') + @patch.object(openlp.core.projectors.pjlink, 'log') + def test_local_send_command_not_connected(self, mock_log, mock_reset, mock_state, mock_disconnect): + """ + Test _underscore_send_command when not connected + """ + # GIVEN: Test object + test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX, + clss=self.pjlink.pjlink_class, + suff=PJLINK_SUFFIX) + log_error_calls = [] + log_warning_calls = [call('({ip}) _send_command() Not connected - abort'.format(ip=self.pjlink.name))] + log_debug_calls = [call('({ip}) _send_command(data="None")'.format(ip=self.pjlink.name)), + call('({ip}) _send_command(): priority_queue: []'.format(ip=self.pjlink.name)), + call("({ip}) _send_command(): send_queue: ['%1CLSS ?\\r']".format(ip=self.pjlink.name)), + call('({ip}) _send_command(): Connection status: S_OK'.format(ip=self.pjlink.name))] + mock_state.return_value = S_NOT_CONNECTED + self.pjlink.send_queue = [test_command] + self.pjlink.priority_queue = [] + + # WHEN: _send_command called with no data and queue's emtpy + # Patch here since pjlink does not have socket_timer until after instantiation + with patch.object(self.pjlink, 'socket_timer') as mock_timer: + self.pjlink._send_command() + + # THEN: + mock_log.error.assert_has_calls(log_error_calls) + mock_log.warning.assert_has_calls(log_warning_calls) + mock_log.debug.assert_has_calls(log_debug_calls) + assert (self.pjlink.send_queue == [test_command]), 'Send queue should have one entry' + assert (not self.pjlink.priority_queue), 'Priority queue should be empty' + assert (not mock_timer.called), 'Timer should not have been called' + assert (not mock_reset.called), 'reset_information() should not have been called' + assert mock_disconnect.called, 'disconnect_from_host() should have been called' + assert (not self.pjlink.send_busy), 'send_busy flag should be False' + + @patch.object(openlp.core.projectors.pjlink.PJLink, 'write') + @patch.object(openlp.core.projectors.pjlink.PJLink, 'disconnect_from_host') + @patch.object(openlp.core.projectors.pjlink.PJLink, 'state') + @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information') + @patch.object(openlp.core.projectors.pjlink, 'log') + def test_local_send_command_priority_send(self, mock_log, mock_reset, mock_state, mock_disconnect, mock_write): + """ + Test _underscore_send_command with priority queue + """ + # GIVEN: Test object + test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX, + clss=self.pjlink.pjlink_class, + suff=PJLINK_SUFFIX) + log_error_calls = [] + log_warning_calls = [] + log_debug_calls = [call('({ip}) _send_command(data="{data}")'.format(ip=self.pjlink.name, + data=test_command.strip())), + call('({ip}) _send_command(): priority_queue: []'.format(ip=self.pjlink.name)), + call('({ip}) _send_command(): send_queue: []'.format(ip=self.pjlink.name)), + call('({ip}) _send_command(): Connection status: S_CONNECTED'.format(ip=self.pjlink.name)), + call('({ip}) _send_command(): Priority packet - ' + 'adding to priority queue'.format(ip=self.pjlink.name)), + call('({ip}) _send_command(): Getting priority queued packet'.format(ip=self.pjlink.name)), + call('({ip}) _send_command(): Sending "{data}"'.format(ip=self.pjlink.name, + data=test_command.strip())) + ] + + mock_state.return_value = QSOCKET_STATE[S_CONNECTED] + mock_write.return_value = len(test_command) + self.pjlink.send_queue = [] + self.pjlink.priority_queue = [] + + # WHEN: _send_command called with no data and queue's emtpy + # Patch some attributes here since they are not available until after instantiation + with patch.object(self.pjlink, 'socket_timer') as mock_timer, \ + patch.object(self.pjlink, 'waitForBytesWritten') as mock_waitBytes: + mock_waitBytes.return_value = True + self.pjlink._send_command(data=test_command) + + # THEN: + mock_log.error.assert_has_calls(log_error_calls) + mock_log.warning.assert_has_calls(log_warning_calls) + mock_log.debug.assert_has_calls(log_debug_calls) + assert (not self.pjlink.send_queue), 'Send queue should be empty' + assert (not self.pjlink.priority_queue), 'Priority queue should be empty' + assert mock_timer.start.called, 'Timer should have been called' + assert (not mock_reset.called), 'reset_information() should not have been called' + assert (not mock_disconnect.called), 'disconnect_from_host() should not have been called' + assert self.pjlink.send_busy, 'send_busy flag should be True' + + @patch.object(openlp.core.projectors.pjlink.PJLink, 'write') + @patch.object(openlp.core.projectors.pjlink.PJLink, 'disconnect_from_host') + @patch.object(openlp.core.projectors.pjlink.PJLink, 'state') + @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information') + @patch.object(openlp.core.projectors.pjlink, 'log') + def test_local_send_command_priority_send_with_normal_queue(self, mock_log, mock_reset, mock_state, + mock_disconnect, mock_write): + """ + Test _underscore_send_command with priority queue when normal queue active + """ + # GIVEN: Test object + test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX, + clss=self.pjlink.pjlink_class, + suff=PJLINK_SUFFIX) + log_error_calls = [] + log_warning_calls = [] + log_debug_calls = [call('({ip}) _send_command(data="{data}")'.format(ip=self.pjlink.name, + data=test_command.strip())), + call('({ip}) _send_command(): priority_queue: []'.format(ip=self.pjlink.name)), + call("({ip}) _send_command(): send_queue: ['{data}\\r']".format(ip=self.pjlink.name, + data=test_command.strip())), + call('({ip}) _send_command(): Connection status: S_CONNECTED'.format(ip=self.pjlink.name)), + call('({ip}) _send_command(): Priority packet - ' + 'adding to priority queue'.format(ip=self.pjlink.name)), + call('({ip}) _send_command(): Getting priority queued packet'.format(ip=self.pjlink.name)), + call('({ip}) _send_command(): Sending "{data}"'.format(ip=self.pjlink.name, + data=test_command.strip())) + ] + + mock_state.return_value = QSOCKET_STATE[S_CONNECTED] + mock_write.return_value = len(test_command) + self.pjlink.send_queue = [test_command] + self.pjlink.priority_queue = [] + + # WHEN: _send_command called with no data and queue's emtpy + # Patch some attributes here since they are not available until after instantiation + with patch.object(self.pjlink, 'socket_timer') as mock_timer, \ + patch.object(self.pjlink, 'waitForBytesWritten') as mock_waitBytes: + mock_waitBytes.return_value = True + self.pjlink._send_command(data=test_command) + + # THEN: + mock_log.error.assert_has_calls(log_error_calls) + mock_log.warning.assert_has_calls(log_warning_calls) + mock_log.debug.assert_has_calls(log_debug_calls) + assert self.pjlink.send_queue, 'Send queue should have one entry' + assert (not self.pjlink.priority_queue), 'Priority queue should be empty' + assert mock_timer.start.called, 'Timer should have been called' + assert (not mock_reset.called), 'reset_information() should not have been called' + assert (not mock_disconnect.called), 'disconnect_from_host() should not have been called' + assert self.pjlink.send_busy, 'send_busy flag should be True' + + @patch.object(openlp.core.projectors.pjlink.PJLink, 'state') + @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information') + @patch.object(openlp.core.projectors.pjlink, 'log') + def test_local_send_command_send_busy_normal_queue(self, mock_log, mock_reset, mock_state): + """ + Test _underscore_send_command send_busy flag with normal queue + """ + # GIVEN: Test object + test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX, + clss=self.pjlink.pjlink_class, + suff=PJLINK_SUFFIX) + log_error_calls = [] + log_warning_calls = [] + log_debug_calls = [call('({ip}) _send_command(data="None")'.format(ip=self.pjlink.name)), + call('({ip}) _send_command(): priority_queue: []'.format(ip=self.pjlink.name)), + call("({ip}) _send_command(): send_queue: ['{data}\\r']".format(ip=self.pjlink.name, + data=test_command.strip())), + call('({ip}) _send_command(): Connection status: S_CONNECTED'.format(ip=self.pjlink.name)), + call('({ip}) _send_command(): Still busy, returning'.format(ip=self.pjlink.name)), + call('({ip}) _send_command(): Priority queue = []'.format(ip=self.pjlink.name)), + call("({ip}) _send_command(): Normal queue = " + "['{data}\\r']".format(ip=self.pjlink.name, data=test_command.strip()))] + + mock_state.return_value = QSOCKET_STATE[S_CONNECTED] + self.pjlink.send_busy = True + self.pjlink.send_queue = [test_command] + self.pjlink.priority_queue = [] + + # WHEN: _send_command called with no data and queue's emtpy + # Patch some attributes here since they are not available until after instantiation + with patch.object(self.pjlink, 'socket_timer') as mock_timer: + self.pjlink._send_command() + + # THEN: + mock_log.error.assert_has_calls(log_error_calls) + mock_log.warning.assert_has_calls(log_warning_calls) + mock_log.debug.assert_has_calls(log_debug_calls) + assert self.pjlink.send_queue, 'Send queue should have one entry' + assert (not self.pjlink.priority_queue), 'Priority queue should be empty' + assert (not mock_timer.start.called), 'Timer should not have been called' + assert (not mock_reset.called), 'reset_information() should not have been called' + assert self.pjlink.send_busy, 'send_busy flag should be True' + + @patch.object(openlp.core.projectors.pjlink.PJLink, 'state') + @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information') + @patch.object(openlp.core.projectors.pjlink, 'log') + def test_local_send_command_send_busy_priority_queue(self, mock_log, mock_reset, mock_state): + """ + Test _underscore_send_command send_busy flag with priority queue + """ + # GIVEN: Test object + test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX, + clss=self.pjlink.pjlink_class, + suff=PJLINK_SUFFIX) + log_error_calls = [] + log_warning_calls = [] + log_debug_calls = [call('({ip}) _send_command(data="None")'.format(ip=self.pjlink.name)), + call("({ip}) _send_command(): priority_queue: " + "['{data}\\r']".format(ip=self.pjlink.name, + data=test_command.strip())), + call('({ip}) _send_command(): send_queue: []'.format(ip=self.pjlink.name)), + call('({ip}) _send_command(): Connection status: S_CONNECTED'.format(ip=self.pjlink.name)), + call('({ip}) _send_command(): Still busy, returning'.format(ip=self.pjlink.name)), + call("({ip}) _send_command(): Priority queue = " + "['{data}\\r']".format(ip=self.pjlink.name, data=test_command.strip())), + call('({ip}) _send_command(): Normal queue = []'.format(ip=self.pjlink.name)) + ] + + mock_state.return_value = QSOCKET_STATE[S_CONNECTED] + self.pjlink.send_busy = True + self.pjlink.send_queue = [] + self.pjlink.priority_queue = [test_command] + + # WHEN: _send_command called with no data and queue's emtpy + # Patch some attributes here since they are not available until after instantiation + with patch.object(self.pjlink, 'socket_timer') as mock_timer: + self.pjlink._send_command() + + # THEN: + mock_log.error.assert_has_calls(log_error_calls) + mock_log.warning.assert_has_calls(log_warning_calls) + mock_log.debug.assert_has_calls(log_debug_calls) + assert (not self.pjlink.send_queue), 'Send queue should be empty' + assert self.pjlink.priority_queue, 'Priority queue should have one entry' + assert (not mock_timer.start.called), 'Timer should not have been called' + assert (not mock_reset.called), 'reset_information() should not have been called' + assert self.pjlink.send_busy, 'send_busy flag should be True' + + # ------------ Test PJLink.send_command ---------- @patch.object(openlp.core.projectors.pjlink.PJLink, 'state') @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information') @patch.object(openlp.core.projectors.pjlink.PJLink, '_send_command') @patch.object(openlp.core.projectors.pjlink, 'log') - def test_send_command_no_data(self, mock_log, mock_send_command, mock_reset, mock_state): + def test_send_command_add_normal_command(self, mock_log, mock_send_command, mock_reset, mock_state): """ - Test _send_command with no data to send + Test send_command adding normal queue item """ # GIVEN: Test object - log_warning_calls = [call('({ip}) send_command(): Not connected - returning'.format(ip=TEST1_DATA['name']))] + test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX, + clss=self.pjlink.pjlink_class, + suff=PJLINK_SUFFIX) + log_error_calls = [] + log_warning_calls = [] + log_debug_calls = [call('({ip}) send_command(): Building cmd="CLSS" opts="?"'.format(ip=self.pjlink.name)), + call('({ip}) send_command(): Adding to normal queue'.format(ip=self.pjlink.name))] + mock_state.return_value = S_CONNECTED - log_debug_calls = [call('PJlink(projector="< Projector(id="None", ip="111.111.111.111", port="1111", ' - 'mac_adx="11:11:11:11:11:11", pin="1111", name="___TEST_ONE___", ' - 'location="location one", notes="notes one", pjlink_name="None", ' - 'pjlink_class="None", manufacturer="None", model="None", ' - 'serial_no="Serial Number 1", other="None", sources="None", source_list="[]", ' - 'model_filter="Filter type 1", model_lamp="Lamp type 1", ' - 'sw_version="Version 1") >", args="()" kwargs="{\'no_poll\': True}")'), - call('PJlinkCommands(args=() kwargs={})')] - mock_state.return_value = S_NOT_CONNECTED - pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) - pjlink.send_queue = [] - pjlink.priority_queue = [] + # Patch here since pjlink does not have priority or send queue's until instantiated + with patch.object(self.pjlink, 'send_queue') as mock_send, \ + patch.object(self.pjlink, 'priority_queue') as mock_priority: - # WHEN: _send_command called with no data and queue's empty - pjlink.send_command(cmd='DONTCARE') + # WHEN: send_command called with valid normal command + self.pjlink.send_command(cmd='CLSS') + + # THEN: + mock_send.append.called_with(test_command) + mock_priority.append.called is False + mock_log.debug.assert_has_calls(log_debug_calls) + mock_log.warning.assert_has_calls(log_warning_calls) + mock_log.error.assert_has_calls(log_error_calls) + assert (not mock_reset.called), 'reset_information() should not have been called' + assert mock_send_command.called, '_underscore_send_command() should have been called' + + @patch.object(openlp.core.projectors.pjlink.PJLink, 'state') + @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information') + @patch.object(openlp.core.projectors.pjlink.PJLink, '_send_command') + @patch.object(openlp.core.projectors.pjlink, 'log') + def test_send_command_add_priority_command(self, mock_log, mock_send_command, mock_reset, mock_state): + """ + Test _send_command adding priority queue item + """ + # GIVEN: Test object + test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX, + clss=self.pjlink.pjlink_class, + suff=PJLINK_SUFFIX) + log_error_calls = [] + log_warning_calls = [] + log_debug_calls = [call('({ip}) send_command(): Building cmd="CLSS" opts="?"'.format(ip=self.pjlink.name)), + call('({ip}) send_command(): Adding to priority queue'.format(ip=self.pjlink.name))] + mock_state.return_value = S_CONNECTED + + # Patch here since pjlink does not have priority or send queue's until instantiated + with patch.object(self.pjlink, 'send_queue') as mock_send, \ + patch.object(self.pjlink, 'priority_queue') as mock_priority: + + # WHEN: send_command called with valid priority command + self.pjlink.send_command(cmd='CLSS', priority=True) + + # THEN: + mock_log.debug.assert_has_calls(log_debug_calls) + mock_log.warning.assert_has_calls(log_warning_calls) + mock_log.error.assert_has_calls(log_error_calls) + mock_priority.append.assert_called_with(test_command) + assert (not mock_send.append.called), 'send_queue should not have changed' + assert (not mock_reset.called), 'reset_information() should not have been called' + assert mock_send_command.called, '_underscore_send_command() should have been called' + + @patch.object(openlp.core.projectors.pjlink.PJLink, 'state') + @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information') + @patch.object(openlp.core.projectors.pjlink.PJLink, '_send_command') + @patch.object(openlp.core.projectors.pjlink, 'log') + def test_send_command_duplicate_normal_command(self, mock_log, mock_send_command, mock_reset, mock_state): + """ + Test send_command with duplicate item for normal queue + """ + # GIVEN: Test object + test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX, + clss=self.pjlink.pjlink_class, + suff=PJLINK_SUFFIX) + log_error_calls = [] + log_warning_calls = [call('({ip}) send_command(): Already in normal queue - ' + 'skipping'.format(ip=self.pjlink.name))] + log_debug_calls = [call('({ip}) send_command(): Building cmd="CLSS" opts="?"'.format(ip=self.pjlink.name))] + mock_state.return_value = S_CONNECTED + self.pjlink.send_queue = [test_command] + self.pjlink.priority_queue = [] + + # WHEN: send_command called with same command in normal queue + self.pjlink.send_command(cmd='CLSS') # THEN: mock_log.debug.assert_has_calls(log_debug_calls) mock_log.warning.assert_has_calls(log_warning_calls) - assert mock_reset.called is True - assert mock_reset.called is True + mock_log.error.assert_has_calls(log_error_calls) + assert (self.pjlink.send_queue == [test_command]), 'Send queue should have one entry' + assert (not self.pjlink.priority_queue), 'Priority queue should be empty' + assert (not mock_reset.called), 'reset_information() should not have been called' + assert mock_send_command.called, '_underscore_send_command() should have been called' - @skip('Needs update to new setup') + @patch.object(openlp.core.projectors.pjlink.PJLink, 'state') + @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information') + @patch.object(openlp.core.projectors.pjlink.PJLink, '_send_command') @patch.object(openlp.core.projectors.pjlink, 'log') - def test_local_send_command_no_data(self, mock_log): + def test_send_command_duplicate_priority_command(self, mock_log, mock_send_command, mock_reset, mock_state): """ - Test _send_command with no data to send + Test send_command with duplicate item for priority queue """ # GIVEN: Test object - log_debug_calls = [call('PJlink(projector="< Projector(id="None", ip="111.111.111.111", port="1111", ' - 'mac_adx="11:11:11:11:11:11", pin="1111", name="___TEST_ONE___", ' - 'location="location one", notes="notes one", pjlink_name="None", ' - 'pjlink_class="None", manufacturer="None", model="None", ' - 'serial_no="Serial Number 1", other="None", sources="None", source_list="[]", ' - 'model_filter="Filter type 1", model_lamp="Lamp type 1", ' - 'sw_version="Version 1") >", args="()" kwargs="{\'no_poll\': True}")'), - call('PJlinkCommands(args=() kwargs={})'), - call('(___TEST_ONE___) reset_information() connect status is S_NOT_CONNECTED'), - call('(___TEST_ONE___) _send_command(): Nothing to send - returning')] + test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX, + clss=self.pjlink.pjlink_class, + suff=PJLINK_SUFFIX) + log_error_calls = [] + log_warning_calls = [call('({ip}) send_command(): Already in priority queue - ' + 'skipping'.format(ip=self.pjlink.name))] + log_debug_calls = [call('({ip}) send_command(): Building cmd="CLSS" opts="?"'.format(ip=self.pjlink.name))] + mock_state.return_value = S_CONNECTED + self.pjlink.send_queue = [] + self.pjlink.priority_queue = [test_command] - pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) - pjlink.send_queue = [] - pjlink.priority_queue = [] + # WHEN: send_command called with same command in priority queue + self.pjlink.send_command(cmd='CLSS', priority=True) - # WHEN: _send_command called with no data and queue's emtpy - # Patch here since pjlink does not have socket_timer until after instantiation - with patch.object(pjlink, 'socket_timer') as mock_timer: - pjlink._send_command(data=None, utf8=False) + # THEN: + mock_log.debug.assert_has_calls(log_debug_calls) + mock_log.warning.assert_has_calls(log_warning_calls) + mock_log.error.assert_has_calls(log_error_calls) + assert (not self.pjlink.send_queue), 'Send queue should be empty' + assert (self.pjlink.priority_queue == [test_command]), 'Priority queue should have one entry' + assert (not mock_reset.called), 'reset_information() should not have been called' + assert mock_send_command.called, '_underscore_send_command() should have been called' - # THEN: - mock_log.debug.assert_has_calls(log_debug_calls) - assert mock_timer.called is False + @patch.object(openlp.core.projectors.pjlink.PJLink, 'state') + @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information') + @patch.object(openlp.core.projectors.pjlink.PJLink, '_send_command') + @patch.object(openlp.core.projectors.pjlink, 'log') + def test_send_command_invalid_command_empty_queues(self, mock_log, mock_send_command, mock_reset, mock_state): + """ + Test send_command with invalid command + """ + # GIVEN: Test object + log_error_calls = [call('({ip}) send_command(): Invalid command requested - ' + 'ignoring.'.format(ip=self.pjlink.name))] + log_warning_calls = [] + log_debug_calls = [] + mock_state.return_value = S_CONNECTED + self.pjlink.send_queue = [] + self.pjlink.priority_queue = [] + + # WHEN: send_command with invalid command + self.pjlink.send_command(cmd='DONTCARE') + + # THEN: + mock_log.debug.assert_has_calls(log_debug_calls) + mock_log.warning.assert_has_calls(log_warning_calls) + mock_log.error.assert_has_calls(log_error_calls) + assert (not self.pjlink.send_queue), 'Send queue should be empty' + assert (not self.pjlink.priority_queue), 'Priority queue should be empty' + assert (not mock_reset.called), 'reset_information() should not have been called' + assert (not mock_send_command.called), '_underscore_send_command() should not have been called' + + @patch.object(openlp.core.projectors.pjlink.PJLink, 'state') + @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information') + @patch.object(openlp.core.projectors.pjlink.PJLink, '_send_command') + @patch.object(openlp.core.projectors.pjlink, 'log') + def test_send_command_invalid_command_normal_queue(self, mock_log, mock_send_command, mock_reset, mock_state): + """ + Test _send_command with invalid command for normal queue + """ + # GIVEN: Test object + test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX, + clss=self.pjlink.pjlink_class, + suff=PJLINK_SUFFIX) + log_error_calls = [call('({ip}) send_command(): Invalid command requested - ' + 'ignoring.'.format(ip=self.pjlink.name))] + log_warning_calls = [] + log_debug_calls = [] + mock_state.return_value = S_CONNECTED + self.pjlink.send_queue = [test_command] + self.pjlink.priority_queue = [] + + # WHEN: send_command with invalid command + self.pjlink.send_command(cmd='DONTCARE') + + # THEN: + mock_log.debug.assert_has_calls(log_debug_calls) + mock_log.warning.assert_has_calls(log_warning_calls) + mock_log.error.assert_has_calls(log_error_calls) + assert self.pjlink.send_queue, 'Send queue should have one entry' + assert (not self.pjlink.priority_queue), 'Priority queue should be empty' + assert (not mock_reset.called), 'reset_information() should not have been called' + assert mock_send_command.called, '_underscore_send_command() should have been called' + + @patch.object(openlp.core.projectors.pjlink.PJLink, 'state') + @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information') + @patch.object(openlp.core.projectors.pjlink.PJLink, '_send_command') + @patch.object(openlp.core.projectors.pjlink, 'log') + def test_send_command_invalid_command_priority_queue(self, mock_log, mock_send_command, mock_reset, mock_state): + """ + Test _send_command with invalid command for priority queue + """ + # GIVEN: Test object + test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX, + clss=self.pjlink.pjlink_class, + suff=PJLINK_SUFFIX) + log_error_calls = [call('({ip}) send_command(): Invalid command requested - ' + 'ignoring.'.format(ip=self.pjlink.name))] + log_warning_calls = [] + log_debug_calls = [] + mock_state.return_value = S_CONNECTED + self.pjlink.send_queue = [] + self.pjlink.priority_queue = [test_command] + + # WHEN: send_command with invalid command + self.pjlink.send_command(cmd='DONTCARE', priority=True) + + # THEN: + mock_log.debug.assert_has_calls(log_debug_calls) + mock_log.warning.assert_has_calls(log_warning_calls) + mock_log.error.assert_has_calls(log_error_calls) + assert (not self.pjlink.send_queue), 'Send queue should be empty' + assert self.pjlink.priority_queue, 'Priority queue should have one entry' + assert (not mock_reset.called), 'reset_information() should not have been called' + assert mock_send_command.called, '_underscore_send_command() should have been called' + + @patch.object(openlp.core.projectors.pjlink.PJLink, 'state') + @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information') + @patch.object(openlp.core.projectors.pjlink.PJLink, '_send_command') + @patch.object(openlp.core.projectors.pjlink, 'log') + def test_send_command_not_connected(self, mock_log, mock_send_command, mock_reset, mock_state): + """ + Test send_command when not connected + """ + # GIVEN: Test object + log_error_calls = [] + log_warning_calls = [call('({ip}) send_command(): Not connected - returning'.format(ip=self.pjlink.name))] + log_debug_calls = [] + mock_state.return_value = S_NOT_CONNECTED + self.pjlink.send_queue = [] + self.pjlink.priority_queue = [] + + # WHEN: send_command called when not connected + self.pjlink.send_command(cmd=None) + + # THEN: + mock_log.debug.assert_has_calls(log_debug_calls) + mock_log.warning.assert_has_calls(log_warning_calls) + mock_log.error.assert_has_calls(log_error_calls) + assert (not self.pjlink.send_queue), 'Send queue should be empty' + assert (not self.pjlink.priority_queue), 'Priority queue should be empty' + assert mock_reset.called, 'reset_information() should have been called' + assert (not mock_send_command.called), '_underscore_send_command() should not have been called' diff --git a/tests/openlp_core/projectors/test_projector_pjlink_cmd_routing.py b/tests/openlp_core/projectors/test_projector_pjlink_cmd_routing.py index 32421a872..26627cda8 100644 --- a/tests/openlp_core/projectors/test_projector_pjlink_cmd_routing.py +++ b/tests/openlp_core/projectors/test_projector_pjlink_cmd_routing.py @@ -59,15 +59,16 @@ class TestPJLinkRouting(TestCase): # GIVEN: Test object self.pjlink.pjlink_functions = MagicMock() log_warning_text = [call('({ip}) get_data(): Invalid packet - ' - 'unknown command "UNKN"'.format(ip=self.pjlink.name))] + 'unknown command "UNKN"'.format(ip=self.pjlink.name)), + call('({ip}) _send_command(): Nothing to send - ' + 'returning'.format(ip=self.pjlink.name))] log_debug_text = [call('(___TEST_ONE___) get_data(buffer="%1UNKN=Huh?"'), call('(___TEST_ONE___) get_data(): Checking new data "%1UNKN=Huh?"'), call('(___TEST_ONE___) get_data() header="%1UNKN" data="Huh?"'), call('(___TEST_ONE___) get_data() version="1" cmd="UNKN"'), call('(___TEST_ONE___) Cleaning buffer - msg = "get_data(): ' 'Invalid packet - unknown command "UNKN""'), - call('(___TEST_ONE___) Finished cleaning buffer - 0 bytes dropped'), - call('(___TEST_ONE___) _send_command(): Nothing to send - returning')] + call('(___TEST_ONE___) Finished cleaning buffer - 0 bytes dropped')] # WHEN: get_data called with an unknown command self.pjlink.get_data(buff='{prefix}1UNKN=Huh?'.format(prefix=PJLINK_PREFIX))