diff --git a/openlp/core/lib/__init__.py b/openlp/core/lib/__init__.py index 329eb0149..77fefee39 100644 --- a/openlp/core/lib/__init__.py +++ b/openlp/core/lib/__init__.py @@ -655,5 +655,5 @@ from .imagemanager import ImageManager from .renderer import Renderer from .mediamanageritem import MediaManagerItem from .projector.db import ProjectorDB, Projector -from .projector.pjlink1 import PJLink +from .projector.pjlink import PJLink from .projector.constants import PJLINK_PORT, ERROR_MSG, ERROR_STRING diff --git a/openlp/core/lib/projector/constants.py b/openlp/core/lib/projector/constants.py index d4e6904e4..cd66c4222 100644 --- a/openlp/core/lib/projector/constants.py +++ b/openlp/core/lib/projector/constants.py @@ -154,7 +154,7 @@ PJLINK_VALID_CMD = { }, 'SRCH': {'version': ['2', ], 'description': translate('OpenLP.PJLinkConstants', - 'UDP broadcast search request for available projectors.') + 'UDP broadcast search request for available projectors. Reply is ACKN.') }, 'SVER': {'version': ['2', ], 'description': translate('OpenLP.PJLinkConstants', diff --git a/openlp/core/lib/projector/pjlink1.py b/openlp/core/lib/projector/pjlink.py similarity index 93% rename from openlp/core/lib/projector/pjlink1.py rename to openlp/core/lib/projector/pjlink.py index b2b1a4af1..cc216b6bc 100644 --- a/openlp/core/lib/projector/pjlink1.py +++ b/openlp/core/lib/projector/pjlink.py @@ -20,14 +20,17 @@ # Temple Place, Suite 330, Boston, MA 02111-1307 USA # ############################################################################### """ - :mod:`openlp.core.lib.projector.pjlink1` module + :mod:`openlp.core.lib.projector.pjlink` module Provides the necessary functions for connecting to a PJLink-capable projector. - See PJLink Class 1 Specifications for details. - http://pjlink.jbmia.or.jp/english/dl.html - + PJLink Class 1 Specifications. + http://pjlink.jbmia.or.jp/english/dl_class1.html Section 5-1 PJLink Specifications + Section 5-5 Guidelines for Input Terminals + PJLink Class 2 Specifications. + http://pjlink.jbmia.or.jp/english/dl_class2.html + Section 5-1 PJLink Specifications Section 5-5 Guidelines for Input Terminals NOTE: @@ -40,7 +43,7 @@ import logging log = logging.getLogger(__name__) -log.debug('pjlink1 loaded') +log.debug('pjlink loaded') __all__ = ['PJLink'] @@ -69,88 +72,17 @@ PJLINK_HEADER = '{prefix}{{linkclass}}'.format(prefix=PJLINK_PREFIX) PJLINK_SUFFIX = CR -class PJLink(QtNetwork.QTcpSocket): +class PJLinkCommands(object): """ - Socket service for connecting to a PJLink-capable projector. + Process replies from PJLink projector. """ - # Signals sent by this module - changeStatus = QtCore.pyqtSignal(str, int, str) - projectorNetwork = QtCore.pyqtSignal(int) # Projector network activity - projectorStatus = QtCore.pyqtSignal(int) # Status update - projectorAuthentication = QtCore.pyqtSignal(str) # Authentication error - projectorNoAuthentication = QtCore.pyqtSignal(str) # PIN set and no authentication needed - projectorReceivedData = QtCore.pyqtSignal() # Notify when received data finished processing - projectorUpdateIcons = QtCore.pyqtSignal() # Update the status icons on toolbar - # New commands available in PJLink Class 2 - pjlink_udp_commands = [ - 'ACKN', - 'ERST', # Class 1 or 2 - 'INPT', # Class 1 or 2 - 'LKUP', - 'POWR', # Class 1 or 2 - 'SRCH' - ] - - def __init__(self, name=None, ip=None, port=PJLINK_PORT, pin=None, *args, **kwargs): + def __init__(self, *args, **kwargs): """ - Setup for instance. - - :param name: Display name - :param ip: IP address to connect to - :param port: Port to use. Default to PJLINK_PORT - :param pin: Access pin (if needed) - - Optional parameters - :param dbid: Database ID number - :param location: Location where projector is physically located - :param notes: Extra notes about the projector - :param poll_time: Time (in seconds) to poll connected projector - :param socket_timeout: Time (in seconds) to abort the connection if no response + Setup for the process commands """ - log.debug('PJlink(args={args} kwargs={kwargs})'.format(args=args, kwargs=kwargs)) - self.name = name - self.ip = ip - self.port = port - self.pin = pin + log.debug('PJlinkCommands(args={args} kwargs={kwargs})'.format(args=args, kwargs=kwargs)) super().__init__() - self.model_lamp = None - self.model_filter = None - self.mac_adx = kwargs.get('mac_adx') - self.serial_no = None - self.serial_no_received = None # Used only if saved serial number is different than received serial number - self.dbid = None - self.db_update = False # Use to check if db needs to be updated prior to exiting - self.location = None - self.notes = None - self.dbid = kwargs.get('dbid') - self.location = kwargs.get('location') - self.notes = kwargs.get('notes') - # Poll time 20 seconds unless called with something else - self.poll_time = 20000 if 'poll_time' not in kwargs else kwargs['poll_time'] * 1000 - # Timeout 5 seconds unless called with something else - self.socket_timeout = 5000 if 'socket_timeout' not in kwargs else kwargs['socket_timeout'] * 1000 - # In case we're called from somewhere that only wants information - self.no_poll = 'no_poll' in kwargs - self.i_am_running = False - self.status_connect = S_NOT_CONNECTED - self.last_command = '' - self.projector_status = S_NOT_CONNECTED - self.error_status = S_OK - # Socket information - # Add enough space to input buffer for extraneous \n \r - self.max_size = PJLINK_MAX_PACKET + 2 - self.setReadBufferSize(self.max_size) - # PJLink information - self.pjlink_class = '1' # Default class - self.reset_information() - # Set from ProjectorManager.add_projector() - self.widget = None # QListBox entry - self.timer = None # Timer that calls the poll_loop - self.send_queue = [] - self.send_busy = False - # Socket timer for some possible brain-dead projectors or network cable pulled - self.socket_timer = None # Map command to function self.pjlink_functions = { 'AVMT': self.process_avmt, @@ -173,29 +105,30 @@ class PJLink(QtNetwork.QTcpSocket): def reset_information(self): """ - Reset projector-specific information to default + Initialize instance variables. Also used to reset projector-specific information to default. """ log.debug('({ip}) reset_information() connect status is {state}'.format(ip=self.ip, state=self.state())) - self.send_queue = [] - self.power = S_OFF - self.pjlink_name = None - self.manufacturer = None - self.model = None - self.serial_no = None + self.fan = None # ERST + self.filter_time = None # FILT + self.lamp = None # LAMP + self.mac_adx_received = None # ACKN + self.manufacturer = None # INF1 + self.model = None # INF2 + self.model_filter = None # RFIL + self.model_lamp = None # RLMP + self.mute = None # AVMT + self.other_info = None # INFO + self.pjlink_class = PJLINK_CLASS # Default class + self.pjlink_name = None # NAME + self.power = S_OFF # POWR + self.serial_no = None # SNUM self.serial_no_received = None - self.sw_version = None + self.sw_version = None # SVER self.sw_version_received = None - self.mac_adx = None - self.shutter = None - self.mute = None - self.lamp = None - self.model_lamp = None - self.fan = None - self.filter_time = None - self.model_filter = None - self.source_available = None - self.source = None - self.other_info = None + self.shutter = None # AVMT + self.source_available = None # INST + self.source = None # INPT + # These should be part of PJLink() class, but set here for convenience if hasattr(self, 'timer'): log.debug('({ip}): Calling timer.stop()'.format(ip=self.ip)) self.timer.stop() @@ -203,6 +136,412 @@ class PJLink(QtNetwork.QTcpSocket): log.debug('({ip}): Calling socket_timer.stop()'.format(ip=self.ip)) self.socket_timer.stop() self.send_busy = False + self.send_queue = [] + + def process_command(self, cmd, data): + """ + Verifies any return error code. Calls the appropriate command handler. + + :param cmd: Command to process + :param data: Data being processed + """ + log.debug('({ip}) Processing command "{cmd}" with data "{data}"'.format(ip=self.ip, + cmd=cmd, + data=data)) + # Check if we have a future command not available yet + _cmd = cmd.upper() + _data = data.upper() + if _cmd not in PJLINK_VALID_CMD: + log.error("({ip}) Ignoring command='{cmd}' (Invalid/Unknown)".format(ip=self.ip, cmd=cmd)) + return + elif _cmd not in self.pjlink_functions: + log.warn("({ip}) Unable to process command='{cmd}' (Future option)".format(ip=self.ip, cmd=cmd)) + return + elif _data in PJLINK_ERRORS: + # Oops - projector error + log.error('({ip}) Projector returned error "{data}"'.format(ip=self.ip, data=data)) + if _data == 'ERRA': + # Authentication error + self.disconnect_from_host() + self.change_status(E_AUTHENTICATION) + log.debug('({ip}) emitting projectorAuthentication() signal'.format(ip=self.ip)) + self.projectorAuthentication.emit(self.name) + elif _data == 'ERR1': + # Undefined command + self.change_status(E_UNDEFINED, '{error}: "{data}"'.format(error=ERROR_MSG[E_UNDEFINED], + data=cmd)) + elif _data == 'ERR2': + # Invalid parameter + self.change_status(E_PARAMETER) + elif _data == 'ERR3': + # Projector busy + self.change_status(E_UNAVAILABLE) + elif _data == 'ERR4': + # Projector/display error + self.change_status(E_PROJECTOR) + self.receive_data_signal() + return + # Command succeeded - no extra information + elif _data == 'OK': + log.debug('({ip}) Command returned OK'.format(ip=self.ip)) + # A command returned successfully + self.receive_data_signal() + return + # Command checks already passed + log.debug('({ip}) Calling function for {cmd}'.format(ip=self.ip, cmd=cmd)) + self.receive_data_signal() + self.pjlink_functions[_cmd](data) + + def process_avmt(self, data): + """ + Process shutter and speaker status. See PJLink specification for format. + Update self.mute (audio) and self.shutter (video shutter). + + :param data: Shutter and audio status + """ + settings = {'11': {'shutter': True, 'mute': self.mute}, + '21': {'shutter': self.shutter, 'mute': True}, + '30': {'shutter': False, 'mute': False}, + '31': {'shutter': True, 'mute': True} + } + if data in settings: + shutter = settings[data]['shutter'] + mute = settings[data]['mute'] + # Check if we need to update the icons + update_icons = (shutter != self.shutter) or (mute != self.mute) + self.shutter = shutter + self.mute = mute + else: + log.warning('({ip}) Unknown shutter response: {data}'.format(ip=self.ip, data=data)) + if update_icons: + self.projectorUpdateIcons.emit() + return + + def process_clss(self, data): + """ + PJLink class that this projector supports. See PJLink specification for format. + Updates self.class. + + :param data: Class that projector supports. + """ + # bug 1550891: Projector returns non-standard class response: + # : Expected: '%1CLSS=1' + # : 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)) + # 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. + try: + clss = re.findall('\d', data)[0] # Should only be the first match + except IndexError: + log.error("({ip}) No numbers found in class version reply - defaulting to class '1'".format(ip=self.ip)) + clss = '1' + elif not data.isdigit(): + log.error("({ip}) NAN class version reply - defaulting to class '1'".format(ip=self.ip)) + clss = '1' + else: + clss = data + self.pjlink_class = clss + log.debug('({ip}) Setting pjlink_class for this projector to "{data}"'.format(ip=self.ip, + data=self.pjlink_class)) + return + + def process_erst(self, data): + """ + Error status. See PJLink Specifications for format. + Updates self.projector_errors + +\ :param data: Error status + """ + try: + datacheck = int(data) + except ValueError: + # Bad data - ignore + return + if datacheck == 0: + self.projector_errors = None + else: + self.projector_errors = {} + # Fan + if data[0] != '0': + self.projector_errors[translate('OpenLP.ProjectorPJLink', 'Fan')] = \ + PJLINK_ERST_STATUS[data[0]] + # Lamp + if data[1] != '0': + self.projector_errors[translate('OpenLP.ProjectorPJLink', 'Lamp')] = \ + PJLINK_ERST_STATUS[data[1]] + # Temp + if data[2] != '0': + self.projector_errors[translate('OpenLP.ProjectorPJLink', 'Temperature')] = \ + PJLINK_ERST_STATUS[data[2]] + # Cover + if data[3] != '0': + self.projector_errors[translate('OpenLP.ProjectorPJLink', 'Cover')] = \ + PJLINK_ERST_STATUS[data[3]] + # Filter + if data[4] != '0': + self.projector_errors[translate('OpenLP.ProjectorPJLink', 'Filter')] = \ + PJLINK_ERST_STATUS[data[4]] + # Other + if data[5] != '0': + self.projector_errors[translate('OpenLP.ProjectorPJLink', 'Other')] = \ + PJLINK_ERST_STATUS[data[5]] + return + + def process_inf1(self, data): + """ + Manufacturer name set in projector. + Updates self.manufacturer + + :param data: Projector manufacturer + """ + self.manufacturer = data + log.debug('({ip}) Setting projector manufacturer data to "{data}"'.format(ip=self.ip, data=self.manufacturer)) + return + + def process_inf2(self, data): + """ + Projector Model set in projector. + Updates self.model. + + :param data: Model name + """ + self.model = data + log.debug('({ip}) Setting projector model to "{data}"'.format(ip=self.ip, data=self.model)) + return + + def process_info(self, data): + """ + Any extra info set in projector. + Updates self.other_info. + + :param data: Projector other info + """ + self.other_info = data + log.debug('({ip}) Setting projector other_info to "{data}"'.format(ip=self.ip, data=self.other_info)) + return + + def process_inpt(self, data): + """ + Current source input selected. See PJLink specification for format. + Update self.source + + :param data: Currently selected source + """ + self.source = data + log.info('({ip}) Setting data source to "{data}"'.format(ip=self.ip, data=self.source)) + return + + def process_inst(self, data): + """ + Available source inputs. See PJLink specification for format. + Updates self.source_available + + :param data: Sources list + """ + sources = [] + check = data.split() + for source in check: + sources.append(source) + sources.sort() + self.source_available = sources + self.projectorUpdateIcons.emit() + log.debug('({ip}) Setting projector sources_available to "{data}"'.format(ip=self.ip, + data=self.source_available)) + return + + def process_lamp(self, data): + """ + Lamp(s) status. See PJLink Specifications for format. + Data may have more than 1 lamp to process. + Update self.lamp dictionary with lamp status. + + :param data: Lamp(s) status. + """ + lamps = [] + data_dict = data.split() + while data_dict: + try: + fill = {'Hours': int(data_dict[0]), 'On': False if data_dict[1] == '0' else True} + except ValueError: + # In case of invalid entry + log.warning('({ip}) process_lamp(): Invalid data "{data}"'.format(ip=self.ip, data=data)) + return + lamps.append(fill) + data_dict.pop(0) # Remove lamp hours + data_dict.pop(0) # Remove lamp on/off + self.lamp = lamps + return + + def process_name(self, data): + """ + Projector name set in projector. + Updates self.pjlink_name + + :param data: Projector name + """ + self.pjlink_name = data + log.debug('({ip}) Setting projector PJLink name to "{data}"'.format(ip=self.ip, data=self.pjlink_name)) + return + + def process_powr(self, data): + """ + Power status. See PJLink specification for format. + Update self.power with status. Update icons if change from previous setting. + + :param data: Power status + """ + log.debug('({ip}: Processing POWR command'.format(ip=self.ip)) + if data in PJLINK_POWR_STATUS: + power = PJLINK_POWR_STATUS[data] + update_icons = self.power != power + self.power = power + self.change_status(PJLINK_POWR_STATUS[data]) + if update_icons: + self.projectorUpdateIcons.emit() + # Update the input sources available + if power == S_ON: + self.send_command('INST') + else: + # Log unknown status response + log.warning('({ip}) Unknown power response: {data}'.format(ip=self.ip, data=data)) + return + + def process_rfil(self, data): + """ + Process replacement filter type + """ + 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)) + + def process_rlmp(self, data): + """ + Process replacement lamp type + """ + 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)) + + def process_snum(self, data): + """ + Serial number of projector. + + :param data: Serial number from projector. + """ + if self.serial_no is None: + log.debug("({ip}) Setting projector serial number to '{data}'".format(ip=self.ip, data=data)) + self.serial_no = data + self.db_update = False + 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)) + self.serial_no_received = data + + def process_sver(self, data): + """ + Software version of projector + """ + if 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)) + self.sw_version_received = data + + +class PJLink(PJLinkCommands, QtNetwork.QTcpSocket): + """ + Socket service for connecting to a PJLink-capable projector. + """ + # Signals sent by this module + changeStatus = QtCore.pyqtSignal(str, int, str) + projectorNetwork = QtCore.pyqtSignal(int) # Projector network activity + projectorStatus = QtCore.pyqtSignal(int) # Status update + projectorAuthentication = QtCore.pyqtSignal(str) # Authentication error + projectorNoAuthentication = QtCore.pyqtSignal(str) # PIN set and no authentication needed + projectorReceivedData = QtCore.pyqtSignal() # Notify when received data finished processing + projectorUpdateIcons = QtCore.pyqtSignal() # Update the status icons on toolbar + + # New commands available in PJLink Class 2 + pjlink_udp_commands = [ + 'ACKN', # Class 2 + 'ERST', # Class 1 or 2 + 'INPT', # Class 1 or 2 + 'LKUP', # Class 2 + 'POWR', # Class 1 or 2 + 'SRCH' # Class 2 + ] + + def __init__(self, port=PJLINK_PORT, *args, **kwargs): + """ + Setup for instance. + Options should be in kwargs except for port which does have a default. + + :param name: Display name + :param ip: IP address to connect to + :param port: Port to use. Default to PJLINK_PORT + :param pin: Access pin (if needed) + + Optional parameters + :param dbid: Database ID number + :param location: Location where projector is physically located + :param notes: Extra notes about the projector + :param poll_time: Time (in seconds) to poll connected projector + :param socket_timeout: Time (in seconds) to abort the connection if no response + """ + log.debug('PJlink(args={args} kwargs={kwargs})'.format(args=args, kwargs=kwargs)) + super().__init__() + self.dbid = kwargs.get('dbid') + self.ip = kwargs.get('ip') + self.location = kwargs.get('location') + self.mac_adx = kwargs.get('mac_adx') + self.name = kwargs.get('name') + self.notes = kwargs.get('notes') + self.pin = kwargs.get('pin') + self.port = port + self.db_update = False # Use to check if db needs to be updated prior to exiting + # Poll time 20 seconds unless called with something else + self.poll_time = 20000 if 'poll_time' not in kwargs else kwargs['poll_time'] * 1000 + # Timeout 5 seconds unless called with something else + self.socket_timeout = 5000 if 'socket_timeout' not in kwargs else kwargs['socket_timeout'] * 1000 + # In case we're called from somewhere that only wants information + self.no_poll = 'no_poll' in kwargs + self.i_am_running = False + self.status_connect = S_NOT_CONNECTED + self.last_command = '' + self.projector_status = S_NOT_CONNECTED + self.error_status = S_OK + # Socket information + # Add enough space to input buffer for extraneous \n \r + self.max_size = PJLINK_MAX_PACKET + 2 + self.setReadBufferSize(self.max_size) + self.reset_information() + # Set from ProjectorManager.add_projector() + self.widget = None # QListBox entry + self.timer = None # Timer that calls the poll_loop + self.send_queue = [] + self.send_busy = False + # Socket timer for some possible brain-dead projectors or network cable pulled + self.socket_timer = None def thread_started(self): """ @@ -290,28 +629,6 @@ class PJLink(QtNetwork.QTcpSocket): if self.model_lamp is None: self.send_command('RLMP', queue=True) - def process_rfil(self, data): - """ - Process replacement filter type - """ - 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)) - - def process_rlmp(self, data): - """ - Process replacement lamp type - """ - 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)) - def _get_status(self, status): """ Helper to retrieve status/error codes and convert to strings. @@ -474,6 +791,7 @@ class PJLink(QtNetwork.QTcpSocket): self.send_busy = False return read = self.readLine(self.max_size) + log.debug("({ip}) get_data(): '{buff}'".format(ip=self.ip, buff=read)) if read == -1: # No data available log.debug('({ip}) get_data(): No data available (-1)'.format(ip=self.ip)) @@ -626,317 +944,6 @@ class PJLink(QtNetwork.QTcpSocket): self.change_status(E_NETWORK, translate('OpenLP.PJLink', 'Error while sending data to projector')) - def process_command(self, cmd, data): - """ - Verifies any return error code. Calls the appropriate command handler. - - :param cmd: Command to process - :param data: Data being processed - """ - log.debug('({ip}) Processing command "{cmd}" with data "{data}"'.format(ip=self.ip, - cmd=cmd, - data=data)) - # Check if we have a future command not available yet - if cmd not in PJLINK_VALID_CMD: - log.error('({ip}) Unknown command received - ignoring'.format(ip=self.ip)) - return - elif cmd not in self.pjlink_functions: - log.warn('({ip}) Future command received - unable to process yet'.format(ip=self.ip)) - return - elif data in PJLINK_ERRORS: - # Oops - projector error - log.error('({ip}) Projector returned error "{data}"'.format(ip=self.ip, data=data)) - if data.upper() == 'ERRA': - # Authentication error - self.disconnect_from_host() - self.change_status(E_AUTHENTICATION) - log.debug('({ip}) emitting projectorAuthentication() signal'.format(ip=self.ip)) - self.projectorAuthentication.emit(self.name) - elif data.upper() == 'ERR1': - # Undefined command - self.change_status(E_UNDEFINED, '{error}: "{data}"'.format(error=ERROR_MSG[E_UNDEFINED], - data=cmd)) - elif data.upper() == 'ERR2': - # Invalid parameter - self.change_status(E_PARAMETER) - elif data.upper() == 'ERR3': - # Projector busy - self.change_status(E_UNAVAILABLE) - elif data.upper() == 'ERR4': - # Projector/display error - self.change_status(E_PROJECTOR) - self.receive_data_signal() - return - # Command succeeded - no extra information - elif data.upper() == 'OK': - log.debug('({ip}) Command returned OK'.format(ip=self.ip)) - # A command returned successfully - self.receive_data_signal() - return - # Command checks already passed - log.debug('({ip}) Calling function for {cmd}'.format(ip=self.ip, cmd=cmd)) - self.receive_data_signal() - self.pjlink_functions[cmd](data) - - def process_lamp(self, data): - """ - Lamp(s) status. See PJLink Specifications for format. - Data may have more than 1 lamp to process. - Update self.lamp dictionary with lamp status. - - :param data: Lamp(s) status. - """ - lamps = [] - data_dict = data.split() - while data_dict: - try: - fill = {'Hours': int(data_dict[0]), 'On': False if data_dict[1] == '0' else True} - except ValueError: - # In case of invalid entry - log.warning('({ip}) process_lamp(): Invalid data "{data}"'.format(ip=self.ip, data=data)) - return - lamps.append(fill) - data_dict.pop(0) # Remove lamp hours - data_dict.pop(0) # Remove lamp on/off - self.lamp = lamps - return - - def process_powr(self, data): - """ - Power status. See PJLink specification for format. - Update self.power with status. Update icons if change from previous setting. - - :param data: Power status - """ - log.debug('({ip}: Processing POWR command'.format(ip=self.ip)) - if data in PJLINK_POWR_STATUS: - power = PJLINK_POWR_STATUS[data] - update_icons = self.power != power - self.power = power - self.change_status(PJLINK_POWR_STATUS[data]) - if update_icons: - self.projectorUpdateIcons.emit() - # Update the input sources available - if power == S_ON: - self.send_command('INST') - else: - # Log unknown status response - log.warning('({ip}) Unknown power response: {data}'.format(ip=self.ip, data=data)) - return - - def process_avmt(self, data): - """ - Process shutter and speaker status. See PJLink specification for format. - Update self.mute (audio) and self.shutter (video shutter). - - :param data: Shutter and audio status - """ - shutter = self.shutter - mute = self.mute - if data == '11': - shutter = True - mute = False - elif data == '21': - shutter = False - mute = True - elif data == '30': - shutter = False - mute = False - elif data == '31': - shutter = True - mute = True - else: - log.warning('({ip}) Unknown shutter response: {data}'.format(ip=self.ip, data=data)) - update_icons = shutter != self.shutter - update_icons = update_icons or mute != self.mute - self.shutter = shutter - self.mute = mute - if update_icons: - self.projectorUpdateIcons.emit() - return - - def process_inpt(self, data): - """ - Current source input selected. See PJLink specification for format. - Update self.source - - :param data: Currently selected source - """ - self.source = data - log.info('({ip}) Setting data source to "{data}"'.format(ip=self.ip, data=self.source)) - return - - def process_clss(self, data): - """ - PJLink class that this projector supports. See PJLink specification for format. - Updates self.class. - - :param data: Class that projector supports. - """ - # bug 1550891: Projector returns non-standard class response: - # : Expected: '%1CLSS=1' - # : 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)) - # 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. - try: - clss = re.findall('\d', data)[0] # Should only be the first match - except IndexError: - log.error("({ip}) No numbers found in class version reply - defaulting to class '1'".format(ip=self.ip)) - clss = '1' - elif not data.isdigit(): - log.error("({ip}) NAN class version reply - defaulting to class '1'".format(ip=self.ip)) - clss = '1' - else: - clss = data - self.pjlink_class = clss - log.debug('({ip}) Setting pjlink_class for this projector to "{data}"'.format(ip=self.ip, - data=self.pjlink_class)) - return - - def process_name(self, data): - """ - Projector name set in projector. - Updates self.pjlink_name - - :param data: Projector name - """ - self.pjlink_name = data - log.debug('({ip}) Setting projector PJLink name to "{data}"'.format(ip=self.ip, data=self.pjlink_name)) - return - - def process_inf1(self, data): - """ - Manufacturer name set in projector. - Updates self.manufacturer - - :param data: Projector manufacturer - """ - self.manufacturer = data - log.debug('({ip}) Setting projector manufacturer data to "{data}"'.format(ip=self.ip, data=self.manufacturer)) - return - - def process_inf2(self, data): - """ - Projector Model set in projector. - Updates self.model. - - :param data: Model name - """ - self.model = data - log.debug('({ip}) Setting projector model to "{data}"'.format(ip=self.ip, data=self.model)) - return - - def process_info(self, data): - """ - Any extra info set in projector. - Updates self.other_info. - - :param data: Projector other info - """ - self.other_info = data - log.debug('({ip}) Setting projector other_info to "{data}"'.format(ip=self.ip, data=self.other_info)) - return - - def process_inst(self, data): - """ - Available source inputs. See PJLink specification for format. - Updates self.source_available - - :param data: Sources list - """ - sources = [] - check = data.split() - for source in check: - sources.append(source) - sources.sort() - self.source_available = sources - self.projectorUpdateIcons.emit() - log.debug('({ip}) Setting projector sources_available to "{data}"'.format(ip=self.ip, - data=self.source_available)) - return - - def process_erst(self, data): - """ - Error status. See PJLink Specifications for format. - Updates self.projector_errors - - :param data: Error status - """ - try: - datacheck = int(data) - except ValueError: - # Bad data - ignore - return - if datacheck == 0: - self.projector_errors = None - else: - self.projector_errors = {} - # Fan - if data[0] != '0': - self.projector_errors[translate('OpenLP.ProjectorPJLink', 'Fan')] = \ - PJLINK_ERST_STATUS[data[0]] - # Lamp - if data[1] != '0': - self.projector_errors[translate('OpenLP.ProjectorPJLink', 'Lamp')] = \ - PJLINK_ERST_STATUS[data[1]] - # Temp - if data[2] != '0': - self.projector_errors[translate('OpenLP.ProjectorPJLink', 'Temperature')] = \ - PJLINK_ERST_STATUS[data[2]] - # Cover - if data[3] != '0': - self.projector_errors[translate('OpenLP.ProjectorPJLink', 'Cover')] = \ - PJLINK_ERST_STATUS[data[3]] - # Filter - if data[4] != '0': - self.projector_errors[translate('OpenLP.ProjectorPJLink', 'Filter')] = \ - PJLINK_ERST_STATUS[data[4]] - # Other - if data[5] != '0': - self.projector_errors[translate('OpenLP.ProjectorPJLink', 'Other')] = \ - PJLINK_ERST_STATUS[data[5]] - return - - def process_snum(self, data): - """ - Serial number of projector. - - :param data: Serial number from projector. - """ - if self.serial_no is None: - log.debug("({ip}) Setting projector serial number to '{data}'".format(ip=self.ip, data=data)) - self.serial_no = data - self.db_update = False - 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)) - self.serial_no_received = data - - def process_sver(self, data): - """ - Software version of projector - """ - if 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)) - self.sw_version_received = data - def connect_to_host(self): """ Initiate connection to projector. @@ -1098,11 +1105,3 @@ class PJLink(QtNetwork.QTcpSocket): self.send_busy = False self.projectorReceivedData.emit() return - - def _not_implemented(self, cmd): - """ - Log when a future PJLink command has not been implemented yet. - """ - log.warn("({ip}) Future command '{cmd}' has not been implemented yet".format(ip=self.ip, - cmd=cmd)) - return diff --git a/openlp/core/lib/projector/upgrade.py b/openlp/core/lib/projector/upgrade.py index acb3c1b0b..83cd2defb 100644 --- a/openlp/core/lib/projector/upgrade.py +++ b/openlp/core/lib/projector/upgrade.py @@ -42,7 +42,7 @@ def upgrade_1(session, metadata): """ Version 1 upgrade - old db might/might not be versioned. """ - log.debug('Skipping upgrade_1 of projector DB - not used') + log.debug('Skipping projector DB upgrade to version 1 - not used') def upgrade_2(session, metadata): @@ -60,14 +60,14 @@ def upgrade_2(session, metadata): :param session: DB session instance :param metadata: Metadata of current DB """ + log.debug('Checking projector DB upgrade to version 2') projector_table = Table('projector', metadata, autoload=True) - if 'mac_adx' not in [col.name for col in projector_table.c.values()]: - log.debug("Upgrading projector DB to version '2'") + upgrade_db = 'mac_adx' not in [col.name for col in projector_table.c.values()] + if upgrade_db: new_op = get_upgrade_op(session) new_op.add_column('projector', Column('mac_adx', types.String(18), server_default=null())) new_op.add_column('projector', Column('serial_no', types.String(30), server_default=null())) new_op.add_column('projector', Column('sw_version', types.String(30), server_default=null())) new_op.add_column('projector', Column('model_filter', types.String(30), server_default=null())) new_op.add_column('projector', Column('model_lamp', types.String(30), server_default=null())) - else: - log.warn("Skipping upgrade_2 of projector DB") + log.debug('{status} projector DB upgrade to version 2'.format(status='Updated' if upgrade_db else 'Skipping')) diff --git a/openlp/core/ui/projector/manager.py b/openlp/core/ui/projector/manager.py index 7eb5451ad..0aac006e0 100644 --- a/openlp/core/ui/projector/manager.py +++ b/openlp/core/ui/projector/manager.py @@ -38,7 +38,7 @@ from openlp.core.lib.projector.constants import ERROR_MSG, ERROR_STRING, E_AUTHE E_NETWORK, E_NOT_CONNECTED, E_UNKNOWN_SOCKET_ERROR, STATUS_STRING, S_CONNECTED, S_CONNECTING, S_COOLDOWN, \ S_INITIALIZE, S_NOT_CONNECTED, S_OFF, S_ON, S_STANDBY, S_WARMUP from openlp.core.lib.projector.db import ProjectorDB -from openlp.core.lib.projector.pjlink1 import PJLink +from openlp.core.lib.projector.pjlink import PJLink from openlp.core.lib.projector.pjlink2 import PJLinkUDP from openlp.core.ui.projector.editform import ProjectorEditForm from openlp.core.ui.projector.sourceselectform import SourceSelectTabs, SourceSelectSingle diff --git a/tests/functional/openlp_core_lib/test_projector_constants.py b/tests/functional/openlp_core_lib/test_projector_constants.py index 019c18888..90fee1e13 100644 --- a/tests/functional/openlp_core_lib/test_projector_constants.py +++ b/tests/functional/openlp_core_lib/test_projector_constants.py @@ -22,7 +22,7 @@ """ Package to test the openlp.core.lib.projector.constants package. """ -from unittest import TestCase, skip +from unittest import TestCase class TestProjectorConstants(TestCase): @@ -40,4 +40,4 @@ class TestProjectorConstants(TestCase): from openlp.core.lib.projector.constants import PJLINK_DEFAULT_CODES # THEN: Verify dictionary was build correctly - self.assertEquals(PJLINK_DEFAULT_CODES, TEST_VIDEO_CODES, 'PJLink video strings should match') + self.assertEqual(PJLINK_DEFAULT_CODES, TEST_VIDEO_CODES, 'PJLink video strings should match') diff --git a/tests/functional/openlp_core_lib/test_projector_db.py b/tests/functional/openlp_core_lib/test_projector_db.py index d4ff4e75c..e49e75245 100644 --- a/tests/functional/openlp_core_lib/test_projector_db.py +++ b/tests/functional/openlp_core_lib/test_projector_db.py @@ -29,8 +29,8 @@ import os import shutil from tempfile import mkdtemp -from unittest import TestCase, skip -from unittest.mock import MagicMock, patch +from unittest import TestCase +from unittest.mock import patch from openlp.core.lib.projector import upgrade from openlp.core.lib.db import upgrade_db @@ -413,7 +413,7 @@ class TestProjectorDB(TestCase): Test add_projector() fail """ # GIVEN: Test entry in the database - ignore_result = self.projector.add_projector(Projector(**TEST1_DATA)) + self.projector.add_projector(Projector(**TEST1_DATA)) # WHEN: Attempt to add same projector entry results = self.projector.add_projector(Projector(**TEST1_DATA)) @@ -439,7 +439,7 @@ class TestProjectorDB(TestCase): Test update_projector() when entry not in database """ # GIVEN: Projector entry in database - ignore_result = self.projector.add_projector(Projector(**TEST1_DATA)) + self.projector.add_projector(Projector(**TEST1_DATA)) projector = Projector(**TEST2_DATA) # WHEN: Attempt to update data with a different ID diff --git a/tests/functional/openlp_core_lib/test_projector_pjlink_base.py b/tests/functional/openlp_core_lib/test_projector_pjlink_base.py new file mode 100644 index 000000000..75981b75d --- /dev/null +++ b/tests/functional/openlp_core_lib/test_projector_pjlink_base.py @@ -0,0 +1,208 @@ +# -*- coding: utf-8 -*- +# vim: autoindent shiftwidth=4 expandtab textwidth=120 tabstop=4 softtabstop=4 + +############################################################################### +# OpenLP - Open Source Lyrics Projection # +# --------------------------------------------------------------------------- # +# Copyright (c) 2008-2015 OpenLP Developers # +# --------------------------------------------------------------------------- # +# This program is free software; you can redistribute it and/or modify it # +# under the terms of the GNU General Public License as published by the Free # +# Software Foundation; version 2 of the License. # +# # +# This program is distributed in the hope that it will be useful, but WITHOUT # +# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or # +# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for # +# more details. # +# # +# You should have received a copy of the GNU General Public License along # +# with this program; if not, write to the Free Software Foundation, Inc., 59 # +# Temple Place, Suite 330, Boston, MA 02111-1307 USA # +############################################################################### +""" +Package to test the openlp.core.lib.projector.pjlink base package. +""" +from unittest import TestCase +from unittest.mock import call, patch, MagicMock + +from openlp.core.lib.projector.pjlink import PJLink +from openlp.core.lib.projector.constants import E_PARAMETER, ERROR_STRING, S_ON, S_CONNECTED + +from tests.resources.projector.data import TEST_PIN, TEST_SALT, TEST_CONNECT_AUTHENTICATE, TEST_HASH + +pjlink_test = PJLink(name='test', ip='127.0.0.1', pin=TEST_PIN, no_poll=True) + + +class TestPJLinkBase(TestCase): + """ + Tests for the PJLink module + """ + @patch.object(pjlink_test, 'readyRead') + @patch.object(pjlink_test, 'send_command') + @patch.object(pjlink_test, 'waitForReadyRead') + @patch('openlp.core.common.qmd5_hash') + def test_authenticated_connection_call(self, + mock_qmd5_hash, + mock_waitForReadyRead, + mock_send_command, + mock_readyRead): + """ + Ticket 92187: Fix for projector connect with PJLink authentication exception. + """ + # GIVEN: Test object + pjlink = pjlink_test + + # WHEN: Calling check_login with authentication request: + pjlink.check_login(data=TEST_CONNECT_AUTHENTICATE) + + # THEN: Should have called qmd5_hash + self.assertTrue(mock_qmd5_hash.called_with(TEST_SALT, + "Connection request should have been called with TEST_SALT")) + self.assertTrue(mock_qmd5_hash.called_with(TEST_PIN, + "Connection request should have been called with TEST_PIN")) + + @patch.object(pjlink_test, 'change_status') + def test_status_change(self, mock_change_status): + """ + Test process_command call with ERR2 (Parameter) status + """ + # GIVEN: Test object + pjlink = pjlink_test + + # WHEN: process_command is called with "ERR2" status from projector + pjlink.process_command('POWR', 'ERR2') + + # THEN: change_status should have called change_status with E_UNDEFINED + # as first parameter + mock_change_status.called_with(E_PARAMETER, + 'change_status should have been called with "{}"'.format( + ERROR_STRING[E_PARAMETER])) + + @patch.object(pjlink_test, 'send_command') + @patch.object(pjlink_test, 'waitForReadyRead') + @patch.object(pjlink_test, 'projectorAuthentication') + @patch.object(pjlink_test, 'timer') + @patch.object(pjlink_test, 'socket_timer') + def test_bug_1593882_no_pin_authenticated_connection(self, + mock_socket_timer, + mock_timer, + mock_authentication, + mock_ready_read, + mock_send_command): + """ + Test bug 1593882 no pin and authenticated request exception + """ + # GIVEN: Test object and mocks + pjlink = pjlink_test + pjlink.pin = None + mock_ready_read.return_value = True + + # WHEN: call with authentication request and pin not set + pjlink.check_login(data=TEST_CONNECT_AUTHENTICATE) + + # THEN: 'No Authentication' signal should have been sent + mock_authentication.emit.assert_called_with(pjlink.ip) + + @patch.object(pjlink_test, 'waitForReadyRead') + @patch.object(pjlink_test, 'state') + @patch.object(pjlink_test, '_send_command') + @patch.object(pjlink_test, 'timer') + @patch.object(pjlink_test, 'socket_timer') + def test_bug_1593883_pjlink_authentication(self, + mock_socket_timer, + mock_timer, + mock_send_command, + mock_state, + mock_waitForReadyRead): + """ + Test bugfix 1593883 pjlink authentication + """ + # GIVEN: Test object and data + pjlink = pjlink_test + pjlink.pin = TEST_PIN + mock_state.return_value = pjlink.ConnectedState + mock_waitForReadyRead.return_value = True + + # WHEN: Athenticated connection is called + pjlink.check_login(data=TEST_CONNECT_AUTHENTICATE) + + # THEN: send_command should have the proper authentication + self.assertEqual("{test}".format(test=mock_send_command.call_args), + "call(data='{hash}%1CLSS ?\\r')".format(hash=TEST_HASH)) + + @patch.object(pjlink_test, 'disconnect_from_host') + def test_socket_abort(self, mock_disconnect): + """ + Test PJLink.socket_abort calls disconnect_from_host + """ + # GIVEN: Test object + pjlink = pjlink_test + + # WHEN: Calling socket_abort + pjlink.socket_abort() + + # THEN: disconnect_from_host should be called + self.assertTrue(mock_disconnect.called, 'Should have called disconnect_from_host') + + def test_poll_loop_not_connected(self): + """ + Test PJLink.poll_loop not connected return + """ + # GIVEN: Test object and mocks + pjlink = pjlink_test + pjlink.state = MagicMock() + pjlink.timer = MagicMock() + pjlink.state.return_value = False + pjlink.ConnectedState = True + + # WHEN: PJLink.poll_loop called + pjlink.poll_loop() + + # THEN: poll_loop should exit without calling any other method + self.assertFalse(pjlink.timer.called, 'Should have returned without calling any other method') + + @patch.object(pjlink_test, 'send_command') + def test_poll_loop_start(self, mock_send_command): + """ + Test PJLink.poll_loop makes correct calls + """ + # GIVEN: test object and test data + pjlink = pjlink_test + pjlink.state = MagicMock() + pjlink.timer = MagicMock() + pjlink.timer.interval = MagicMock() + pjlink.timer.setInterval = MagicMock() + pjlink.timer.start = MagicMock() + pjlink.poll_time = 20 + pjlink.power = S_ON + pjlink.source_available = None + pjlink.other_info = None + pjlink.manufacturer = None + pjlink.model = None + pjlink.pjlink_name = None + pjlink.ConnectedState = S_CONNECTED + pjlink.timer.interval.return_value = 10 + pjlink.state.return_value = S_CONNECTED + call_list = [ + call('POWR', queue=True), + call('ERST', queue=True), + call('LAMP', queue=True), + call('AVMT', queue=True), + call('INPT', queue=True), + call('INST', queue=True), + call('INFO', queue=True), + call('INF1', queue=True), + call('INF2', queue=True), + call('NAME', queue=True), + ] + + # WHEN: PJLink.poll_loop is called + pjlink.poll_loop() + + # THEN: proper calls were made to retrieve projector data + # First, call to update the timer with the next interval + self.assertTrue(pjlink.timer.setInterval.called, 'Should have updated the timer') + # Next, should have called the timer to start + self.assertTrue(pjlink.timer.start.called, 'Should have started the timer') + # Finally, should have called send_command with a list of projetctor status checks + mock_send_command.assert_has_calls(call_list, 'Should have queued projector queries') diff --git a/tests/functional/openlp_core_lib/test_projector_pjlink1.py b/tests/functional/openlp_core_lib/test_projector_pjlink_commands.py similarity index 63% rename from tests/functional/openlp_core_lib/test_projector_pjlink1.py rename to tests/functional/openlp_core_lib/test_projector_pjlink_commands.py index 969ad72e1..19f6fbdd4 100644 --- a/tests/functional/openlp_core_lib/test_projector_pjlink1.py +++ b/tests/functional/openlp_core_lib/test_projector_pjlink_commands.py @@ -20,44 +20,360 @@ # Temple Place, Suite 330, Boston, MA 02111-1307 USA # ############################################################################### """ -Package to test the openlp.core.lib.projector.pjlink1 package. +Package to test the openlp.core.lib.projector.pjlink commands package. """ from unittest import TestCase -from unittest.mock import call, patch, MagicMock +from unittest.mock import patch, MagicMock -from openlp.core.lib.projector.pjlink1 import PJLink -from openlp.core.lib.projector.constants import E_PARAMETER, ERROR_STRING, S_OFF, S_STANDBY, S_ON, \ - PJLINK_POWR_STATUS, S_CONNECTED +from openlp.core.lib.projector.pjlink import PJLink +from openlp.core.lib.projector.constants import PJLINK_ERST_STATUS, PJLINK_POWR_STATUS, \ + S_OFF, S_STANDBY, S_ON -from tests.resources.projector.data import TEST_PIN, TEST_SALT, TEST_CONNECT_AUTHENTICATE, TEST_HASH +from tests.resources.projector.data import TEST_PIN pjlink_test = PJLink(name='test', ip='127.0.0.1', pin=TEST_PIN, no_poll=True) +# ERST status codes +ERST_OK, ERST_WARN, ERST_ERR = '0', '1', '2' -class TestPJLink(TestCase): + +class TestPJLinkCommands(TestCase): """ Tests for the PJLink module """ - @patch.object(pjlink_test, 'readyRead') - @patch.object(pjlink_test, 'send_command') - @patch.object(pjlink_test, 'waitForReadyRead') - @patch('openlp.core.common.qmd5_hash') - def test_authenticated_connection_call(self, mock_qmd5_hash, mock_waitForReadyRead, mock_send_command, - mock_readyRead): + def test_projector_reset_information(self): """ - Ticket 92187: Fix for projector connect with PJLink authentication exception. + 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 + pjlink.timer = MagicMock() + pjlink.socket_timer = MagicMock() + + # 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() + + # 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') + + @patch.object(pjlink_test, 'projectorUpdateIcons') + def test_projector_process_avmt_closed_muted(self, mock_projectorReceivedData): + """ + Test avmt status shutter closed and mute off + """ + # GIVEN: Test object + pjlink = pjlink_test + pjlink.shutter = False + pjlink.mute = False + + # WHEN: Called with setting shutter to closed and mute on + pjlink.process_avmt('31') + + # THEN: Shutter should be closed and mute should be True + self.assertTrue(pjlink.shutter, 'Shutter should have been set to closed') + self.assertTrue(pjlink.mute, 'Audio should be off') + + @patch.object(pjlink_test, 'projectorUpdateIcons') + def test_projector_process_avmt_shutter_closed(self, mock_projectorReceivedData): + """ + Test avmt status shutter closed and audio muted + """ + # GIVEN: Test object + pjlink = pjlink_test + pjlink.shutter = False + pjlink.mute = True + + # WHEN: Called with setting shutter closed and mute off + pjlink.process_avmt('11') + + # THEN: Shutter should be True and mute should be False + self.assertTrue(pjlink.shutter, 'Shutter should have been set to closed') + self.assertTrue(pjlink.mute, 'Audio should not have changed') + + @patch.object(pjlink_test, 'projectorUpdateIcons') + def test_projector_process_avmt_audio_muted(self, mock_projectorReceivedData): + """ + Test avmt status shutter open and mute on + """ + # GIVEN: Test object + pjlink = pjlink_test + pjlink.shutter = True + pjlink.mute = False + + # WHEN: Called with setting shutter closed and mute on + pjlink.process_avmt('21') + + # THEN: Shutter should be closed and mute should be True + self.assertTrue(pjlink.shutter, 'Shutter should not have changed') + self.assertTrue(pjlink.mute, 'Audio should be off') + + @patch.object(pjlink_test, 'projectorUpdateIcons') + def test_projector_process_avmt_open_unmuted(self, mock_projectorReceivedData): + """ + Test avmt status shutter open and mute off off + """ + # GIVEN: Test object + pjlink = pjlink_test + pjlink.shutter = True + pjlink.mute = True + + # WHEN: Called with setting shutter to closed and mute on + pjlink.process_avmt('30') + + # THEN: Shutter should be closed and mute should be True + self.assertFalse(pjlink.shutter, 'Shutter should have been set to open') + self.assertFalse(pjlink.mute, 'Audio should be on') + + def test_projector_process_clss_one(self): + """ + Test class 1 sent from projector """ # GIVEN: Test object pjlink = pjlink_test - # WHEN: Calling check_login with authentication request: - pjlink.check_login(data=TEST_CONNECT_AUTHENTICATE) + # WHEN: Process class response + pjlink.process_clss('1') - # THEN: Should have called qmd5_hash - self.assertTrue(mock_qmd5_hash.called_with(TEST_SALT, - "Connection request should have been called with TEST_SALT")) - self.assertTrue(mock_qmd5_hash.called_with(TEST_PIN, - "Connection request should have been called with TEST_PIN")) + # THEN: Projector class should be set to 1 + self.assertEqual(pjlink.pjlink_class, '1', + 'Projector should have returned class=1') + + def test_projector_process_clss_two(self): + """ + Test class 2 sent from projector + """ + # GIVEN: Test object + pjlink = pjlink_test + + # WHEN: Process class response + pjlink.process_clss('2') + + # THEN: Projector class should be set to 1 + self.assertEqual(pjlink.pjlink_class, '2', + 'Projector should have returned class=2') + + def test_projector_process_clss_nonstandard_reply(self): + """ + Bugfix 1550891: CLSS request returns non-standard reply + """ + # GIVEN: Test object + pjlink = pjlink_test + + # WHEN: Process non-standard reply + pjlink.process_clss('Class 1') + + # THEN: Projector class should be set with proper value + self.assertEqual(pjlink.pjlink_class, '1', + 'Non-standard class reply should have set class=1') + + def test_projector_process_erst_all_ok(self): + """ + Test test_projector_process_erst_all_ok + """ + # GIVEN: Test object + pjlink = pjlink_test + chk_test = ERST_OK + + # WHEN: process_erst with no errors + pjlink.process_erst('{fan}{lamp}{temp}{cover}{filter}{other}'.format(fan=chk_test, + lamp=chk_test, + temp=chk_test, + cover=chk_test, + filter=chk_test, + other=chk_test)) + + # PJLink instance errors should be None + self.assertIsNone(pjlink.projector_errors, 'projector_errors should have been set to None') + + def test_projector_process_erst_all_warn(self): + """ + Test test_projector_process_erst_all_warn + """ + # GIVEN: Test object + pjlink = pjlink_test + chk_test = ERST_WARN + chk_code = PJLINK_ERST_STATUS[chk_test] + chk_value = {'Fan': chk_code, + 'Lamp': chk_code, + 'Temperature': chk_code, + 'Cover': chk_code, + 'Filter': chk_code, + 'Other': chk_code + } + + # WHEN: process_erst with status set to WARN + pjlink.process_erst('{fan}{lamp}{temp}{cover}{filter}{other}'.format(fan=chk_test, + lamp=chk_test, + temp=chk_test, + cover=chk_test, + filter=chk_test, + other=chk_test)) + + # PJLink instance errors should match chk_value + self.assertEqual(pjlink.projector_errors, chk_value, + 'projector_errors should have been set to all {err}'.format(err=chk_code)) + + def test_projector_process_erst_all_error(self): + """ + Test test_projector_process_erst_all_error + """ + # GIVEN: Test object + pjlink = pjlink_test + chk_test = ERST_ERR + chk_code = PJLINK_ERST_STATUS[chk_test] + chk_value = {'Fan': chk_code, + 'Lamp': chk_code, + 'Temperature': chk_code, + 'Cover': chk_code, + 'Filter': chk_code, + 'Other': chk_code + } + + # WHEN: process_erst with status set to ERROR + pjlink.process_erst('{fan}{lamp}{temp}{cover}{filter}{other}'.format(fan=chk_test, + lamp=chk_test, + temp=chk_test, + cover=chk_test, + filter=chk_test, + other=chk_test)) + + # PJLink instance errors should be set to chk_value + self.assertEqual(pjlink.projector_errors, chk_value, + 'projector_errors should have been set to all {err}'.format(err=chk_code)) + + def test_projector_process_inpt(self): + """ + Test input source status shows current input + """ + # GIVEN: Test object + pjlink = pjlink_test + pjlink.source = '0' + + # WHEN: Called with input source + pjlink.process_inpt('1') + + # 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): + """ + Test status lamp on/off and hours + """ + # GIVEN: Test object + pjlink = pjlink_test + + # WHEN: Call process_command with lamp data + pjlink.process_command('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(pjlink_test, 'projectorReceivedData') + def test_projector_process_lamp_multiple(self, mock_projectorReceivedData): + """ + Test status multiple lamp on/off and hours + """ + # GIVEN: Test object + pjlink = pjlink_test + + # WHEN: Call process_command with lamp data + pjlink.process_command('LAMP', '11111 1 22222 0 33333 1') + + # THEN: Lamp should have been set with proper lamp status + self.assertEqual(len(pjlink.lamp), 3, + 'Projector should have 3 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'], 11111, + 'Lamp 1 hours should have been set to 11111') + self.assertEqual(pjlink.lamp[1]['On'], False, + 'Lamp 2 power status should have been set to FALSE') + self.assertEqual(pjlink.lamp[1]['Hours'], 22222, + 'Lamp 2 hours should have been set to 22222') + self.assertEqual(pjlink.lamp[2]['On'], True, + 'Lamp 3 power status should have been set to TRUE') + self.assertEqual(pjlink.lamp[2]['Hours'], 33333, + 'Lamp 3 hours should have been set to 33333') + + @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_on(self, + mock_change_status, + mock_send_command, + mock_UpdateIcons, + mock_ReceivedData): + """ + Test status power to ON + """ + # GIVEN: Test object and preset + pjlink = pjlink_test + pjlink.power = S_STANDBY + + # WHEN: Call process_command with turn power on command + pjlink.process_command('POWR', PJLINK_POWR_STATUS[S_ON]) + + # 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') + 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_off(self, + mock_change_status, + mock_send_command, + mock_UpdateIcons, + mock_ReceivedData): + """ + Test status power to STANDBY + """ + # GIVEN: Test object and preset + pjlink = pjlink_test + pjlink.power = S_ON + + # WHEN: Call process_command with turn power on command + pjlink.process_command('POWR', PJLINK_POWR_STATUS[S_STANDBY]) + + # 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') def test_projector_process_rfil_save(self): """ @@ -150,419 +466,3 @@ class TestPJLink(TestCase): # THEN: Serial number should be set self.assertNotEquals(pjlink.serial_no, test_number, 'Projector serial number should NOT have been set') - - def test_projector_clss_one(self): - """ - Test class 1 sent from projector - """ - # GIVEN: Test object - pjlink = pjlink_test - - # WHEN: Process class response - pjlink.process_clss('1') - - # THEN: Projector class should be set to 1 - self.assertEqual(pjlink.pjlink_class, '1', - 'Projector should have returned class=1') - - def test_projector_clss_two(self): - """ - Test class 2 sent from projector - """ - # GIVEN: Test object - pjlink = pjlink_test - - # WHEN: Process class response - pjlink.process_clss('2') - - # THEN: Projector class should be set to 1 - self.assertEqual(pjlink.pjlink_class, '2', - 'Projector should have returned class=2') - - def test_bug_1550891_non_standard_class_reply(self): - """ - Bugfix 1550891: CLSS request returns non-standard reply - """ - # GIVEN: Test object - pjlink = pjlink_test - - # WHEN: Process non-standard reply - pjlink.process_clss('Class 1') - - # THEN: Projector class should be set with proper value - self.assertEqual(pjlink.pjlink_class, '1', - 'Non-standard class reply should have set class=1') - - @patch.object(pjlink_test, 'change_status') - def test_status_change(self, mock_change_status): - """ - Test process_command call with ERR2 (Parameter) status - """ - # GIVEN: Test object - pjlink = pjlink_test - - # WHEN: process_command is called with "ERR2" status from projector - pjlink.process_command('POWR', 'ERR2') - - # THEN: change_status should have called change_status with E_UNDEFINED - # as first parameter - mock_change_status.called_with(E_PARAMETER, - 'change_status should have been called with "{}"'.format( - ERROR_STRING[E_PARAMETER])) - - @patch.object(pjlink_test, 'process_inpt') - def test_projector_return_ok(self, mock_process_inpt): - """ - Test projector calls process_inpt command when process_command is called with INPT option - """ - # GIVEN: Test object - pjlink = pjlink_test - - # WHEN: process_command is called with INST command and 31 input: - pjlink.process_command('INPT', '31') - - # THEN: process_inpt method should have been called with 31 - mock_process_inpt.called_with('31', - "process_inpt should have been called with 31") - - @patch.object(pjlink_test, 'projectorReceivedData') - def test_projector_process_lamp(self, mock_projectorReceivedData): - """ - Test status lamp on/off and hours - """ - # GIVEN: Test object - pjlink = pjlink_test - - # WHEN: Call process_command with lamp data - pjlink.process_command('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(pjlink_test, 'projectorReceivedData') - def test_projector_process_multiple_lamp(self, mock_projectorReceivedData): - """ - Test status multiple lamp on/off and hours - """ - # GIVEN: Test object - pjlink = pjlink_test - - # WHEN: Call process_command with lamp data - pjlink.process_command('LAMP', '11111 1 22222 0 33333 1') - - # THEN: Lamp should have been set with proper lamp status - self.assertEqual(len(pjlink.lamp), 3, - 'Projector should have 3 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'], 11111, - 'Lamp 1 hours should have been set to 11111') - self.assertEqual(pjlink.lamp[1]['On'], False, - 'Lamp 2 power status should have been set to FALSE') - self.assertEqual(pjlink.lamp[1]['Hours'], 22222, - 'Lamp 2 hours should have been set to 22222') - self.assertEqual(pjlink.lamp[2]['On'], True, - 'Lamp 3 power status should have been set to TRUE') - self.assertEqual(pjlink.lamp[2]['Hours'], 33333, - 'Lamp 3 hours should have been set to 33333') - - @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_power_on(self, mock_change_status, - mock_send_command, - mock_UpdateIcons, - mock_ReceivedData): - """ - Test status power to ON - """ - # GIVEN: Test object and preset - pjlink = pjlink_test - pjlink.power = S_STANDBY - - # WHEN: Call process_command with turn power on command - pjlink.process_command('POWR', PJLINK_POWR_STATUS[S_ON]) - - # 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') - 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_power_off(self, mock_change_status, - mock_send_command, - mock_UpdateIcons, - mock_ReceivedData): - """ - Test status power to STANDBY - """ - # GIVEN: Test object and preset - pjlink = pjlink_test - pjlink.power = S_ON - - # WHEN: Call process_command with turn power on command - pjlink.process_command('POWR', PJLINK_POWR_STATUS[S_STANDBY]) - - # 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') - - @patch.object(pjlink_test, 'projectorUpdateIcons') - def test_projector_process_avmt_closed_unmuted(self, mock_projectorReceivedData): - """ - Test avmt status shutter closed and audio muted - """ - # GIVEN: Test object - pjlink = pjlink_test - pjlink.shutter = False - pjlink.mute = True - - # WHEN: Called with setting shutter closed and mute off - pjlink.process_avmt('11') - - # THEN: Shutter should be True and mute should be False - self.assertTrue(pjlink.shutter, 'Shutter should have been set to closed') - self.assertFalse(pjlink.mute, 'Audio should be off') - - @patch.object(pjlink_test, 'projectorUpdateIcons') - def test_projector_process_avmt_open_muted(self, mock_projectorReceivedData): - """ - Test avmt status shutter open and mute on - """ - # GIVEN: Test object - pjlink = pjlink_test - pjlink.shutter = True - pjlink.mute = False - - # WHEN: Called with setting shutter closed and mute on - pjlink.process_avmt('21') - - # THEN: Shutter should be closed and mute should be True - self.assertFalse(pjlink.shutter, 'Shutter should have been set to closed') - self.assertTrue(pjlink.mute, 'Audio should be off') - - @patch.object(pjlink_test, 'projectorUpdateIcons') - def test_projector_process_avmt_open_unmuted(self, mock_projectorReceivedData): - """ - Test avmt status shutter open and mute off off - """ - # GIVEN: Test object - pjlink = pjlink_test - pjlink.shutter = True - pjlink.mute = True - - # WHEN: Called with setting shutter to closed and mute on - pjlink.process_avmt('30') - - # THEN: Shutter should be closed and mute should be True - self.assertFalse(pjlink.shutter, 'Shutter should have been set to open') - self.assertFalse(pjlink.mute, 'Audio should be on') - - @patch.object(pjlink_test, 'projectorUpdateIcons') - def test_projector_process_avmt_closed_muted(self, mock_projectorReceivedData): - """ - Test avmt status shutter closed and mute off - """ - # GIVEN: Test object - pjlink = pjlink_test - pjlink.shutter = False - pjlink.mute = False - - # WHEN: Called with setting shutter to closed and mute on - pjlink.process_avmt('31') - - # THEN: Shutter should be closed and mute should be True - self.assertTrue(pjlink.shutter, 'Shutter should have been set to closed') - self.assertTrue(pjlink.mute, 'Audio should be on') - - def test_projector_process_input(self): - """ - Test input source status shows current input - """ - # GIVEN: Test object - pjlink = pjlink_test - pjlink.source = '0' - - # WHEN: Called with input source - pjlink.process_inpt('1') - - # THEN: Input selected should reflect current input - self.assertEqual(pjlink.source, '1', 'Input source should be set to "1"') - - 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 - pjlink.timer = MagicMock() - pjlink.socket_timer = MagicMock() - - # 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() - - # 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') - - @patch.object(pjlink_test, 'send_command') - @patch.object(pjlink_test, 'waitForReadyRead') - @patch.object(pjlink_test, 'projectorAuthentication') - @patch.object(pjlink_test, 'timer') - @patch.object(pjlink_test, 'socket_timer') - def test_bug_1593882_no_pin_authenticated_connection(self, mock_socket_timer, - mock_timer, - mock_authentication, - mock_ready_read, - mock_send_command): - """ - Test bug 1593882 no pin and authenticated request exception - """ - # GIVEN: Test object and mocks - pjlink = pjlink_test - pjlink.pin = None - mock_ready_read.return_value = True - - # WHEN: call with authentication request and pin not set - pjlink.check_login(data=TEST_CONNECT_AUTHENTICATE) - - # THEN: 'No Authentication' signal should have been sent - mock_authentication.emit.assert_called_with(pjlink.ip) - - @patch.object(pjlink_test, 'waitForReadyRead') - @patch.object(pjlink_test, 'state') - @patch.object(pjlink_test, '_send_command') - @patch.object(pjlink_test, 'timer') - @patch.object(pjlink_test, 'socket_timer') - def test_bug_1593883_pjlink_authentication(self, mock_socket_timer, - mock_timer, - mock_send_command, - mock_state, - mock_waitForReadyRead): - """ - Test bugfix 1593883 pjlink authentication - """ - # GIVEN: Test object and data - pjlink = pjlink_test - pjlink.pin = TEST_PIN - mock_state.return_value = pjlink.ConnectedState - mock_waitForReadyRead.return_value = True - - # WHEN: Athenticated connection is called - pjlink.check_login(data=TEST_CONNECT_AUTHENTICATE) - - # THEN: send_command should have the proper authentication - self.assertEqual("{test}".format(test=mock_send_command.call_args), - "call(data='{hash}%1CLSS ?\\r')".format(hash=TEST_HASH)) - - @patch.object(pjlink_test, 'disconnect_from_host') - def socket_abort_test(self, mock_disconnect): - """ - Test PJLink.socket_abort calls disconnect_from_host - """ - # GIVEN: Test object - pjlink = pjlink_test - - # WHEN: Calling socket_abort - pjlink.socket_abort() - - # THEN: disconnect_from_host should be called - self.assertTrue(mock_disconnect.called, 'Should have called disconnect_from_host') - - def poll_loop_not_connected_test(self): - """ - Test PJLink.poll_loop not connected return - """ - # GIVEN: Test object and mocks - pjlink = pjlink_test - pjlink.state = MagicMock() - pjlink.timer = MagicMock() - pjlink.state.return_value = False - pjlink.ConnectedState = True - - # WHEN: PJLink.poll_loop called - pjlink.poll_loop() - - # THEN: poll_loop should exit without calling any other method - self.assertFalse(pjlink.timer.called, 'Should have returned without calling any other method') - - @patch.object(pjlink_test, 'send_command') - def poll_loop_start_test(self, mock_send_command): - """ - Test PJLink.poll_loop makes correct calls - """ - # GIVEN: test object and test data - pjlink = pjlink_test - pjlink.state = MagicMock() - pjlink.timer = MagicMock() - pjlink.timer.interval = MagicMock() - pjlink.timer.setInterval = MagicMock() - pjlink.timer.start = MagicMock() - pjlink.poll_time = 20 - pjlink.power = S_ON - pjlink.source_available = None - pjlink.other_info = None - pjlink.manufacturer = None - pjlink.model = None - pjlink.pjlink_name = None - pjlink.ConnectedState = S_CONNECTED - pjlink.timer.interval.return_value = 10 - pjlink.state.return_value = S_CONNECTED - call_list = [ - call('POWR', queue=True), - call('ERST', queue=True), - call('LAMP', queue=True), - call('AVMT', queue=True), - call('INPT', queue=True), - call('INST', queue=True), - call('INFO', queue=True), - call('INF1', queue=True), - call('INF2', queue=True), - call('NAME', queue=True), - ] - - # WHEN: PJLink.poll_loop is called - pjlink.poll_loop() - - # THEN: proper calls were made to retrieve projector data - # First, call to update the timer with the next interval - self.assertTrue(pjlink.timer.setInterval.called, 'Should have updated the timer') - # Next, should have called the timer to start - self.assertTrue(pjlink.timer.start.called, 'Should have started the timer') - # Finally, should have called send_command with a list of projetctor status checks - mock_send_command.assert_has_calls(call_list, 'Should have queued projector queries')