diff --git a/openlp/core/lib/projector/pjlink.py b/openlp/core/lib/projector/pjlink.py index 14d91f4ba..071356a0e 100644 --- a/openlp/core/lib/projector/pjlink.py +++ b/openlp/core/lib/projector/pjlink.py @@ -57,7 +57,7 @@ from openlp.core.lib.projector.constants import CONNECTION_ERRORS, CR, ERROR_MSG E_AUTHENTICATION, E_CONNECTION_REFUSED, E_GENERAL, E_INVALID_DATA, E_NETWORK, E_NOT_CONNECTED, E_OK, \ E_PARAMETER, E_PROJECTOR, E_SOCKET_TIMEOUT, E_UNAVAILABLE, E_UNDEFINED, PJLINK_ERRORS, PJLINK_ERST_DATA, \ PJLINK_ERST_STATUS, PJLINK_MAX_PACKET, PJLINK_PORT, PJLINK_POWR_STATUS, PJLINK_VALID_CMD, \ - STATUS_STRING, S_CONNECTED, S_CONNECTING, S_NETWORK_RECEIVED, S_NETWORK_SENDING, \ + STATUS_STRING, S_CONNECTED, S_CONNECTING, S_INFO, S_NETWORK_RECEIVED, S_NETWORK_SENDING, \ S_NOT_CONNECTED, S_OFF, S_OK, S_ON, S_STATUS # Shortcuts @@ -159,7 +159,7 @@ class PJLinkCommands(object): # A command returned successfully, no further processing needed return elif _cmd not in self.pjlink_functions: - log.warn("({ip}) Unable to process command='{cmd}' (Future option)".format(ip=self.ip, cmd=cmd)) + log.warning("({ip}) Unable to process command='{cmd}' (Future option)".format(ip=self.ip, cmd=cmd)) return elif _data in PJLINK_ERRORS: # Oops - projector error @@ -231,7 +231,7 @@ class PJLinkCommands(object): # : Received: '%1CLSS=Class 1' (Optoma) # : Received: '%1CLSS=Version1' (BenQ) if len(data) > 1: - log.warn("({ip}) Non-standard CLSS reply: '{data}'".format(ip=self.ip, data=data)) + log.warning("({ip}) Non-standard CLSS reply: '{data}'".format(ip=self.ip, data=data)) # Due to stupid projectors not following standards (Optoma, BenQ comes to mind), # AND the different responses that can be received, the semi-permanent way to # fix the class reply is to just remove all non-digit characters. @@ -261,15 +261,15 @@ class PJLinkCommands(object): """ if len(data) != PJLINK_ERST_DATA['DATA_LENGTH']: count = PJLINK_ERST_DATA['DATA_LENGTH'] - log.warn("{ip}) Invalid error status response '{data}': length != {count}".format(ip=self.ip, - data=data, - count=count)) + log.warning("{ip}) Invalid error status response '{data}': length != {count}".format(ip=self.ip, + data=data, + count=count)) return try: datacheck = int(data) except ValueError: # Bad data - ignore - log.warn("({ip}) Invalid error status response '{data}'".format(ip=self.ip, data=data)) + log.warning("({ip}) Invalid error status response '{data}'".format(ip=self.ip, data=data)) return if datacheck == 0: self.projector_errors = None @@ -429,9 +429,9 @@ class PJLinkCommands(object): if self.model_filter is None: self.model_filter = data else: - log.warn("({ip}) Filter model already set".format(ip=self.ip)) - log.warn("({ip}) Saved model: '{old}'".format(ip=self.ip, old=self.model_filter)) - log.warn("({ip}) New model: '{new}'".format(ip=self.ip, new=data)) + log.warning("({ip}) Filter model already set".format(ip=self.ip)) + log.warning("({ip}) Saved model: '{old}'".format(ip=self.ip, old=self.model_filter)) + log.warning("({ip}) New model: '{new}'".format(ip=self.ip, new=data)) def process_rlmp(self, data): """ @@ -440,9 +440,9 @@ class PJLinkCommands(object): if self.model_lamp is None: self.model_lamp = data else: - log.warn("({ip}) Lamp model already set".format(ip=self.ip)) - log.warn("({ip}) Saved lamp: '{old}'".format(ip=self.ip, old=self.model_lamp)) - log.warn("({ip}) New lamp: '{new}'".format(ip=self.ip, new=data)) + log.warning("({ip}) Lamp model already set".format(ip=self.ip)) + log.warning("({ip}) Saved lamp: '{old}'".format(ip=self.ip, old=self.model_lamp)) + log.warning("({ip}) New lamp: '{new}'".format(ip=self.ip, new=data)) def process_snum(self, data): """ @@ -457,27 +457,32 @@ class PJLinkCommands(object): else: # Compare serial numbers and see if we got the same projector if self.serial_no != data: - log.warn("({ip}) Projector serial number does not match saved serial number".format(ip=self.ip)) - log.warn("({ip}) Saved: '{old}'".format(ip=self.ip, old=self.serial_no)) - log.warn("({ip}) Received: '{new}'".format(ip=self.ip, new=data)) - log.warn("({ip}) NOT saving serial number".format(ip=self.ip)) + log.warning("({ip}) Projector serial number does not match saved serial number".format(ip=self.ip)) + log.warning("({ip}) Saved: '{old}'".format(ip=self.ip, old=self.serial_no)) + log.warning("({ip}) Received: '{new}'".format(ip=self.ip, new=data)) + log.warning("({ip}) NOT saving serial number".format(ip=self.ip)) self.serial_no_received = data def process_sver(self, data): """ Software version of projector """ - if self.sw_version is None: + if len(data) > 32: + # Defined in specs max version is 32 characters + log.warning("Invalid software version - too long") + return + elif self.sw_version is None: log.debug("({ip}) Setting projector software version to '{data}'".format(ip=self.ip, data=data)) self.sw_version = data self.db_update = True else: # Compare software version and see if we got the same projector if self.serial_no != data: - log.warn("({ip}) Projector software version does not match saved software version".format(ip=self.ip)) - log.warn("({ip}) Saved: '{old}'".format(ip=self.ip, old=self.sw_version)) - log.warn("({ip}) Received: '{new}'".format(ip=self.ip, new=data)) - log.warn("({ip}) NOT saving serial number".format(ip=self.ip)) + log.warning("({ip}) Projector software version does not match saved " + "software version".format(ip=self.ip)) + log.warning("({ip}) Saved: '{old}'".format(ip=self.ip, old=self.sw_version)) + log.warning("({ip}) Received: '{new}'".format(ip=self.ip, new=data)) + log.warning("({ip}) Saving new serial number as sw_version_received".format(ip=self.ip)) self.sw_version_received = data @@ -605,7 +610,7 @@ class PJLink(PJLinkCommands, QtNetwork.QTcpSocket): Normally called by timer(). """ if self.state() != self.ConnectedState: - log.warn("({ip}) poll_loop(): Not connected - returning".format(ip=self.ip)) + log.warning("({ip}) poll_loop(): Not connected - returning".format(ip=self.ip)) return log.debug('({ip}) Updating projector status'.format(ip=self.ip)) # Reset timer in case we were called from a set command @@ -649,7 +654,9 @@ class PJLink(PJLinkCommands, QtNetwork.QTcpSocket): :param status: Status/Error code :returns: (Status/Error code, String) """ - if status in ERROR_STRING: + if not isinstance(status, int): + return -1, 'Invalid status code' + elif status in ERROR_STRING: return ERROR_STRING[status], ERROR_MSG[status] elif status in STATUS_STRING: return STATUS_STRING[status], ERROR_MSG[status] @@ -674,7 +681,7 @@ class PJLink(PJLinkCommands, QtNetwork.QTcpSocket): elif status >= S_NOT_CONNECTED and status < S_STATUS: self.status_connect = status self.projector_status = S_NOT_CONNECTED - elif status < S_NETWORK_SENDING: + elif status <= S_INFO: self.status_connect = S_CONNECTED self.projector_status = status (status_code, status_message) = self._get_status(self.status_connect) @@ -803,7 +810,8 @@ class PJLink(PJLinkCommands, QtNetwork.QTcpSocket): log.debug('({ip}) get_data(): Not connected - returning'.format(ip=self.ip)) self.send_busy = False return - read = self.readLine(self.max_size) + # Although we have a packet length limit, go ahead and use a larger buffer + read = self.readLine(1024) log.debug("({ip}) get_data(): '{buff}'".format(ip=self.ip, buff=read)) if read == -1: # No data available @@ -816,6 +824,8 @@ class PJLink(PJLinkCommands, QtNetwork.QTcpSocket): data = data_in.strip() if (len(data) < 7) or (not data.startswith(PJLINK_PREFIX)): return self._trash_buffer(msg='get_data(): Invalid packet - length or prefix') + elif len(data) > self.max_size: + return self._trash_buffer(msg='get_data(): Invalid packet - too long') elif '=' not in data: return self._trash_buffer(msg='get_data(): Invalid packet does not have equal') log.debug('({ip}) get_data(): Checking new data "{data}"'.format(ip=self.ip, data=data)) @@ -830,8 +840,8 @@ class PJLink(PJLinkCommands, QtNetwork.QTcpSocket): log.warning('({ip}) get_data(): Invalid packet - unknown command "{data}"'.format(ip=self.ip, data=cmd)) return self._trash_buffer(msg='get_data(): Unknown command "{data}"'.format(data=cmd)) if int(self.pjlink_class) < int(version): - log.warn('({ip}) get_data(): Projector returned class reply higher ' - 'than projector stated class'.format(ip=self.ip)) + log.warning('({ip}) get_data(): Projector returned class reply higher ' + 'than projector stated class'.format(ip=self.ip)) return self.process_command(cmd, data) @QtCore.pyqtSlot(QtNetwork.QAbstractSocket.SocketError) @@ -993,6 +1003,13 @@ class PJLink(PJLinkCommands, QtNetwork.QTcpSocket): self.reset_information() self.projectorUpdateIcons.emit() + def get_av_mute_status(self): + """ + Send command to retrieve shutter status. + """ + log.debug('({ip}) Sending AVMT command'.format(ip=self.ip)) + return self.send_command(cmd='AVMT') + def get_available_inputs(self): """ Send command to retrieve available source inputs. @@ -1056,13 +1073,6 @@ class PJLink(PJLinkCommands, QtNetwork.QTcpSocket): log.debug('({ip}) Sending POWR command'.format(ip=self.ip)) return self.send_command(cmd='POWR') - def get_shutter_status(self): - """ - Send command to retrieve shutter status. - """ - log.debug('({ip}) Sending AVMT command'.format(ip=self.ip)) - return self.send_command(cmd='AVMT') - def set_input_source(self, src=None): """ Verify input source available as listed in 'INST' command, diff --git a/tests/functional/openlp_core_lib/test_projector_pjlink_cmd_routing.py b/tests/functional/openlp_core_lib/test_projector_pjlink_cmd_routing.py index 0a962146a..4c45e3e58 100644 --- a/tests/functional/openlp_core_lib/test_projector_pjlink_cmd_routing.py +++ b/tests/functional/openlp_core_lib/test_projector_pjlink_cmd_routing.py @@ -179,7 +179,7 @@ class TestPJLinkRouting(TestCase): # THEN: Error should be logged and no command called self.assertFalse(mock_functions.called, 'Should not have gotten to the end of the method') - mock_log.warn.assert_called_once_with(log_text) + mock_log.warning.assert_called_once_with(log_text) @patch.object(pjlink_test, 'pjlink_functions') @patch.object(openlp.core.lib.projector.pjlink, 'log') diff --git a/tests/functional/openlp_core_lib/test_projector_pjlink_commands.py b/tests/functional/openlp_core_lib/test_projector_pjlink_commands.py index fec890758..8b774aa8a 100644 --- a/tests/functional/openlp_core_lib/test_projector_pjlink_commands.py +++ b/tests/functional/openlp_core_lib/test_projector_pjlink_commands.py @@ -23,12 +23,14 @@ Package to test the openlp.core.lib.projector.pjlink commands package. """ from unittest import TestCase -from unittest.mock import patch, MagicMock +from unittest.mock import patch import openlp.core.lib.projector.pjlink from openlp.core.lib.projector.pjlink import PJLink from openlp.core.lib.projector.constants import ERROR_STRING, PJLINK_ERST_DATA, PJLINK_ERST_STATUS, \ - PJLINK_POWR_STATUS, E_WARN, E_ERROR, S_OFF, S_STANDBY, S_ON + PJLINK_POWR_STATUS, \ + E_ERROR, E_NOT_CONNECTED, E_SOCKET_ADDRESS_NOT_AVAILABLE, E_UNKNOWN_SOCKET_ERROR, E_WARN, \ + S_CONNECTED, S_OFF, S_ON, S_NOT_CONNECTED, S_CONNECTING, S_STANDBY from tests.resources.projector.data import TEST_PIN @@ -45,48 +47,408 @@ class TestPJLinkCommands(TestCase): """ Tests for the PJLink module """ - def test_projector_reset_information(self): + @patch.object(pjlink_test, 'changeStatus') + @patch.object(openlp.core.lib.projector.pjlink, 'log') + def test_projector_change_status_connection_error(self, mock_log, mock_change_status): """ - Test reset_information() resets all information and stops timers + Test change_status with connection error """ - # GIVEN: Test object and test data + # GIVEN: Test object pjlink = pjlink_test - pjlink.power = S_ON - pjlink.pjlink_name = 'OPENLPTEST' - pjlink.manufacturer = 'PJLINK' - pjlink.model = '1' - pjlink.shutter = True - pjlink.mute = True - pjlink.lamp = True - pjlink.fan = True - pjlink.source_available = True - pjlink.other_info = 'ANOTHER TEST' - pjlink.send_queue = True - pjlink.send_busy = True - pjlink.timer = MagicMock() - pjlink.socket_timer = MagicMock() + pjlink.projector_status = 0 + pjlink.status_connect = 0 + test_code = E_UNKNOWN_SOCKET_ERROR + mock_change_status.reset_mock() + mock_log.reset_mock() - # WHEN: reset_information() is called - with patch.object(pjlink.timer, 'stop') as mock_timer: - with patch.object(pjlink.socket_timer, 'stop') as mock_socket_timer: - pjlink.reset_information() + # WHEN: change_status called with unknown socket error + pjlink.change_status(status=test_code, msg=None) - # THEN: All information should be reset and timers stopped - self.assertEqual(pjlink.power, S_OFF, 'Projector power should be OFF') - self.assertIsNone(pjlink.pjlink_name, 'Projector pjlink_name should be None') - self.assertIsNone(pjlink.manufacturer, 'Projector manufacturer should be None') - self.assertIsNone(pjlink.model, 'Projector model should be None') - self.assertIsNone(pjlink.shutter, 'Projector shutter should be None') - self.assertIsNone(pjlink.mute, 'Projector shuttter should be None') - self.assertIsNone(pjlink.lamp, 'Projector lamp should be None') - self.assertIsNone(pjlink.fan, 'Projector fan should be None') - self.assertIsNone(pjlink.source_available, 'Projector source_available should be None') - self.assertIsNone(pjlink.source, 'Projector source should be None') - self.assertIsNone(pjlink.other_info, 'Projector other_info should be None') - self.assertEqual(pjlink.send_queue, [], 'Projector send_queue should be an empty list') - self.assertFalse(pjlink.send_busy, 'Projector send_busy should be False') - self.assertTrue(mock_timer.called, 'Projector timer.stop() should have been called') - self.assertTrue(mock_socket_timer.called, 'Projector socket_timer.stop() should have been called') + # THEN: Proper settings should change and signals sent + self.assertEqual(pjlink.projector_status, E_NOT_CONNECTED, 'Projector status should be NOT CONNECTED') + self.assertEqual(pjlink.status_connect, E_NOT_CONNECTED, 'Status connect should be NOT CONNECTED') + mock_change_status.emit.assert_called_once_with(pjlink.ip, E_UNKNOWN_SOCKET_ERROR, + 'An unidentified error occurred') + self.assertEqual(mock_log.debug.call_count, 3, 'Debug log should have been called 3 times') + + @patch.object(pjlink_test, 'changeStatus') + @patch.object(openlp.core.lib.projector.pjlink, 'log') + def test_projector_change_status_connection_status_connecting(self, mock_log, mock_change_status): + """ + Test change_status with connection status + """ + # GIVEN: Test object + pjlink = pjlink_test + pjlink.projector_status = 0 + pjlink.status_connect = 0 + test_code = S_CONNECTING + mock_change_status.reset_mock() + mock_log.reset_mock() + + # WHEN: change_status called with unknown socket error + pjlink.change_status(status=test_code, msg=None) + + # THEN: Proper settings should change and signals sent + self.assertEqual(pjlink.projector_status, S_NOT_CONNECTED, 'Projector status should be NOT CONNECTED') + self.assertEqual(pjlink.status_connect, S_CONNECTING, 'Status connect should be CONNECTING') + mock_change_status.emit.assert_called_once_with(pjlink.ip, S_CONNECTING, 'Connecting') + self.assertEqual(mock_log.debug.call_count, 3, 'Debug log should have been called 3 times') + + @patch.object(pjlink_test, 'changeStatus') + @patch.object(openlp.core.lib.projector.pjlink, 'log') + def test_projector_change_status_connection_status_connected(self, mock_log, mock_change_status): + """ + Test change_status with connection status + """ + # GIVEN: Test object + pjlink = pjlink_test + pjlink.projector_status = 0 + pjlink.status_connect = 0 + test_code = S_ON + mock_change_status.reset_mock() + mock_log.reset_mock() + + # WHEN: change_status called with unknown socket error + pjlink.change_status(status=test_code, msg=None) + + # THEN: Proper settings should change and signals sent + self.assertEqual(pjlink.projector_status, S_ON, 'Projector status should be ON') + self.assertEqual(pjlink.status_connect, S_CONNECTED, 'Status connect should be CONNECTED') + mock_change_status.emit.assert_called_once_with(pjlink.ip, S_ON, 'Power is on') + self.assertEqual(mock_log.debug.call_count, 3, 'Debug log should have been called 3 times') + + @patch.object(pjlink_test, 'changeStatus') + @patch.object(openlp.core.lib.projector.pjlink, 'log') + def test_projector_change_status_connection_status_with_message(self, mock_log, mock_change_status): + """ + Test change_status with connection status + """ + # GIVEN: Test object + pjlink = pjlink_test + pjlink.projector_status = 0 + pjlink.status_connect = 0 + test_message = 'Different Status Message than default' + test_code = S_ON + mock_change_status.reset_mock() + mock_log.reset_mock() + + # WHEN: change_status called with unknown socket error + pjlink.change_status(status=test_code, msg=test_message) + + # THEN: Proper settings should change and signals sent + self.assertEqual(pjlink.projector_status, S_ON, 'Projector status should be ON') + self.assertEqual(pjlink.status_connect, S_CONNECTED, 'Status connect should be CONNECTED') + mock_change_status.emit.assert_called_once_with(pjlink.ip, S_ON, test_message) + self.assertEqual(mock_log.debug.call_count, 3, 'Debug log should have been called 3 times') + + @patch.object(pjlink_test, 'send_command') + @patch.object(openlp.core.lib.projector.pjlink, 'log') + def test_projector_get_av_mute_status(self, mock_log, mock_send_command): + """ + Test sending command to retrieve shutter/audio state + """ + # GIVEN: Test object + pjlink = pjlink_test + mock_log.reset_mock() + mock_send_command.reset_mock() + test_data = 'AVMT' + test_log = '(127.0.0.1) Sending AVMT command' + + # WHEN: get_av_mute_status is called + pjlink.get_av_mute_status() + + # THEN: log data and send_command should have been called + mock_log.debug.assert_called_once_with(test_log) + mock_send_command.assert_called_once_with(cmd=test_data) + + @patch.object(pjlink_test, 'send_command') + @patch.object(openlp.core.lib.projector.pjlink, 'log') + def test_projector_get_available_inputs(self, mock_log, mock_send_command): + """ + Test sending command to retrieve avaliable inputs + """ + # GIVEN: Test object + pjlink = pjlink_test + mock_log.reset_mock() + mock_send_command.reset_mock() + test_data = 'INST' + test_log = '(127.0.0.1) Sending INST command' + + # WHEN: get_available_inputs is called + pjlink.get_available_inputs() + + # THEN: log data and send_command should have been called + mock_log.debug.assert_called_once_with(test_log) + mock_send_command.assert_called_once_with(cmd=test_data) + + @patch.object(pjlink_test, 'send_command') + @patch.object(openlp.core.lib.projector.pjlink, 'log') + def test_projector_get_error_status(self, mock_log, mock_send_command): + """ + Test sending command to retrieve projector error status + """ + # GIVEN: Test object + pjlink = pjlink_test + mock_log.reset_mock() + mock_send_command.reset_mock() + test_data = 'ERST' + test_log = '(127.0.0.1) Sending ERST command' + + # WHEN: get_error_status is called + pjlink.get_error_status() + + # THEN: log data and send_command should have been called + mock_log.debug.assert_called_once_with(test_log) + mock_send_command.assert_called_once_with(cmd=test_data) + + @patch.object(pjlink_test, 'send_command') + @patch.object(openlp.core.lib.projector.pjlink, 'log') + def test_projector_get_input_source(self, mock_log, mock_send_command): + """ + Test sending command to retrieve current input + """ + # GIVEN: Test object + pjlink = pjlink_test + mock_log.reset_mock() + mock_send_command.reset_mock() + test_data = 'INPT' + test_log = '(127.0.0.1) Sending INPT command' + + # WHEN: get_input_source is called + pjlink.get_input_source() + + # THEN: log data and send_command should have been called + mock_log.debug.assert_called_once_with(test_log) + mock_send_command.assert_called_once_with(cmd=test_data) + + @patch.object(pjlink_test, 'send_command') + @patch.object(openlp.core.lib.projector.pjlink, 'log') + def test_projector_get_lamp_status(self, mock_log, mock_send_command): + """ + Test sending command to retrieve lamp(s) status + """ + # GIVEN: Test object + pjlink = pjlink_test + mock_log.reset_mock() + mock_send_command.reset_mock() + test_data = 'LAMP' + test_log = '(127.0.0.1) Sending LAMP command' + + # WHEN: get_lamp_status is called + pjlink.get_lamp_status() + + # THEN: log data and send_command should have been called + mock_log.debug.assert_called_once_with(test_log) + mock_send_command.assert_called_once_with(cmd=test_data) + + @patch.object(pjlink_test, 'send_command') + @patch.object(openlp.core.lib.projector.pjlink, 'log') + def test_projector_get_manufacturer(self, mock_log, mock_send_command): + """ + Test sending command to retrieve manufacturer name + """ + # GIVEN: Test object + pjlink = pjlink_test + mock_log.reset_mock() + mock_send_command.reset_mock() + test_data = 'INF1' + test_log = '(127.0.0.1) Sending INF1 command' + + # WHEN: get_manufacturer is called + pjlink.get_manufacturer() + + # THEN: log data and send_command should have been called + mock_log.debug.assert_called_once_with(test_log) + mock_send_command.assert_called_once_with(cmd=test_data) + + @patch.object(pjlink_test, 'send_command') + @patch.object(openlp.core.lib.projector.pjlink, 'log') + def test_projector_get_model(self, mock_log, mock_send_command): + """ + Test sending command to get model information + """ + # GIVEN: Test object + pjlink = pjlink_test + mock_log.reset_mock() + mock_send_command.reset_mock() + test_data = 'INF2' + test_log = '(127.0.0.1) Sending INF2 command' + + # WHEN: get_model is called + pjlink.get_model() + + # THEN: log data and send_command should have been called + mock_log.debug.assert_called_once_with(test_log) + mock_send_command.assert_called_once_with(cmd=test_data) + + @patch.object(pjlink_test, 'send_command') + @patch.object(openlp.core.lib.projector.pjlink, 'log') + def test_projector_get_name(self, mock_log, mock_send_command): + """ + Test sending command to get user-assigned name + """ + # GIVEN: Test object + pjlink = pjlink_test + mock_log.reset_mock() + mock_send_command.reset_mock() + test_data = 'NAME' + test_log = '(127.0.0.1) Sending NAME command' + + # WHEN: get_name is called + pjlink.get_name() + + # THEN: log data and send_command should have been called + mock_log.debug.assert_called_once_with(test_log) + mock_send_command.assert_called_once_with(cmd=test_data) + + @patch.object(pjlink_test, 'send_command') + @patch.object(openlp.core.lib.projector.pjlink, 'log') + def test_projector_get_other_info(self, mock_log, mock_send_command): + """ + Test sending command to retrieve other information + """ + # GIVEN: Test object + pjlink = pjlink_test + mock_log.reset_mock() + mock_send_command.reset_mock() + test_data = 'INFO' + test_log = '(127.0.0.1) Sending INFO command' + + # WHEN: get_other_info is called + pjlink.get_other_info() + + # THEN: log data and send_command should have been called + mock_log.debug.assert_called_once_with(test_log) + mock_send_command.assert_called_once_with(cmd=test_data) + + @patch.object(pjlink_test, 'send_command') + @patch.object(openlp.core.lib.projector.pjlink, 'log') + def test_projector_get_power_status(self, mock_log, mock_send_command): + """ + Test sending command to retrieve current power state + """ + # GIVEN: Test object + pjlink = pjlink_test + mock_log.reset_mock() + mock_send_command.reset_mock() + test_data = 'POWR' + test_log = '(127.0.0.1) Sending POWR command' + + # WHEN: get_power_status called + pjlink.get_power_status() + + # THEN: log data and send_command should have been called + mock_log.debug.assert_called_once_with(test_log) + mock_send_command.assert_called_once_with(cmd=test_data) + + def test_projector_get_status_error(self): + """ + Test to check returned information for error code + """ + # GIVEN: Test object + pjlink = pjlink_test + test_string = 'E_SOCKET_ADDRESS_NOT_AVAILABLE' + test_message = 'The address specified to socket.bind() does not belong to the host' + + # WHEN: get_status called + string, message = pjlink._get_status(status=E_SOCKET_ADDRESS_NOT_AVAILABLE) + + # THEN: Proper strings should have been returned + self.assertEqual(string, test_string, 'Code as string should have been returned') + self.assertEqual(message, test_message, 'Description of code should have been returned') + + def test_projector_get_status_invalid(self): + """ + Test to check returned information for error code + """ + # GIVEN: Test object + pjlink = pjlink_test + test_string = 'Test string since get_status will only work with int' + test_message = 'Invalid status code' + + # WHEN: get_status called + string, message = pjlink._get_status(status=test_string) + + # THEN: Proper strings should have been returned + self.assertEqual(string, -1, 'Should have returned -1 as a bad status check') + self.assertEqual(message, test_message, 'Error message should have been returned') + + def test_projector_get_status_status(self): + """ + Test to check returned information for status codes + """ + # GIVEN: Test object + pjlink = pjlink_test + test_string = 'S_NOT_CONNECTED' + test_message = 'Not connected' + + # WHEN: get_status called + string, message = pjlink._get_status(status=S_NOT_CONNECTED) + + # THEN: Proper strings should have been returned + self.assertEqual(string, test_string, 'Code as string should have been returned') + self.assertEqual(message, test_message, 'Description of code should have been returned') + + def test_projector_get_status_unknown(self): + """ + Test to check returned information for unknown code + """ + # GIVEN: Test object + pjlink = pjlink_test + test_string = 999999 + test_message = 'Unknown status' + + # WHEN: get_status called + string, message = pjlink._get_status(status=test_string) + + # THEN: Proper strings should have been returned + self.assertEqual(string, test_string, 'Received code should have been returned') + self.assertEqual(message, test_message, 'Unknown status string should have been returned') + + def test_projector_process_inf1(self): + """ + Test saving INF1 data (manufacturer) + """ + # GIVEN: Test object + pjlink = pjlink_test + pjlink.manufacturer = None + test_data = 'TEst INformation MultiCase' + + # WHEN: process_inf called with test data + pjlink.process_inf1(data=test_data) + + # THEN: Data should be saved + self.assertEqual(pjlink.manufacturer, test_data, 'Test data should have been saved') + + def test_projector_process_inf2(self): + """ + Test saving INF2 data (model) + """ + # GIVEN: Test object + pjlink = pjlink_test + pjlink.model = None + test_data = 'TEst moDEl MultiCase' + + # WHEN: process_inf called with test data + pjlink.process_inf2(data=test_data) + + # THEN: Data should be saved + self.assertEqual(pjlink.model, test_data, 'Test data should have been saved') + + def test_projector_process_info(self): + """ + Test saving INFO data (other information) + """ + # GIVEN: Test object + pjlink = pjlink_test + pjlink.other_info = None + test_data = 'TEst ExtrANEous MultiCase INformatoin that MFGR might Set' + + # WHEN: process_inf called with test data + pjlink.process_info(data=test_data) + + # THEN: Data should be saved + self.assertEqual(pjlink.other_info, test_data, 'Test data should have been saved') @patch.object(pjlink_test, 'projectorUpdateIcons') def test_projector_process_avmt_bad_data(self, mock_UpdateIcons): @@ -245,12 +607,12 @@ class TestPJLinkCommands(TestCase): # WHEN: Process invalid reply pjlink.process_clss('Z') - log_warn_text = "(127.0.0.1) NAN clss version reply 'Z' - defaulting to class '1'" + log_text = "(127.0.0.1) NAN clss version reply 'Z' - defaulting to class '1'" # THEN: Projector class should be set with default value self.assertEqual(pjlink.pjlink_class, '1', 'Non-standard class reply should have set class=1') - mock_log.error.assert_called_once_with(log_warn_text) + mock_log.error.assert_called_once_with(log_text) @patch.object(openlp.core.lib.projector.pjlink, 'log') def test_projector_process_clss_invalid_no_version(self, mock_log): @@ -262,12 +624,12 @@ class TestPJLinkCommands(TestCase): # WHEN: Process invalid reply pjlink.process_clss('Invalid') - log_warn_text = "(127.0.0.1) No numbers found in class version reply 'Invalid' - defaulting to class '1'" + log_text = "(127.0.0.1) No numbers found in class version reply 'Invalid' - defaulting to class '1'" # THEN: Projector class should be set with default value self.assertEqual(pjlink.pjlink_class, '1', 'Non-standard class reply should have set class=1') - mock_log.error.assert_called_once_with(log_warn_text) + mock_log.error.assert_called_once_with(log_text) def test_projector_process_erst_all_ok(self): """ @@ -292,15 +654,15 @@ class TestPJLinkCommands(TestCase): # GIVEN: Test object pjlink = pjlink_test pjlink.projector_errors = None - log_warn_text = "127.0.0.1) Invalid error status response '11111111': length != 6" + log_text = "127.0.0.1) Invalid error status response '11111111': length != 6" # WHEN: process_erst called with invalid data (too many values pjlink.process_erst('11111111') # THEN: pjlink.projector_errors should be empty and warning logged self.assertIsNone(pjlink.projector_errors, 'There should be no errors') - self.assertTrue(mock_log.warn.called, 'Warning should have been logged') - mock_log.warn.assert_called_once_with(log_warn_text) + self.assertTrue(mock_log.warning.called, 'Warning should have been logged') + mock_log.warning.assert_called_once_with(log_text) @patch.object(openlp.core.lib.projector.pjlink, 'log') def test_projector_process_erst_data_invalid_nan(self, mock_log): @@ -310,15 +672,15 @@ class TestPJLinkCommands(TestCase): # GIVEN: Test object pjlink = pjlink_test pjlink.projector_errors = None - log_warn_text = "(127.0.0.1) Invalid error status response '1111Z1'" + log_text = "(127.0.0.1) Invalid error status response '1111Z1'" # WHEN: process_erst called with invalid data (too many values pjlink.process_erst('1111Z1') # THEN: pjlink.projector_errors should be empty and warning logged self.assertIsNone(pjlink.projector_errors, 'There should be no errors') - self.assertTrue(mock_log.warn.called, 'Warning should have been logged') - mock_log.warn.assert_called_once_with(log_warn_text) + self.assertTrue(mock_log.warning.called, 'Warning should have been logged') + mock_log.warning.assert_called_once_with(log_text) def test_projector_process_erst_all_warn(self): """ @@ -399,33 +761,67 @@ class TestPJLinkCommands(TestCase): # THEN: Input selected should reflect current input self.assertEqual(pjlink.source, '1', 'Input source should be set to "1"') - @patch.object(pjlink_test, 'projectorReceivedData') - def test_projector_process_lamp_single(self, mock_projectorReceivedData): + @patch.object(pjlink_test, 'projectorUpdateIcons') + @patch.object(openlp.core.lib.projector.pjlink, 'log') + def test_projector_process_inst(self, mock_log, mock_UpdateIcons): """ - Test status lamp on/off and hours + Test saving video source available information """ # GIVEN: Test object pjlink = pjlink_test + pjlink.source_available = [] + test_data = '21 10 30 31 11 20' + test_saved = ['10', '11', '20', '21', '30', '31'] + log_data = '(127.0.0.1) Setting projector sources_available to ' \ + '"[\'10\', \'11\', \'20\', \'21\', \'30\', \'31\']"' + mock_UpdateIcons.reset_mock() + mock_log.reset_mock() - # WHEN: Call process_command with lamp data - pjlink.process_command('LAMP', '22222 1') + # WHEN: process_inst called with test data + pjlink.process_inst(data=test_data) - # THEN: Lamp should have been set with status=ON and hours=22222 - self.assertEqual(pjlink.lamp[0]['On'], True, - 'Lamp power status should have been set to TRUE') - self.assertEqual(pjlink.lamp[0]['Hours'], 22222, - 'Lamp hours should have been set to 22222') + # THEN: Data should have been sorted and saved properly + self.assertEqual(pjlink.source_available, test_saved, "Sources should have been sorted and saved") + mock_log.debug.assert_called_once_with(log_data) + self.assertTrue(mock_UpdateIcons.emit.called, 'Update Icons should have been called') - @patch.object(pjlink_test, 'projectorReceivedData') - def test_projector_process_lamp_multiple(self, mock_projectorReceivedData): + @patch.object(openlp.core.lib.projector.pjlink, 'log') + def test_projector_process_lamp_invalid(self, mock_log): """ Test status multiple lamp on/off and hours """ # GIVEN: Test object pjlink = pjlink_test + pjlink.lamp = [{'Hours': 00000, 'On': True}, + {'Hours': 11111, 'On': False}] + log_data = '(127.0.0.1) process_lamp(): Invalid data "11111 1 22222 0 333A3 1"' + + # WHEN: Call process_command with invalid lamp data + pjlink.process_lamp('11111 1 22222 0 333A3 1') + + # THEN: lamps should not have changed + self.assertEqual(len(pjlink.lamp), 2, + 'Projector should have kept 2 lamps specified') + self.assertEqual(pjlink.lamp[0]['On'], True, + 'Lamp 1 power status should have been set to TRUE') + self.assertEqual(pjlink.lamp[0]['Hours'], 00000, + 'Lamp 1 hours should have been left at 00000') + self.assertEqual(pjlink.lamp[1]['On'], False, + 'Lamp 2 power status should have been set to FALSE') + self.assertEqual(pjlink.lamp[1]['Hours'], 11111, + 'Lamp 2 hours should have been left at 11111') + mock_log.warning.assert_called_once_with(log_data) + + def test_projector_process_lamp_multiple(self): + """ + Test status multiple lamp on/off and hours + """ + # GIVEN: Test object + pjlink = pjlink_test + pjlink.lamps = [] # WHEN: Call process_command with lamp data - pjlink.process_command('LAMP', '11111 1 22222 0 33333 1') + pjlink.process_lamp('11111 1 22222 0 33333 1') # THEN: Lamp should have been set with proper lamp status self.assertEqual(len(pjlink.lamp), 3, @@ -443,53 +839,112 @@ class TestPJLinkCommands(TestCase): self.assertEqual(pjlink.lamp[2]['Hours'], 33333, 'Lamp 3 hours should have been set to 33333') - @patch.object(pjlink_test, 'projectorReceivedData') + def test_projector_process_lamp_single(self): + """ + Test status lamp on/off and hours + """ + # GIVEN: Test object + pjlink = pjlink_test + pjlink.lamps = [] + + # WHEN: Call process_command with lamp data + pjlink.process_lamp('22222 1') + + # THEN: Lamp should have been set with status=ON and hours=22222 + self.assertEqual(pjlink.lamp[0]['On'], True, + 'Lamp power status should have been set to TRUE') + self.assertEqual(pjlink.lamp[0]['Hours'], 22222, + 'Lamp hours should have been set to 22222') + + @patch.object(openlp.core.lib.projector.pjlink, 'log') + def test_projector_process_name(self, mock_log): + """ + Test saving NAME data from projector + """ + # GIVEN: Test data + pjlink = pjlink_test + test_data = "Some Name the End-User Set IN Projector" + test_log = '(127.0.0.1) Setting projector PJLink name to "Some Name the End-User Set IN Projector"' + mock_log.reset_mock() + + # WHEN: process_name called with test data + pjlink.process_name(data=test_data) + + # THEN: name should be set and logged + self.assertEqual(pjlink.pjlink_name, test_data, 'Name test data should have been saved') + mock_log.debug.assert_called_once_with(test_log) + @patch.object(pjlink_test, 'projectorUpdateIcons') @patch.object(pjlink_test, 'send_command') @patch.object(pjlink_test, 'change_status') def test_projector_process_powr_on(self, mock_change_status, mock_send_command, - mock_UpdateIcons, - mock_ReceivedData): + mock_UpdateIcons): """ Test status power to ON """ # GIVEN: Test object and preset pjlink = pjlink_test pjlink.power = S_STANDBY + test_data = PJLINK_POWR_STATUS[S_ON] # WHEN: Call process_command with turn power on command - pjlink.process_command('POWR', PJLINK_POWR_STATUS[S_ON]) + pjlink.process_command(cmd='POWR', data=test_data) # THEN: Power should be set to ON self.assertEqual(pjlink.power, S_ON, 'Power should have been set to ON') mock_send_command.assert_called_once_with('INST') + mock_change_status.assert_called_once_with(PJLINK_POWR_STATUS[test_data]) self.assertEqual(mock_UpdateIcons.emit.called, True, 'projectorUpdateIcons should have been called') - @patch.object(pjlink_test, 'projectorReceivedData') + @patch.object(pjlink_test, 'projectorUpdateIcons') + @patch.object(pjlink_test, 'send_command') + @patch.object(pjlink_test, 'change_status') + def test_projector_process_powr_invalid(self, + mock_change_status, + mock_send_command, + mock_UpdateIcons): + """ + Test process_powr invalid call + """ + # GIVEN: Test object and preset + pjlink = pjlink_test + pjlink.power = S_STANDBY + test_data = '99' + + # WHEN: Call process_command with turn power on command + pjlink.process_command(cmd='POWR', data=test_data) + + # THEN: Power should be set to ON + self.assertEqual(pjlink.power, S_STANDBY, 'Power should not have changed') + self.assertFalse(mock_change_status.called, 'Change status should not have been called') + self.assertFalse(mock_send_command.called, 'send_command("INST") should not have been called') + self.assertFalse(mock_UpdateIcons.emit.called, 'projectorUpdateIcons should not have been called') + @patch.object(pjlink_test, 'projectorUpdateIcons') @patch.object(pjlink_test, 'send_command') @patch.object(pjlink_test, 'change_status') def test_projector_process_powr_off(self, mock_change_status, mock_send_command, - mock_UpdateIcons, - mock_ReceivedData): + mock_UpdateIcons): """ Test status power to STANDBY """ # GIVEN: Test object and preset pjlink = pjlink_test pjlink.power = S_ON + test_data = PJLINK_POWR_STATUS[S_STANDBY] # WHEN: Call process_command with turn power on command - pjlink.process_command('POWR', PJLINK_POWR_STATUS[S_STANDBY]) + pjlink.process_command(cmd='POWR', data=test_data) # THEN: Power should be set to STANDBY self.assertEqual(pjlink.power, S_STANDBY, 'Power should have been set to STANDBY') - self.assertEqual(mock_send_command.called, False, 'send_command should not have been called') self.assertEqual(mock_UpdateIcons.emit.called, True, 'projectorUpdateIcons should have been called') + mock_change_status.assert_called_once_with(PJLINK_POWR_STATUS[test_data]) + self.assertFalse(mock_send_command.called, "send_command['INST'] should not have been called") def test_projector_process_rfil_save(self): """ @@ -582,3 +1037,111 @@ class TestPJLinkCommands(TestCase): # THEN: Serial number should be set self.assertNotEquals(pjlink.serial_no, test_number, 'Projector serial number should NOT have been set') + + @patch.object(openlp.core.lib.projector.pjlink, 'log') + def test_projector_process_sver(self, mock_log): + """ + Test invalid software version information - too long + """ + # GIVEN: Test object + pjlink = pjlink_test + pjlink.sw_version = None + pjlink.sw_version_received = None + test_data = 'Test 1 Subtest 1' + test_log = "(127.0.0.1) Setting projector software version to 'Test 1 Subtest 1'" + mock_log.reset_mock() + + # WHEN: process_sver called with invalid data + pjlink.process_sver(data=test_data) + + # THEN: Version information should not change + self.assertEqual(pjlink.sw_version, test_data, 'Software version should have been updated') + self.assertIsNone(pjlink.sw_version_received, 'Received software version should not have changed') + mock_log.debug.assert_called_once_with(test_log) + + @patch.object(openlp.core.lib.projector.pjlink, 'log') + def test_projector_process_sver_changed(self, mock_log): + """ + Test invalid software version information - Received different than saved + """ + # GIVEN: Test object + pjlink = pjlink_test + test_data_new = 'Test 1 Subtest 2' + test_data_old = 'Test 1 Subtest 1' + pjlink.sw_version = test_data_old + pjlink.sw_version_received = None + test_log = '(127.0.0.1) Saving new serial number as sw_version_received' + mock_log.reset_mock() + + # WHEN: process_sver called with invalid data + pjlink.process_sver(data=test_data_new) + + # THEN: Version information should not change + self.assertEqual(pjlink.sw_version, test_data_old, 'Software version should not have been updated') + self.assertEqual(pjlink.sw_version_received, test_data_new, + 'Received software version should have been changed') + self.assertEqual(mock_log.warning.call_count, 4, 'log.warn should have been called 4 times') + # There was 4 calls, but only the last one is checked with this method + mock_log.warning.assert_called_with(test_log) + + @patch.object(openlp.core.lib.projector.pjlink, 'log') + def test_projector_process_sver_invalid(self, mock_log): + """ + Test invalid software version information - too long + """ + # GIVEN: Test object + pjlink = pjlink_test + pjlink.sw_version = None + pjlink.sw_version_received = None + test_data = 'This is a test software version line that is too long based on PJLink version 2 specs' + test_log = "Invalid software version - too long" + mock_log.reset_mock() + + # WHEN: process_sver called with invalid data + pjlink.process_sver(data=test_data) + + # THEN: Version information should not change + self.assertIsNone(pjlink.sw_version, 'Software version should not have changed') + self.assertIsNone(pjlink.sw_version_received, 'Received software version should not have changed') + mock_log.warning.assert_called_once_with(test_log) + + def test_projector_reset_information(self): + """ + Test reset_information() resets all information and stops timers + """ + # GIVEN: Test object and test data + pjlink = pjlink_test + pjlink.power = S_ON + pjlink.pjlink_name = 'OPENLPTEST' + pjlink.manufacturer = 'PJLINK' + pjlink.model = '1' + pjlink.shutter = True + pjlink.mute = True + pjlink.lamp = True + pjlink.fan = True + pjlink.source_available = True + pjlink.other_info = 'ANOTHER TEST' + pjlink.send_queue = True + pjlink.send_busy = True + + # WHEN: reset_information() is called + with patch.object(pjlink, 'timer') as mock_timer: + with patch.object(pjlink, 'socket_timer') as mock_socket_timer: + pjlink.reset_information() + + # THEN: All information should be reset and timers stopped + self.assertEqual(pjlink.power, S_OFF, 'Projector power should be OFF') + self.assertIsNone(pjlink.pjlink_name, 'Projector pjlink_name should be None') + self.assertIsNone(pjlink.manufacturer, 'Projector manufacturer should be None') + self.assertIsNone(pjlink.model, 'Projector model should be None') + self.assertIsNone(pjlink.shutter, 'Projector shutter should be None') + self.assertIsNone(pjlink.mute, 'Projector shuttter should be None') + self.assertIsNone(pjlink.lamp, 'Projector lamp should be None') + self.assertIsNone(pjlink.fan, 'Projector fan should be None') + self.assertIsNone(pjlink.source_available, 'Projector source_available should be None') + self.assertIsNone(pjlink.source, 'Projector source should be None') + self.assertIsNone(pjlink.other_info, 'Projector other_info should be None') + self.assertEqual(pjlink.send_queue, [], 'Projector send_queue should be an empty list') + self.assertFalse(pjlink.send_busy, 'Projector send_busy should be False') + self.assertTrue(mock_timer.stop.called, 'Projector timer.stop() should have been called') + self.assertTrue(mock_socket_timer.stop.called, 'Projector socket_timer.stop() should have been called')