diff --git a/openlp/core/projectors/db.py b/openlp/core/projectors/db.py index a2f060d75..db8f4ac4b 100644 --- a/openlp/core/projectors/db.py +++ b/openlp/core/projectors/db.py @@ -152,6 +152,7 @@ class Projector(Base, CommonMixin): location: Column(String(30)) notes: Column(String(200)) pjlink_name: Column(String(128)) # From projector + pjlink_class Column(String(1)) # From projector manufacturer: Column(String(128)) # From projector model: Column(String(128)) # From projector other: Column(String(128)) # From projector @@ -168,7 +169,7 @@ class Projector(Base, CommonMixin): Return basic representation of Source table entry. """ return '< Projector(id="{data}", ip="{ip}", port="{port}", mac_adx="{mac}", pin="{pin}", name="{name}", ' \ - 'location="{location}", notes="{notes}", pjlink_name="{pjlink_name}", ' \ + 'location="{location}", notes="{notes}", pjlink_name="{pjlink_name}", pjlink_class="{pjlink_class}", ' \ 'manufacturer="{manufacturer}", model="{model}", serial_no="{serial}", other="{other}", ' \ 'sources="{sources}", source_list="{source_list}", model_filter="{mfilter}", ' \ 'model_lamp="{mlamp}", sw_version="{sw_ver}") >'.format(data=self.id, @@ -180,6 +181,7 @@ class Projector(Base, CommonMixin): location=self.location, notes=self.notes, pjlink_name=self.pjlink_name, + pjlink_class=self.pjlink_class, manufacturer=self.manufacturer, model=self.model, other=self.other, @@ -197,6 +199,7 @@ class Projector(Base, CommonMixin): location = Column(String(30)) notes = Column(String(200)) pjlink_name = Column(String(128)) + pjlink_class = Column(String(1)) manufacturer = Column(String(128)) model = Column(String(128)) other = Column(String(128)) diff --git a/openlp/core/projectors/manager.py b/openlp/core/projectors/manager.py index 3144f7274..3bc7373fa 100644 --- a/openlp/core/projectors/manager.py +++ b/openlp/core/projectors/manager.py @@ -522,8 +522,8 @@ class ProjectorManager(QtWidgets.QWidget, RegistryBase, UiProjectorManager, LogM except (AttributeError, TypeError): pass try: - projector.timer.stop() - projector.timer.timeout.disconnect(projector.link.poll_loop) + projector.poll_timer.stop() + projector.poll_timer.timeout.disconnect(projector.link.poll_loop) except (AttributeError, TypeError): pass try: @@ -531,7 +531,6 @@ class ProjectorManager(QtWidgets.QWidget, RegistryBase, UiProjectorManager, LogM projector.socket_timer.timeout.disconnect(projector.link.socket_abort) except (AttributeError, TypeError): pass - projector.thread.quit() new_list = [] for item in self.projector_list: if item.link.db_item.id == projector.link.db_item.id: @@ -733,39 +732,18 @@ class ProjectorManager(QtWidgets.QWidget, RegistryBase, UiProjectorManager, LogM """ item = ProjectorItem(link=self._add_projector(projector)) item.db_item = projector - icon = QtGui.QIcon(QtGui.QPixmap(STATUS_ICONS[S_NOT_CONNECTED])) - item.icon = icon - widget = QtWidgets.QListWidgetItem(icon, + item.icon = QtGui.QIcon(QtGui.QPixmap(STATUS_ICONS[S_NOT_CONNECTED])) + widget = QtWidgets.QListWidgetItem(item.icon, item.link.name, self.projector_list_widget ) widget.setData(QtCore.Qt.UserRole, item) item.link.db_item = item.db_item item.widget = widget - thread = QtCore.QThread(parent=self) - thread.my_parent = self - item.moveToThread(thread) - thread.started.connect(item.link.thread_started) - thread.finished.connect(item.link.thread_stopped) - thread.finished.connect(thread.deleteLater) item.link.changeStatus.connect(self.update_status) item.link.projectorAuthentication.connect(self.authentication_error) item.link.projectorNoAuthentication.connect(self.no_authentication_error) item.link.projectorUpdateIcons.connect(self.update_icons) - timer = QtCore.QTimer(self) - timer.setInterval(self.poll_time) - timer.timeout.connect(item.link.poll_loop) - item.timer = timer - # Timeout in case of brain-dead projectors or cable disconnected - socket_timer = QtCore.QTimer(self) - socket_timer.setInterval(11000) - socket_timer.timeout.connect(item.link.socket_abort) - item.socket_timer = socket_timer - thread.start() - item.thread = thread - item.link.timer = timer - item.link.socket_timer = socket_timer - item.link.widget = item.widget self.projector_list.append(item) if start: item.link.connect_to_host() diff --git a/openlp/core/projectors/pjlink.py b/openlp/core/projectors/pjlink.py index 3e3bb3d2f..3517fb1e4 100644 --- a/openlp/core/projectors/pjlink.py +++ b/openlp/core/projectors/pjlink.py @@ -58,8 +58,8 @@ from openlp.core.projectors.constants import CONNECTION_ERRORS, PJLINK_CLASS, PJ PJLINK_ERST_DATA, PJLINK_ERST_STATUS, PJLINK_MAX_PACKET, PJLINK_PREFIX, PJLINK_PORT, PJLINK_POWR_STATUS, \ PJLINK_SUFFIX, PJLINK_VALID_CMD, PROJECTOR_STATE, STATUS_CODE, STATUS_MSG, QSOCKET_STATE, \ E_AUTHENTICATION, E_CONNECTION_REFUSED, E_GENERAL, E_INVALID_DATA, E_NETWORK, E_NOT_CONNECTED, \ - E_OK, E_SOCKET_TIMEOUT, \ - S_CONNECTED, S_CONNECTING, S_INFO, S_NOT_CONNECTED, S_OFF, S_OK, S_ON, S_STATUS + E_SOCKET_TIMEOUT, \ + S_CONNECTED, S_CONNECTING, S_NOT_CONNECTED, S_OFF, S_OK, S_ON log = logging.getLogger(__name__) log.debug('pjlink loaded') @@ -121,8 +121,6 @@ class PJLinkCommands(object): 'LAMP': self.process_lamp, 'NAME': self.process_name, 'PJLINK': self.process_pjlink, - # TODO: Part of check_login refactor - remove when done - # 'PJLINK': self.check_login, 'POWR': self.process_powr, 'SNUM': self.process_snum, 'SVER': self.process_sver, @@ -135,7 +133,7 @@ class PJLinkCommands(object): Initialize instance variables. Also used to reset projector-specific information to default. """ conn_state = STATUS_CODE[QSOCKET_STATE[self.state()]] - log.debug('({ip}) reset_information() connect status is {state}'.format(ip=self.ip, + log.debug('({ip}) reset_information() connect status is {state}'.format(ip=self.entry.name, state=conn_state)) self.fan = None # ERST self.filter_time = None # FILT @@ -147,7 +145,6 @@ class PJLinkCommands(object): 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 @@ -158,11 +155,11 @@ class PJLinkCommands(object): 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() + if hasattr(self, 'poll_timer'): + log.debug('({ip}): Calling poll_timer.stop()'.format(ip=self.entry.name)) + self.poll_timer.stop() if hasattr(self, 'socket_timer'): - log.debug('({ip}): Calling socket_timer.stop()'.format(ip=self.ip)) + log.debug('({ip}): Calling socket_timer.stop()'.format(ip=self.entry.name)) self.socket_timer.stop() self.send_busy = False self.send_queue = [] @@ -175,7 +172,7 @@ class PJLinkCommands(object): :param cmd: Command to process :param data: Data being processed """ - log.debug('({ip}) Processing command "{cmd}" with data "{data}"'.format(ip=self.ip, + log.debug('({ip}) Processing command "{cmd}" with data "{data}"'.format(ip=self.entry.name, cmd=cmd, data=data)) # cmd should already be in uppercase, but data may be in mixed-case. @@ -183,15 +180,15 @@ class PJLinkCommands(object): _data = data.upper() # Check if we have a future command not available yet if cmd not in self.pjlink_functions: - log.warning('({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.entry.name, cmd=cmd)) return elif _data == 'OK': - log.debug('({ip}) Command "{cmd}" returned OK'.format(ip=self.ip, cmd=cmd)) + log.debug('({ip}) Command "{cmd}" returned OK'.format(ip=self.entry.name, cmd=cmd)) # A command returned successfully, so do a query on command to verify status return self.send_command(cmd=cmd) elif _data in PJLINK_ERRORS: # Oops - projector error - log.error('({ip}) {cmd}: {err}'.format(ip=self.ip, + log.error('({ip}) {cmd}: {err}'.format(ip=self.entry.name, cmd=cmd, err=STATUS_MSG[PJLINK_ERRORS[_data]])) if PJLINK_ERRORS[_data] == E_AUTHENTICATION: @@ -199,7 +196,7 @@ class PJLinkCommands(object): self.projectorAuthentication.emit(self.name) return self.change_status(status=E_AUTHENTICATION) # Command checks already passed - log.debug('({ip}) Calling function for {cmd}'.format(ip=self.ip, cmd=cmd)) + log.debug('({ip}) Calling function for {cmd}'.format(ip=self.entry.name, cmd=cmd)) self.pjlink_functions[cmd](data=data) def process_avmt(self, data): @@ -219,7 +216,7 @@ class PJLinkCommands(object): '31': {'shutter': True, 'mute': True} } if data not in settings: - log.warning('({ip}) Invalid shutter response: {data}'.format(ip=self.ip, data=data)) + log.warning('({ip}) Invalid shutter response: {data}'.format(ip=self.entry.name, data=data)) return shutter = settings[data]['shutter'] mute = settings[data]['mute'] @@ -243,7 +240,7 @@ class PJLinkCommands(object): # : Received: '%1CLSS=Class 1' (Optoma) # : Received: '%1CLSS=Version1' (BenQ) if len(data) > 1: - log.warning('({ip}) Non-standard CLSS reply: "{data}"'.format(ip=self.ip, data=data)) + log.warning('({ip}) Non-standard CLSS reply: "{data}"'.format(ip=self.entry.name, 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. @@ -251,17 +248,23 @@ class PJLinkCommands(object): clss = re.findall('\d', data)[0] # Should only be the first match except IndexError: log.error('({ip}) No numbers found in class version reply "{data}" - ' - 'defaulting to class "1"'.format(ip=self.ip, data=data)) + 'defaulting to class "1"'.format(ip=self.entry.name, data=data)) clss = '1' elif not data.isdigit(): log.error('({ip}) NAN CLSS version reply "{data}" - ' - 'defaulting to class "1"'.format(ip=self.ip, data=data)) + 'defaulting to class "1"'.format(ip=self.entry.name, data=data)) clss = '1' else: clss = data self.pjlink_class = clss - log.debug('({ip}) Setting pjlink_class for this projector to "{data}"'.format(ip=self.ip, + log.debug('({ip}) Setting pjlink_class for this projector to "{data}"'.format(ip=self.entry.name, data=self.pjlink_class)) + # Since we call this one on first connect, setup polling from here + if not self.no_poll: + log.debug('({ip}) process_pjlink(): Starting timer'.format(ip=self.entry.name)) + self.poll_timer.setInterval(1000) # Set 1 second for initial information + self.poll_timer.start() + return def process_erst(self, data): @@ -273,15 +276,15 @@ class PJLinkCommands(object): """ if len(data) != PJLINK_ERST_DATA['DATA_LENGTH']: count = PJLINK_ERST_DATA['DATA_LENGTH'] - log.warning('{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.entry.name, + data=data, + count=count)) return try: datacheck = int(data) except ValueError: # Bad data - ignore - log.warning('({ip}) Invalid error status response "{data}"'.format(ip=self.ip, data=data)) + log.warning('({ip}) Invalid error status response "{data}"'.format(ip=self.entry.name, data=data)) return if datacheck == 0: self.projector_errors = None @@ -323,7 +326,8 @@ class PJLinkCommands(object): :param data: Projector manufacturer """ self.manufacturer = data - log.debug('({ip}) Setting projector manufacturer data to "{data}"'.format(ip=self.ip, data=self.manufacturer)) + log.debug('({ip}) Setting projector manufacturer data to "{data}"'.format(ip=self.entry.name, + data=self.manufacturer)) return def process_inf2(self, data): @@ -334,7 +338,7 @@ class PJLinkCommands(object): :param data: Model name """ self.model = data - log.debug('({ip}) Setting projector model to "{data}"'.format(ip=self.ip, data=self.model)) + log.debug('({ip}) Setting projector model to "{data}"'.format(ip=self.entry.name, data=self.model)) return def process_info(self, data): @@ -345,7 +349,7 @@ class PJLinkCommands(object): :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)) + log.debug('({ip}) Setting projector other_info to "{data}"'.format(ip=self.entry.name, data=self.other_info)) return def process_inpt(self, data): @@ -359,14 +363,15 @@ class PJLinkCommands(object): if self.source_available is not None: # We have available inputs, so verify it's in the list if data not in self.source_available: - log.warn('({ip}) Input source not listed in available sources - ignoring'.format(ip=self.ip)) + log.warn('({ip}) Input source not listed in available sources - ignoring'.format(ip=self.entry.name)) return elif data not in PJLINK_DEFAULT_CODES: # Hmm - no sources available yet, so check with PJLink defaults - log.warn('({ip}) Input source not listed as a PJLink available source - ignoring'.format(ip=self.ip)) + log.warn('({ip}) Input source not listed as a PJLink available source ' + '- ignoring'.format(ip=self.entry.name)) return self.source = data - log.debug('({ip}) Setting data source to "{data}"'.format(ip=self.ip, data=self.source)) + log.debug('({ip}) Setting data source to "{data}"'.format(ip=self.entry.name, data=self.source)) return def process_inst(self, data): @@ -382,7 +387,7 @@ class PJLinkCommands(object): sources.append(source) sources.sort() self.source_available = sources - log.debug('({ip}) Setting projector sources_available to "{data}"'.format(ip=self.ip, + log.debug('({ip}) Setting projector sources_available to "{data}"'.format(ip=self.entry.name, data=self.source_available)) return @@ -404,7 +409,7 @@ class PJLinkCommands(object): fill = {'Hours': int(lamp_list[0]), 'On': False if lamp_list[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)) + log.warning('({ip}) process_lamp(): Invalid data "{data}"'.format(ip=self.entry.name, data=data)) return lamps.append(fill) lamp_list.pop(0) # Remove lamp hours @@ -420,7 +425,7 @@ class PJLinkCommands(object): :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)) + log.debug('({ip}) Setting projector PJLink name to "{data}"'.format(ip=self.entry.name, data=self.pjlink_name)) return def process_pjlink(self, data): @@ -429,42 +434,38 @@ class PJLinkCommands(object): :param data: Initial packet with authentication scheme """ - log.debug('({ip}) Processing PJLINK command'.format(ip=self.ip)) + log.debug('({ip}) Processing PJLINK command'.format(ip=self.entry.name)) chk = data.split(' ') if len(chk[0]) != 1: # Invalid - after splitting, first field should be 1 character, either '0' or '1' only - log.error('({ip}) Invalid initial authentication scheme - aborting'.format(ip=self.ip)) + log.error('({ip}) Invalid initial authentication scheme - aborting'.format(ip=self.entry.name)) return self.disconnect_from_host() elif chk[0] == '0': # Normal connection no authentication if len(chk) > 1: # Invalid data - there should be nothing after a normal authentication scheme - log.error('({ip}) Normal connection with extra information - aborting'.format(ip=self.ip)) + log.error('({ip}) Normal connection with extra information - aborting'.format(ip=self.entry.name)) return self.disconnect_from_host() elif self.pin: - log.error('({ip}) Normal connection but PIN set - aborting'.format(ip=self.ip)) + log.error('({ip}) Normal connection but PIN set - aborting'.format(ip=self.entry.name)) return self.disconnect_from_host() else: data_hash = None elif chk[0] == '1': if len(chk) < 2: # Not enough information for authenticated connection - log.error('({ip}) Authenticated connection but not enough info - aborting'.format(ip=self.ip)) + log.error('({ip}) Authenticated connection but not enough info - aborting'.format(ip=self.entry.name)) return self.disconnect_from_host() elif not self.pin: - log.error('({ip}) Authenticate connection but no PIN - aborting'.format(ip=self.ip)) + log.error('({ip}) Authenticate connection but no PIN - aborting'.format(ip=self.entry.name)) return self.disconnect_from_host() else: data_hash = str(qmd5_hash(salt=chk[1].encode('utf-8'), data=self.pin.encode('utf-8')), encoding='ascii') # Passed basic checks, so start connection self.readyRead.connect(self.get_socket) - if not self.no_poll: - log.debug('({ip}) process_pjlink(): Starting timer'.format(ip=self.ip)) - self.timer.setInterval(2000) # Set 2 seconds for initial information - self.timer.start() self.change_status(S_CONNECTED) - log.debug('({ip}) process_pjlink(): Sending "CLSS" initial command'.format(ip=self.ip)) + log.debug('({ip}) process_pjlink(): Sending "CLSS" initial command'.format(ip=self.entry.name)) # Since this is an initial connection, make it a priority just in case return self.send_command(cmd="CLSS", salt=data_hash, priority=True) @@ -475,7 +476,7 @@ class PJLinkCommands(object): :param data: Power status """ - log.debug('({ip}: Processing POWR command'.format(ip=self.ip)) + log.debug('({ip}: Processing POWR command'.format(ip=self.entry.name)) if data in PJLINK_POWR_STATUS: power = PJLINK_POWR_STATUS[data] update_icons = self.power != power @@ -488,7 +489,7 @@ class PJLinkCommands(object): self.send_command('INST') else: # Log unknown status response - log.warning('({ip}) Unknown power response: "{data}"'.format(ip=self.ip, data=data)) + log.warning('({ip}) Unknown power response: "{data}"'.format(ip=self.entry.name, data=data)) return def process_rfil(self, data): @@ -498,9 +499,9 @@ class PJLinkCommands(object): if self.model_filter is None: self.model_filter = data else: - 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)) + log.warning('({ip}) Filter model already set'.format(ip=self.entry.name)) + log.warning('({ip}) Saved model: "{old}"'.format(ip=self.entry.name, old=self.model_filter)) + log.warning('({ip}) New model: "{new}"'.format(ip=self.entry.name, new=data)) def process_rlmp(self, data): """ @@ -509,9 +510,9 @@ class PJLinkCommands(object): if self.model_lamp is None: self.model_lamp = data else: - 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)) + log.warning('({ip}) Lamp model already set'.format(ip=self.entry.name)) + log.warning('({ip}) Saved lamp: "{old}"'.format(ip=self.entry.name, old=self.model_lamp)) + log.warning('({ip}) New lamp: "{new}"'.format(ip=self.entry.name, new=data)) def process_snum(self, data): """ @@ -520,16 +521,17 @@ class PJLinkCommands(object): :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)) + log.debug('({ip}) Setting projector serial number to "{data}"'.format(ip=self.entry.name, 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.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)) + log.warning('({ip}) Projector serial number does not match saved serial ' + 'number'.format(ip=self.entry.name)) + log.warning('({ip}) Saved: "{old}"'.format(ip=self.entry.name, old=self.serial_no)) + log.warning('({ip}) Received: "{new}"'.format(ip=self.entry.name, new=data)) + log.warning('({ip}) NOT saving serial number'.format(ip=self.entry.name)) self.serial_no_received = data def process_sver(self, data): @@ -541,14 +543,14 @@ class PJLinkCommands(object): 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)) + log.debug('({ip}) Setting projector software version to "{data}"'.format(ip=self.entry.name, data=data)) else: if self.sw_version != data: 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}) Updating software version'.format(ip=self.ip)) + 'software version'.format(ip=self.entry.name)) + log.warning('({ip}) Saved: "{old}"'.format(ip=self.entry.name, old=self.sw_version)) + log.warning('({ip}) Received: "{new}"'.format(ip=self.entry.name, new=data)) + log.warning('({ip}) Updating software version'.format(ip=self.entry.name)) self.sw_version = data self.db_update = True @@ -588,14 +590,14 @@ class PJLink(QtNetwork.QTcpSocket, PJLinkCommands): self.notes = self.entry.notes self.pin = self.entry.pin self.port = self.entry.port + self.pjlink_class = PJLINK_CLASS if self.entry.pjlink_class is None else self.entry.pjlink_class 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 + # Socket timeout (in case of brain-dead projectors) 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 @@ -605,61 +607,29 @@ class PJLink(QtNetwork.QTcpSocket, PJLinkCommands): 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.priority_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): - """ - Connects signals to methods when thread is started. - """ - log.debug('({ip}) Thread starting'.format(ip=self.ip)) - self.i_am_running = True + # Poll timer for status updates + self.poll_timer = QtCore.QTimer(self) # Timer that calls the poll_loop + self.poll_timer.setInterval(self.poll_time) + self.poll_timer.timeout.connect(self.poll_loop) + # Socket timer for some possible brain-dead projectors or network issues + self.socket_timer = QtCore.QTimer(self) + self.socket_timer.setInterval(self.socket_timeout) + self.socket_timer.timeout.connect(self.socket_abort) + # Socket status signals self.connected.connect(self.check_login) self.disconnected.connect(self.disconnect_from_host) self.error.connect(self.get_error) self.projectorReceivedData.connect(self._send_command) - def thread_stopped(self): - """ - Cleanups when thread is stopped. - """ - log.debug('({ip}) Thread stopped'.format(ip=self.ip)) - try: - self.connected.disconnect(self.check_login) - except TypeError: - pass - try: - self.disconnected.disconnect(self.disconnect_from_host) - except TypeError: - pass - try: - self.error.disconnect(self.get_error) - except TypeError: - pass - try: - self.projectorReceivedData.disconnect(self._send_command) - except TypeError: - pass - try: - self.readyRead.disconnect(self.get_socket) # Set in process_pjlink - except TypeError: - pass - self.disconnect_from_host() - self.deleteLater() - self.i_am_running = False - def socket_abort(self): """ Aborts connection and closes socket in case of brain-dead projectors. Should normally be called by socket_timer(). """ - log.debug('({ip}) socket_abort() - Killing connection'.format(ip=self.ip)) + log.debug('({ip}) socket_abort() - Killing connection'.format(ip=self.entry.name)) self.disconnect_from_host(abort=True) def poll_loop(self): @@ -668,15 +638,11 @@ class PJLink(QtNetwork.QTcpSocket, PJLinkCommands): Normally called by timer(). """ if QSOCKET_STATE[self.state()] != S_CONNECTED: - log.warning('({ip}) poll_loop(): Not connected - returning'.format(ip=self.ip)) + log.warning('({ip}) poll_loop(): Not connected - returning'.format(ip=self.entry.name)) + # Stop timer just in case it's missed elsewhere + self.poll_timer.stop() return - log.debug('({ip}) poll_loop(): Updating projector status'.format(ip=self.ip)) - # Reset timer in case we were called from a set command - if self.timer.interval() < self.poll_time: - # Reset timer to 5 seconds - self.timer.setInterval(self.poll_time) - # Restart timer - self.timer.start() + log.debug('({ip}) poll_loop(): Updating projector status'.format(ip=self.entry.name)) # The following commands do not change, so only check them once # Call them first in case other functions rely on something here if self.power == S_ON and self.source_available is None: @@ -705,6 +671,8 @@ class PJLink(QtNetwork.QTcpSocket, PJLinkCommands): check_list.extend(['FILT', 'FREZ']) for command in check_list: self.send_command(command) + # Reset the poll_timer for normal operations in case of initial connection + self.poll_timer.setInterval(self.poll_time) def _get_status(self, status): """ @@ -732,11 +700,11 @@ class PJLink(QtNetwork.QTcpSocket, PJLinkCommands): """ if status in STATUS_CODE: log.debug('({ip}) Changing status to {status} ' - '"{msg}"'.format(ip=self.ip, + '"{msg}"'.format(ip=self.entry.name, status=STATUS_CODE[status], msg=msg if msg is not None else STATUS_MSG[status])) else: - log.warning('({ip}) Unknown status change code: {code}'.format(ip=self.ip, + log.warning('({ip}) Unknown status change code: {code}'.format(ip=self.entry.name, code=status)) return if status in CONNECTION_ERRORS: @@ -752,15 +720,15 @@ class PJLink(QtNetwork.QTcpSocket, PJLinkCommands): # These log entries are for troubleshooting only (status_code, status_message) = self._get_status(self.status_connect) - log.debug('({ip}) status_connect: {code}: "{message}"'.format(ip=self.ip, + log.debug('({ip}) status_connect: {code}: "{message}"'.format(ip=self.entry.name, code=status_code, message=status_message if msg is None else msg)) (status_code, status_message) = self._get_status(self.projector_status) - log.debug('({ip}) projector_status: {code}: "{message}"'.format(ip=self.ip, + log.debug('({ip}) projector_status: {code}: "{message}"'.format(ip=self.entry.name, code=status_code, message=status_message if msg is None else msg)) (status_code, status_message) = self._get_status(self.error_status) - log.debug('({ip}) error_status: {code}: "{message}"'.format(ip=self.ip, + log.debug('({ip}) error_status: {code}: "{message}"'.format(ip=self.entry.name, code=status_code, message=status_message if msg is None else msg)) @@ -782,27 +750,27 @@ class PJLink(QtNetwork.QTcpSocket, PJLinkCommands): :param data: Optional data if called from another routine """ - log.debug('({ip}) check_login(data="{data}")'.format(ip=self.ip, data=data)) + log.debug('({ip}) check_login(data="{data}")'.format(ip=self.entry.name, data=data)) if data is None: # Reconnected setup? if not self.waitForReadyRead(2000): # Possible timeout issue - log.error('({ip}) Socket timeout waiting for login'.format(ip=self.ip)) + log.error('({ip}) Socket timeout waiting for login'.format(ip=self.entry.name)) self.change_status(E_SOCKET_TIMEOUT) return read = self.readLine(self.max_size) self.readLine(self.max_size) # Clean out any trailing whitespace if read is None: - log.warning('({ip}) read is None - socket error?'.format(ip=self.ip)) + log.warning('({ip}) read is None - socket error?'.format(ip=self.entry.name)) return elif len(read) < 8: - log.warning('({ip}) Not enough data read - skipping'.format(ip=self.ip)) + log.warning('({ip}) Not enough data read - skipping'.format(ip=self.entry.name)) return data = decode(read, 'utf-8') # Possibility of extraneous data on input when reading. # Clean out extraneous characters in buffer. self.read(1024) - log.debug('({ip}) check_login() read "{data}"'.format(ip=self.ip, data=data.strip())) + log.debug('({ip}) check_login() read "{data}"'.format(ip=self.entry.name, data=data.strip())) # At this point, we should only have the initial login prompt with # possible authentication # PJLink initial login will be: @@ -810,76 +778,26 @@ class PJLink(QtNetwork.QTcpSocket, PJLinkCommands): # 'PJLink 1 XXXXXX' Authenticated login - extra processing required. if not data.startswith('PJLINK'): # Invalid initial packet - close socket - log.error('({ip}) Invalid initial packet received - closing socket'.format(ip=self.ip)) + log.error('({ip}) Invalid initial packet received - closing socket'.format(ip=self.entry.name)) return self.disconnect_from_host() # Convert the initial login prompt with the expected PJLink normal command format for processing - log.debug('({ip}) check_login(): Formatting initial connection prompt to PJLink packet'.format(ip=self.ip)) + log.debug('({ip}) check_login(): Formatting initial connection prompt' + 'to PJLink packet'.format(ip=self.entry.name)) return self.get_data('{start}{clss}{data}'.format(start=PJLINK_PREFIX, clss='1', data=data.replace(' ', '=', 1)).encode('utf-8')) - # TODO: The below is replaced by process_pjlink() - remove when working properly - """ - if '=' in data: - # Processing a login reply - data_check = data.strip().split('=') - else: - # Process initial connection - data_check = data.strip().split(' ') - log.debug('({ip}) data_check="{data}"'.format(ip=self.ip, data=data_check)) - # Check for projector reporting an error - if data_check[1].upper() == 'ERRA': - # Authentication error - self.disconnect_from_host() - self.change_status(E_AUTHENTICATION) - log.debug('({ip}) emitting projectorAuthentication() signal'.format(ip=self.ip)) - return - elif (data_check[1] == '0') and (self.pin): - # Pin set and no authentication needed - log.warning('({ip}) Regular connection but PIN set'.format(ip=self.name)) - self.disconnect_from_host() - self.change_status(E_AUTHENTICATION) - log.debug('({ip}) Emitting projectorNoAuthentication() signal'.format(ip=self.ip)) - self.projectorNoAuthentication.emit(self.name) - return - elif data_check[1] == '1': - # Authenticated login with salt - if not self.pin: - log.warning('({ip}) Authenticated connection but no pin set'.format(ip=self.ip)) - self.disconnect_from_host() - self.change_status(E_AUTHENTICATION) - log.debug('({ip}) Emitting projectorAuthentication() signal'.format(ip=self.ip)) - self.projectorAuthentication.emit(self.ip) - return - else: - log.debug('({ip}) Setting hash with salt="{data}"'.format(ip=self.ip, data=data_check[2])) - log.debug('({ip}) pin="{data}"'.format(ip=self.ip, data=self.pin)) - data_hash = str(qmd5_hash(salt=data_check[2].encode('utf-8'), data=self.pin.encode('utf-8')), - encoding='ascii') - else: - data_hash = None - # We're connected at this point, so go ahead and setup regular I/O - self.readyRead.connect(self.get_socket) - self.projectorReceivedData.connect(self._send_command) - # Initial data we should know about - self.send_command(cmd='CLSS', salt=data_hash) - self.waitForReadyRead() - if (not self.no_poll) and (self.state() == self.ConnectedState): - log.debug('({ip}) Starting timer'.format(ip=self.ip)) - self.timer.setInterval(2000) # Set 2 seconds for initial information - self.timer.start() - """ def _trash_buffer(self, msg=None): """ Clean out extraneous stuff in the buffer. """ - log.warning('({ip}) {message}'.format(ip=self.ip, message='Invalid packet' if msg is None else msg)) + log.warning('({ip}) {message}'.format(ip=self.entry.name, message='Invalid packet' if msg is None else msg)) self.send_busy = False trash_count = 0 while self.bytesAvailable() > 0: trash = self.read(self.max_size) trash_count += len(trash) - log.debug('({ip}) Finished cleaning buffer - {count} bytes dropped'.format(ip=self.ip, + log.debug('({ip}) Finished cleaning buffer - {count} bytes dropped'.format(ip=self.entry.name, count=trash_count)) return @@ -891,9 +809,9 @@ class PJLink(QtNetwork.QTcpSocket, PJLinkCommands): :param data: Data to process. buffer must be formatted as a proper PJLink packet. :param ip: Destination IP for buffer. """ - log.debug('({ip}) get_buffer(data="{buff}" ip="{ip_in}"'.format(ip=self.ip, buff=data, ip_in=ip)) + log.debug('({ip}) get_buffer(data="{buff}" ip="{ip_in}"'.format(ip=self.entry.name, buff=data, ip_in=ip)) if ip is None: - log.debug("({ip}) get_buffer() Don't know who data is for - exiting".format(ip=self.ip)) + log.debug("({ip}) get_buffer() Don't know who data is for - exiting".format(ip=self.entry.name)) return return self.get_data(buff=data, ip=ip) @@ -902,17 +820,17 @@ class PJLink(QtNetwork.QTcpSocket, PJLinkCommands): """ Get data from TCP socket. """ - log.debug('({ip}) get_socket(): Reading data'.format(ip=self.ip)) + log.debug('({ip}) get_socket(): Reading data'.format(ip=self.entry.name)) if QSOCKET_STATE[self.state()] != S_CONNECTED: - log.debug('({ip}) get_socket(): Not connected - returning'.format(ip=self.ip)) + log.debug('({ip}) get_socket(): Not connected - returning'.format(ip=self.entry.name)) self.send_busy = False return # Although we have a packet length limit, go ahead and use a larger buffer read = self.readLine(1024) - log.debug('({ip}) get_socket(): "{buff}"'.format(ip=self.ip, buff=read)) + log.debug('({ip}) get_socket(): "{buff}"'.format(ip=self.entry.name, buff=read)) if read == -1: # No data available - log.debug('({ip}) get_socket(): No data available (-1)'.format(ip=self.ip)) + log.debug('({ip}) get_socket(): No data available (-1)'.format(ip=self.entry.name)) return self.receive_data_signal() self.socket_timer.stop() return self.get_data(buff=read, ip=self.ip) @@ -928,7 +846,7 @@ class PJLink(QtNetwork.QTcpSocket, PJLinkCommands): # set to default here if ip is None: ip = self.ip - log.debug('({ip}) get_data(ip="{ip_in}" buffer="{buff}"'.format(ip=self.ip, ip_in=ip, buff=buff)) + log.debug('({ip}) get_data(ip="{ip_in}" buffer="{buff}"'.format(ip=self.entry.name, ip_in=ip, buff=buff)) # NOTE: Class2 has changed to some values being UTF-8 data_in = decode(buff, 'utf-8') data = data_in.strip() @@ -945,7 +863,7 @@ class PJLink(QtNetwork.QTcpSocket, PJLinkCommands): elif '=' not in data: self._trash_buffer(msg='get_data(): Invalid reply - Does not have "="') return self.receive_data_signal() - log.debug('({ip}) get_data(): Checking new data "{data}"'.format(ip=self.ip, data=data)) + log.debug('({ip}) get_data(): Checking new data "{data}"'.format(ip=self.entry.name, data=data)) header, data = data.split('=') # At this point, the header should contain: # "PVCCCC" @@ -957,16 +875,17 @@ class PJLink(QtNetwork.QTcpSocket, PJLinkCommands): version, cmd = header[1], header[2:].upper() except ValueError as e: self.change_status(E_INVALID_DATA) - log.warning('({ip}) get_data(): Received data: "{data}"'.format(ip=self.ip, data=data_in)) + log.warning('({ip}) get_data(): Received data: "{data}"'.format(ip=self.entry.name, data=data_in)) self._trash_buffer('get_data(): Expected header + command + data') return self.receive_data_signal() if cmd not in PJLINK_VALID_CMD: - log.warning('({ip}) get_data(): Invalid packet - unknown command "{data}"'.format(ip=self.ip, data=cmd)) + log.warning('({ip}) get_data(): Invalid packet - unknown command "{data}"'.format(ip=self.entry.name, + data=cmd)) self._trash_buffer(msg='get_data(): Unknown command "{data}"'.format(data=cmd)) return self.receive_data_signal() if int(self.pjlink_class) < int(version): log.warning('({ip}) get_data(): Projector returned class reply higher ' - 'than projector stated class'.format(ip=self.ip)) + 'than projector stated class'.format(ip=self.entry.name)) self.process_command(cmd, data) return self.receive_data_signal() @@ -978,12 +897,14 @@ class PJLink(QtNetwork.QTcpSocket, PJLinkCommands): :param err: Error code """ - log.debug('({ip}) get_error(err={error}): {data}'.format(ip=self.ip, error=err, data=self.errorString())) + log.debug('({ip}) get_error(err={error}): {data}'.format(ip=self.entry.name, + error=err, + data=self.errorString())) if err <= 18: # QSocket errors. Redefined in projector.constants so we don't mistake # them for system errors check = err + E_CONNECTION_REFUSED - self.timer.stop() + self.poll_timer.stop() else: check = err if check < E_GENERAL: @@ -1007,12 +928,12 @@ class PJLink(QtNetwork.QTcpSocket, PJLinkCommands): :param priority: Option to send packet now rather than queue it up """ if QSOCKET_STATE[self.state()] != S_CONNECTED: - log.warning('({ip}) send_command(): Not connected - returning'.format(ip=self.ip)) + log.warning('({ip}) send_command(): Not connected - returning'.format(ip=self.entry.name)) return self.reset_information() if cmd not in PJLINK_VALID_CMD: - log.error('({ip}) send_command(): Invalid command requested - ignoring.'.format(ip=self.ip)) + log.error('({ip}) send_command(): Invalid command requested - ignoring.'.format(ip=self.entry.name)) return - log.debug('({ip}) send_command(): Building cmd="{command}" opts="{data}"{salt}'.format(ip=self.ip, + log.debug('({ip}) send_command(): Building cmd="{command}" opts="{data}"{salt}'.format(ip=self.entry.name, command=cmd, data=opts, salt='' if salt is None @@ -1025,7 +946,7 @@ class PJLink(QtNetwork.QTcpSocket, PJLinkCommands): header = PJLINK_HEADER.format(linkclass=cmd_ver[0]) else: # NOTE: Once we get to version 3 then think about looping - log.error('({ip}): send_command(): PJLink class check issue? Aborting'.format(ip=self.ip)) + log.error('({ip}): send_command(): PJLink class check issue? Aborting'.format(ip=self.entry.name)) return out = '{salt}{header}{command} {options}{suffix}'.format(salt="" if salt is None else salt, header=header, @@ -1033,15 +954,15 @@ class PJLink(QtNetwork.QTcpSocket, PJLinkCommands): options=opts, suffix=PJLINK_SUFFIX) if out in self.priority_queue: - log.debug('({ip}) send_command(): Already in priority queue - skipping'.format(ip=self.ip)) + log.debug('({ip}) send_command(): Already in priority queue - skipping'.format(ip=self.entry.name)) elif out in self.send_queue: - log.debug('({ip}) send_command(): Already in normal queue - skipping'.format(ip=self.ip)) + log.debug('({ip}) send_command(): Already in normal queue - skipping'.format(ip=self.entry.name)) else: if priority: - log.debug('({ip}) send_command(): Adding to priority queue'.format(ip=self.ip)) + log.debug('({ip}) send_command(): Adding to priority queue'.format(ip=self.entry.name)) self.priority_queue.append(out) else: - log.debug('({ip}) send_command(): Adding to normal queue'.format(ip=self.ip)) + log.debug('({ip}) send_command(): Adding to normal queue'.format(ip=self.entry.name)) self.send_queue.append(out) if self.priority_queue or self.send_queue: # May be some initial connection setup so make sure we send data @@ -1056,44 +977,45 @@ class PJLink(QtNetwork.QTcpSocket, PJLinkCommands): :param utf8: Send as UTF-8 string otherwise send as ASCII string """ # Funny looking data check, but it's a quick check for data=None - log.debug('({ip}) _send_command(data="{data}")'.format(ip=self.ip, data=data.strip() if data else data)) + log.debug('({ip}) _send_command(data="{data}")'.format(ip=self.entry.name, data=data.strip() if data else data)) conn_state = STATUS_CODE[QSOCKET_STATE[self.state()]] - log.debug('({ip}) _send_command(): Connection status: {data}'.format(ip=self.ip, + log.debug('({ip}) _send_command(): Connection status: {data}'.format(ip=self.entry.name, data=conn_state)) if QSOCKET_STATE[self.state()] != S_CONNECTED: - log.debug('({ip}) _send_command() Not connected - abort'.format(ip=self.ip)) + log.debug('({ip}) _send_command() Not connected - abort'.format(ip=self.entry.name)) self.send_busy = False return self.disconnect_from_host() if data and data not in self.priority_queue: - log.debug('({ip}) _send_command(): Priority packet - adding to priority queue'.format(ip=self.ip)) + log.debug('({ip}) _send_command(): Priority packet - adding to priority queue'.format(ip=self.entry.name)) self.priority_queue.append(data) if self.send_busy: # Still waiting for response from last command sent - log.debug('({ip}) _send_command(): Still busy, returning'.format(ip=self.ip)) - log.debug('({ip}) _send_command(): Priority queue = {data}'.format(ip=self.ip, data=self.priority_queue)) - log.debug('({ip}) _send_command(): Normal queue = {data}'.format(ip=self.ip, data=self.send_queue)) + log.debug('({ip}) _send_command(): Still busy, returning'.format(ip=self.entry.name)) + log.debug('({ip}) _send_command(): Priority queue = {data}'.format(ip=self.entry.name, + data=self.priority_queue)) + log.debug('({ip}) _send_command(): Normal queue = {data}'.format(ip=self.entry.name, data=self.send_queue)) return if len(self.priority_queue) != 0: out = self.priority_queue.pop(0) - log.debug('({ip}) _send_command(): Getting priority queued packet'.format(ip=self.ip)) + log.debug('({ip}) _send_command(): Getting priority queued packet'.format(ip=self.entry.name)) elif len(self.send_queue) != 0: out = self.send_queue.pop(0) - log.debug('({ip}) _send_command(): Getting normal queued packet'.format(ip=self.ip)) + log.debug('({ip}) _send_command(): Getting normal queued packet'.format(ip=self.entry.name)) else: # No data to send - log.debug('({ip}) _send_command(): No data to send'.format(ip=self.ip)) + log.debug('({ip}) _send_command(): No data to send'.format(ip=self.entry.name)) self.send_busy = False return self.send_busy = True - log.debug('({ip}) _send_command(): Sending "{data}"'.format(ip=self.ip, data=out.strip())) + log.debug('({ip}) _send_command(): Sending "{data}"'.format(ip=self.entry.name, data=out.strip())) self.socket_timer.start() sent = self.write(out.encode('{string_encoding}'.format(string_encoding='utf-8' if utf8 else 'ascii'))) self.waitForBytesWritten(2000) # 2 seconds should be enough if sent == -1: # Network error? - log.warning('({ip}) _send_command(): -1 received - disconnecting from host'.format(ip=self.ip)) + log.warning('({ip}) _send_command(): -1 received - disconnecting from host'.format(ip=self.entry.name)) self.change_status(E_NETWORK, translate('OpenLP.PJLink', 'Error while sending data to projector')) self.disconnect_from_host() @@ -1102,9 +1024,9 @@ class PJLink(QtNetwork.QTcpSocket, PJLinkCommands): """ Initiate connection to projector. """ - log.debug('({ip}) connect_to_host(): Starting connection'.format(ip=self.ip)) + log.debug('({ip}) connect_to_host(): Starting connection'.format(ip=self.entry.name)) if QSOCKET_STATE[self.state()] == S_CONNECTED: - log.warning('({ip}) connect_to_host(): Already connected - returning'.format(ip=self.ip)) + log.warning('({ip}) connect_to_host(): Already connected - returning'.format(ip=self.entry.name)) return self.error_status = S_OK self.change_status(S_CONNECTING) @@ -1117,17 +1039,17 @@ class PJLink(QtNetwork.QTcpSocket, PJLinkCommands): """ if abort or QSOCKET_STATE[self.state()] != S_NOT_CONNECTED: if abort: - log.warning('({ip}) disconnect_from_host(): Aborting connection'.format(ip=self.ip)) + log.warning('({ip}) disconnect_from_host(): Aborting connection'.format(ip=self.entry.name)) self.abort() else: - log.warning('({ip}) disconnect_from_host(): Not connected'.format(ip=self.ip)) + log.warning('({ip}) disconnect_from_host(): Not connected'.format(ip=self.entry.name)) self.disconnectFromHost() try: self.readyRead.disconnect(self.get_socket) except TypeError: pass log.debug('({ip}) disconnect_from_host() ' - 'Current status {data}'.format(ip=self.ip, data=self._get_status(self.status_connect)[0])) + 'Current status {data}'.format(ip=self.entry.name, data=self._get_status(self.status_connect)[0])) if abort: self.change_status(E_NOT_CONNECTED) else: @@ -1138,70 +1060,70 @@ class PJLink(QtNetwork.QTcpSocket, PJLinkCommands): """ Send command to retrieve shutter status. """ - log.debug('({ip}) Sending AVMT command'.format(ip=self.ip)) + log.debug('({ip}) Sending AVMT command'.format(ip=self.entry.name)) return self.send_command(cmd='AVMT') def get_available_inputs(self): """ Send command to retrieve available source inputs. """ - log.debug('({ip}) Sending INST command'.format(ip=self.ip)) + log.debug('({ip}) Sending INST command'.format(ip=self.entry.name)) return self.send_command(cmd='INST') def get_error_status(self): """ Send command to retrieve currently known errors. """ - log.debug('({ip}) Sending ERST command'.format(ip=self.ip)) + log.debug('({ip}) Sending ERST command'.format(ip=self.entry.name)) return self.send_command(cmd='ERST') def get_input_source(self): """ Send command to retrieve currently selected source input. """ - log.debug('({ip}) Sending INPT command'.format(ip=self.ip)) + log.debug('({ip}) Sending INPT command'.format(ip=self.entry.name)) return self.send_command(cmd='INPT') def get_lamp_status(self): """ Send command to return the lap status. """ - log.debug('({ip}) Sending LAMP command'.format(ip=self.ip)) + log.debug('({ip}) Sending LAMP command'.format(ip=self.entry.name)) return self.send_command(cmd='LAMP') def get_manufacturer(self): """ Send command to retrieve manufacturer name. """ - log.debug('({ip}) Sending INF1 command'.format(ip=self.ip)) + log.debug('({ip}) Sending INF1 command'.format(ip=self.entry.name)) return self.send_command(cmd='INF1') def get_model(self): """ Send command to retrieve the model name. """ - log.debug('({ip}) Sending INF2 command'.format(ip=self.ip)) + log.debug('({ip}) Sending INF2 command'.format(ip=self.entry.name)) return self.send_command(cmd='INF2') def get_name(self): """ Send command to retrieve name as set by end-user (if set). """ - log.debug('({ip}) Sending NAME command'.format(ip=self.ip)) + log.debug('({ip}) Sending NAME command'.format(ip=self.entry.name)) return self.send_command(cmd='NAME') def get_other_info(self): """ Send command to retrieve extra info set by manufacturer. """ - log.debug('({ip}) Sending INFO command'.format(ip=self.ip)) + log.debug('({ip}) Sending INFO command'.format(ip=self.entry.name)) return self.send_command(cmd='INFO') def get_power_status(self): """ Send command to retrieve power status. """ - log.debug('({ip}) Sending POWR command'.format(ip=self.ip)) + log.debug('({ip}) Sending POWR command'.format(ip=self.entry.name)) return self.send_command(cmd='POWR') def set_input_source(self, src=None): @@ -1211,12 +1133,12 @@ class PJLink(QtNetwork.QTcpSocket, PJLinkCommands): :param src: Video source to select in projector """ - log.debug('({ip}) set_input_source(src="{data}")'.format(ip=self.ip, data=src)) + log.debug('({ip}) set_input_source(src="{data}")'.format(ip=self.entry.name, data=src)) if self.source_available is None: return elif src not in self.source_available: return - log.debug('({ip}) Setting input source to "{data}"'.format(ip=self.ip, data=src)) + log.debug('({ip}) Setting input source to "{data}"'.format(ip=self.entry.name, data=src)) self.send_command(cmd='INPT', opts=src, priority=True) self.poll_loop() @@ -1224,7 +1146,7 @@ class PJLink(QtNetwork.QTcpSocket, PJLinkCommands): """ Send command to turn power to on. """ - log.debug('({ip}) Setting POWR to 1 (on)'.format(ip=self.ip)) + log.debug('({ip}) Setting POWR to 1 (on)'.format(ip=self.entry.name)) self.send_command(cmd='POWR', opts='1', priority=True) self.poll_loop() @@ -1232,7 +1154,7 @@ class PJLink(QtNetwork.QTcpSocket, PJLinkCommands): """ Send command to turn power to standby. """ - log.debug('({ip}) Setting POWR to 0 (standby)'.format(ip=self.ip)) + log.debug('({ip}) Setting POWR to 0 (standby)'.format(ip=self.entry.name)) self.send_command(cmd='POWR', opts='0', priority=True) self.poll_loop() @@ -1240,7 +1162,7 @@ class PJLink(QtNetwork.QTcpSocket, PJLinkCommands): """ Send command to set shutter to closed position. """ - log.debug('({ip}) Setting AVMT to 11 (shutter closed)'.format(ip=self.ip)) + log.debug('({ip}) Setting AVMT to 11 (shutter closed)'.format(ip=self.entry.name)) self.send_command(cmd='AVMT', opts='11', priority=True) self.poll_loop() @@ -1248,7 +1170,7 @@ class PJLink(QtNetwork.QTcpSocket, PJLinkCommands): """ Send command to set shutter to open position. """ - log.debug('({ip}) Setting AVMT to "10" (shutter open)'.format(ip=self.ip)) + log.debug('({ip}) Setting AVMT to "10" (shutter open)'.format(ip=self.entry.name)) self.send_command(cmd='AVMT', opts='10', priority=True) self.poll_loop() self.projectorUpdateIcons.emit() diff --git a/openlp/core/projectors/upgrade.py b/openlp/core/projectors/upgrade.py index 1bcd756dc..b02667638 100644 --- a/openlp/core/projectors/upgrade.py +++ b/openlp/core/projectors/upgrade.py @@ -33,7 +33,7 @@ from openlp.core.lib.db import get_upgrade_op log = logging.getLogger(__name__) # Initial projector DB was unversioned -__version__ = 2 +__version__ = 3 log.debug('Projector DB upgrade module loading') @@ -71,3 +71,23 @@ def upgrade_2(session, metadata): 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())) log.debug('{status} projector DB upgrade to version 2'.format(status='Updated' if upgrade_db else 'Skipping')) + + +def upgrade_3(session, metadata): + """ + Version 3 upgrade. + + Update Projector() table to inlcude PJLink class as part of record. + + pjlink_version: Column(String(1)) + + :param session: DB Session instance + :param metadata: Metadata of current DB + """ + log.debug('Checking projector DB upgrade to version 3') + projector_table = Table('projector', metadata, autoload=True) + upgrade_db = 'pjlink_class' 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('pjlink_class', types.String(1), server_default=null())) + log.debug('{status} projector DB upgrade to version 3'.format(status='Updated' if upgrade_db else 'Skipping')) diff --git a/tests/functional/openlp_core/projectors/test_projector_bugfixes_01.py b/tests/functional/openlp_core/projectors/test_projector_bugfixes_01.py index 8cceb6ef6..0f5307b53 100644 --- a/tests/functional/openlp_core/projectors/test_projector_bugfixes_01.py +++ b/tests/functional/openlp_core/projectors/test_projector_bugfixes_01.py @@ -24,57 +24,32 @@ Package to test the openlp.core.projectors.pjlink base package. """ from unittest import TestCase -from openlp.core.projectors.db import Projector -from openlp.core.projectors.pjlink import PJLink -from tests.resources.projector.data import TEST1_DATA +# from openlp.core.projectors.db import Projector +# from openlp.core.projectors.pjlink import PJLink +# from tests.resources.projector.data import TEST1_DATA class TestPJLinkBugs(TestCase): """ Tests for the PJLink module bugfixes """ - def setUp(self): - ''' - Initialization - ''' - self.pjlink_test = PJLink(Projector(**TEST1_DATA), no_poll=True) - - def tearDown(self): - ''' - Cleanups - ''' - self.pjlink_test = None - - def test_bug_1550891_process_clss_nonstandard_reply_1(self): + def bug_1550891_process_clss_nonstandard_reply_1(self): """ Bugfix 1550891: CLSS request returns non-standard reply with Optoma/Viewsonic projector """ - # GIVEN: Test object - pjlink = self.pjlink_test + # Test now part of test_projector_pjlink_commands_01 + # Keeping here for bug reference + pass - # 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_bug_1550891_process_clss_nonstandard_reply_2(self): + def bug_1550891_process_clss_nonstandard_reply_2(self): """ Bugfix 1550891: CLSS request returns non-standard reply with BenQ projector """ - # GIVEN: Test object - pjlink = self.pjlink_test + # Test now part of test_projector_pjlink_commands_01 + # Keeping here for bug reference + pass - # WHEN: Process non-standard reply - pjlink.process_clss('Version2') - - # THEN: Projector class should be set with proper value - # NOTE: At this time BenQ is Class 1, but we're trying a different value to verify - self.assertEqual(pjlink.pjlink_class, '2', - 'Non-standard class reply should have set class=2') - - def test_bug_1593882_no_pin_authenticated_connection(self): + def bug_1593882_no_pin_authenticated_connection(self): """ Test bug 1593882 no pin and authenticated request exception """ @@ -82,7 +57,7 @@ class TestPJLinkBugs(TestCase): # Keeping here for bug reference pass - def test_bug_1593883_pjlink_authentication(self): + def bug_1593883_pjlink_authentication(self): """ Test bugfix 1593883 pjlink authentication and ticket 92187 """ @@ -90,17 +65,10 @@ class TestPJLinkBugs(TestCase): # Keeping here for bug reference pass - def test_bug_1734275_process_lamp_nonstandard_reply(self): + def bug_1734275_process_lamp_nonstandard_reply(self): """ - Test bugfix 17342785 non-standard LAMP response + Test bugfix 17342785 non-standard LAMP response with one lamp hours only """ - # GIVEN: Test object - pjlink = self.pjlink_test - - # WHEN: Process lamp command called with only hours and no lamp power state - pjlink.process_lamp("45") - - # THEN: Lamp should show hours as 45 and lamp power as Unavailable - self.assertEqual(len(pjlink.lamp), 1, 'There should only be 1 lamp available') - self.assertEqual(pjlink.lamp[0]['Hours'], 45, 'Lamp hours should have equalled 45') - self.assertIsNone(pjlink.lamp[0]['On'], 'Lamp power should be "None"') + # Test now part of test_projector_pjlink_commands_01 + # Keeping here for bug reference + pass diff --git a/tests/functional/openlp_core/projectors/test_projector_db.py b/tests/functional/openlp_core/projectors/test_projector_db.py index 619f16bf2..81bd3f006 100644 --- a/tests/functional/openlp_core/projectors/test_projector_db.py +++ b/tests/functional/openlp_core/projectors/test_projector_db.py @@ -107,7 +107,7 @@ class TestProjectorDBUpdate(TestCase): def test_upgrade_old_projector_db(self): """ - Test that we can upgrade an old song db to the current schema + Test that we can upgrade a version 1 db to the current schema """ # GIVEN: An old prjector db old_db = os.path.join(TEST_RESOURCES_PATH, "projector", TEST_DB_PJLINK1) @@ -119,8 +119,7 @@ class TestProjectorDBUpdate(TestCase): updated_to_version, latest_version = upgrade_db(db_url, upgrade) # THEN: the song db should have been upgraded to the latest version - self.assertEqual(updated_to_version, latest_version, - 'The projector DB should have been upgrade to the latest version') + assert updated_to_version == latest_version, 'The projector DB should have been upgrade to the latest version' class TestProjectorDB(TestCase): @@ -157,8 +156,7 @@ class TestProjectorDB(TestCase): record = self.projector.get_projector_by_ip(TEST2_DATA['ip']) # THEN: Verify proper record returned - self.assertTrue(compare_data(Projector(**TEST2_DATA), record), - 'Record found should have been test_2 data') + assert compare_data(Projector(**TEST2_DATA), record) is True, 'Record found should have been test_2 data' def test_find_record_by_name(self): """ @@ -171,8 +169,7 @@ class TestProjectorDB(TestCase): record = self.projector.get_projector_by_name(TEST2_DATA['name']) # THEN: Verify proper record returned - self.assertTrue(compare_data(Projector(**TEST2_DATA), record), - 'Record found should have been test_2 data') + assert compare_data(Projector(**TEST2_DATA), record) is True, 'Record found should have been test_2 data' def test_record_delete(self): """ @@ -187,7 +184,7 @@ class TestProjectorDB(TestCase): # THEN: Verify record not retrievable found = self.projector.get_projector_by_ip(TEST3_DATA['ip']) - self.assertFalse(found, 'test_3 record should have been deleted') + assert found is None, 'test_3 record should have been deleted' def test_record_edit(self): """ @@ -212,12 +209,12 @@ class TestProjectorDB(TestCase): record.model_filter = TEST3_DATA['model_filter'] record.model_lamp = TEST3_DATA['model_lamp'] updated = self.projector.update_projector(record) - self.assertTrue(updated, 'Save updated record should have returned True') + assert updated is True, 'Save updated record should have returned True' record = self.projector.get_projector_by_ip(TEST3_DATA['ip']) # THEN: Record ID should remain the same, but data should be changed - self.assertEqual(record_id, record.id, 'Edited record should have the same ID') - self.assertTrue(compare_data(Projector(**TEST3_DATA), record), 'Edited record should have new data') + assert record_id == record.id, 'Edited record should have the same ID' + assert compare_data(Projector(**TEST3_DATA), record) is True, 'Edited record should have new data' def test_source_add(self): """ @@ -235,7 +232,7 @@ class TestProjectorDB(TestCase): # THEN: Projector should have the same source entry item = self.projector.get_projector_by_id(item_id) - self.assertTrue(compare_source(item.source_list[0], source)) + assert compare_source(item.source_list[0], source) is True, 'Source entry should be the same' def test_manufacturer_repr(self): """ @@ -248,8 +245,8 @@ class TestProjectorDB(TestCase): manufacturer.name = 'OpenLP Test' # THEN: __repr__ should return a proper string - self.assertEqual(str(manufacturer), '', - 'Manufacturer.__repr__() should have returned a proper representation string') + assert str(manufacturer) == '', \ + 'Manufacturer.__repr__() should have returned a proper representation string' def test_model_repr(self): """ @@ -262,8 +259,8 @@ class TestProjectorDB(TestCase): model.name = 'OpenLP Test' # THEN: __repr__ should return a proper string - self.assertEqual(str(model), '', - 'Model.__repr__() should have returned a proper representation string') + assert str(model) == '', \ + 'Model.__repr__() should have returned a proper representation string' def test_source_repr(self): """ @@ -278,19 +275,25 @@ class TestProjectorDB(TestCase): source.text = 'Input text' # THEN: __repr__ should return a proper string - self.assertEqual(str(source), '', - 'Source.__repr__() should have returned a proper representation string') + assert str(source) == '', \ + 'Source.__repr__() should have returned a proper representation string' def test_projector_repr(self): """ Test Projector.__repr__() text """ # GIVEN: Test object + test_string = '< Projector(id="0", ip="127.0.0.1", port="4352", mac_adx="None", pin="None", ' \ + 'name="Test One", location="Somewhere over the rainbow", notes="Not again", ' \ + 'pjlink_name="TEST", pjlink_class="None", manufacturer="IN YOUR DREAMS", model="OpenLP", ' \ + 'serial_no="None", other="None", sources="None", source_list="[]", model_filter="None", ' \ + 'model_lamp="None", sw_version="None") >' + projector = Projector() # WHEN: projector() is populated # NOTE: projector.[pin, other, sources, sw_version, serial_no, sw_version, model_lamp, model_filter] - # should all return None. + # should all return None. # projector.source_list should return an empty list projector.id = 0 projector.ip = '127.0.0.1' @@ -303,13 +306,7 @@ class TestProjectorDB(TestCase): projector.model = 'OpenLP' # THEN: __repr__ should return a proper string - self.assertEqual(str(projector), - '< Projector(id="0", ip="127.0.0.1", port="4352", mac_adx="None", pin="None", ' - 'name="Test One", location="Somewhere over the rainbow", notes="Not again", ' - 'pjlink_name="TEST", manufacturer="IN YOUR DREAMS", model="OpenLP", serial_no="None", ' - 'other="None", sources="None", source_list="[]", model_filter="None", model_lamp="None", ' - 'sw_version="None") >', - 'Projector.__repr__() should have returned a proper representation string') + assert str(projector) == test_string, 'Projector.__repr__() should have returned a proper string' def test_projectorsource_repr(self): """ @@ -326,9 +323,8 @@ class TestProjectorDB(TestCase): self.projector.add_source(source) # THEN: __repr__ should return a proper string - self.assertEqual(str(source), - '', - 'ProjectorSource.__repr__)_ should have returned a proper representation string') + assert str(source) == '', \ + 'ProjectorSource.__repr__)_ should have returned a proper representation string' def test_get_projector_by_id_none(self): """ @@ -341,7 +337,7 @@ class TestProjectorDB(TestCase): results = projector.get_projector_by_id(dbid=123134556409824506) # THEN: Verify return was None - self.assertEqual(results, None, 'Returned results should have equaled None') + assert results is None, 'Returned results should have equaled None' def test_get_projector_all_none(self): """ @@ -354,7 +350,7 @@ class TestProjectorDB(TestCase): results = projector.get_projector_all() # THEN: Verify results is None - self.assertEqual(results, [], 'Returned results should have returned an empty list') + assert [] == results, 'Returned results should have returned an empty list' def test_get_projector_all_one(self): """ @@ -368,8 +364,8 @@ class TestProjectorDB(TestCase): results = self.projector.get_projector_all() # THEN: We should have a list with one entry - self.assertEqual(len(results), 1, 'Returned results should have returned a list with one entry') - self.assertTrue((projector in results), 'Result should have been equal to TEST1_DATA') + assert 1 == len(results), 'Returned results should have returned a list with one entry' + assert (projector in results) is True, 'Result should have been equal to TEST1_DATA' def test_get_projector_all_many(self): """ @@ -387,11 +383,9 @@ class TestProjectorDB(TestCase): results = self.projector.get_projector_all() # THEN: We should have a list with three entries - self.assertEqual(len(results), len(projector_list), - 'Returned results should have returned a list with three entries') + assert len(results) == len(projector_list), 'Returned results should have returned a list with three entries' for projector in results: - self.assertTrue((projector in projector_list), - 'Projector DB entry should have been in expected list') + assert (projector in projector_list) is True, 'Projector DB entry should have been in expected list' def test_get_projector_by_name_fail(self): """ @@ -404,7 +398,7 @@ class TestProjectorDB(TestCase): results = self.projector.get_projector_by_name(name=TEST2_DATA['name']) # THEN: We should have None - self.assertEqual(results, None, 'projector.get_projector_by_name() should have returned None') + assert results is None, 'projector.get_projector_by_name() should have returned None' def test_add_projector_fail(self): """ @@ -417,7 +411,7 @@ class TestProjectorDB(TestCase): results = self.projector.add_projector(Projector(**TEST1_DATA)) # THEN: We should have failed to add new entry - self.assertFalse(results, 'add_projector() should have failed') + assert results is False, 'add_projector() should have failed' def test_update_projector_default_fail(self): """ @@ -430,7 +424,7 @@ class TestProjectorDB(TestCase): results = projector.update_projector() # THEN: We should have failed - self.assertFalse(results, 'update_projector(projector=None) should have returned False') + assert results is False, 'update_projector(projector=None) should have returned False' def test_update_projector_not_in_db_fail(self): """ @@ -444,7 +438,7 @@ class TestProjectorDB(TestCase): results = self.projector.update_projector(projector) # THEN: Results should be False - self.assertFalse(results, 'update_projector(projector=projector) should have returned False') + assert results is False, 'update_projector(projector=projector) should have returned False' def test_delete_projector_fail(self): """ @@ -457,4 +451,4 @@ class TestProjectorDB(TestCase): results = self.projector.delete_projector(Projector(**TEST2_DATA)) # THEN: Results should be False - self.assertFalse(results, 'delete_projector() should have returned False') + assert results is False, 'delete_projector() should have returned False' diff --git a/tests/functional/openlp_core/projectors/test_projector_pjlink_base.py b/tests/functional/openlp_core/projectors/test_projector_pjlink_base.py index ca699e4eb..b3d45fb68 100644 --- a/tests/functional/openlp_core/projectors/test_projector_pjlink_base.py +++ b/tests/functional/openlp_core/projectors/test_projector_pjlink_base.py @@ -25,7 +25,19 @@ Package to test the openlp.core.projectors.pjlink base package. from unittest import TestCase from unittest.mock import call, patch, MagicMock -from openlp.core.projectors.constants import E_PARAMETER, STATUS_CODE, S_ON, S_CONNECTED, QSOCKET_STATE +import openlp.core.projectors.pjlink +from openlp.core.projectors.constants import \ + E_NOT_CONNECTED, \ + E_PARAMETER, \ + E_UNKNOWN_SOCKET_ERROR, \ + STATUS_CODE, \ + STATUS_MSG, \ + S_CONNECTED, \ + S_CONNECTING, \ + S_NOT_CONNECTED, \ + S_OK, \ + S_ON, \ + QSOCKET_STATE from openlp.core.projectors.db import Projector from openlp.core.projectors.pjlink import PJLink from tests.resources.projector.data import TEST1_DATA @@ -37,55 +49,44 @@ class TestPJLinkBase(TestCase): """ Tests for the PJLink module """ - def setUp(self): - ''' - TestPJLinkCommands part 2 initialization - ''' - self.pjlink_test = PJLink(Projector(**TEST1_DATA), no_poll=True) - - def tearDown(self): - ''' - TestPJLinkCommands part 2 cleanups - ''' - self.pjlink_test = None - - @patch.object(pjlink_test, 'change_status') - def test_status_change(self, mock_change_status): + def test_status_change(self): """ Test process_command call with ERR2 (Parameter) status """ # GIVEN: Test object - pjlink = pjlink_test + with patch('openlp.core.projectors.pjlink.PJLink.changeStatus') as mock_changeStatus: - # WHEN: process_command is called with "ERR2" status from projector - pjlink.process_command('POWR', 'ERR2') + pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) - # 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( - STATUS_CODE[E_PARAMETER])) + # WHEN: process_command is called with "ERR2" status from projector + pjlink.process_command('POWR', 'ERR2') - @patch.object(pjlink_test, 'disconnect_from_host') - def test_socket_abort(self, mock_disconnect): + # THEN: change_status should have called change_status with E_UNDEFINED + # as first parameter + mock_changeStatus.called_with(E_PARAMETER, + 'change_status should have been called with "{}"'.format( + STATUS_CODE[E_PARAMETER])) + + def test_socket_abort(self): """ Test PJLink.socket_abort calls disconnect_from_host """ # GIVEN: Test object - pjlink = pjlink_test + with patch('openlp.core.projectors.pjlink.PJLink.disconnect_from_host') as mock_disconnect: + pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) - # WHEN: Calling socket_abort - pjlink.socket_abort() + # 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') + # THEN: disconnect_from_host should be called + assert mock_disconnect.called is True, '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 = PJLink(Projector(**TEST1_DATA), no_poll=True) pjlink.state = MagicMock() pjlink.timer = MagicMock() pjlink.state.return_value = False @@ -95,47 +96,432 @@ class TestPJLinkBase(TestCase): 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') + assert pjlink.timer.called is False, 'Should have returned without calling any other method' - def test_poll_loop_start(self): + def test_poll_loop_set_interval(self): """ Test PJLink.poll_loop makes correct calls """ # GIVEN: Mocks and test data - mock_state = patch.object(self.pjlink_test, 'state').start() - mock_state.return_value = QSOCKET_STATE[S_CONNECTED] - mock_timer = patch.object(self.pjlink_test, 'timer').start() - mock_timer.interval.return_value = 10 - mock_send_command = patch.object(self.pjlink_test, 'send_command').start() + with patch('openlp.core.projectors.pjlink.PJLink.send_command') as mock_send_command: - pjlink = self.pjlink_test - 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 - call_list = [ - call('POWR'), - call('ERST'), - call('LAMP'), - call('AVMT'), - call('INPT'), - call('INST'), - call('INFO'), - call('INF1'), - call('INF2'), - call('NAME'), - ] + pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) + pjlink.state = MagicMock() + pjlink.state.return_value = QSOCKET_STATE[S_CONNECTED] + pjlink.poll_timer = MagicMock() + pjlink.poll_timer.interval.return_value = 10 - # WHEN: PJLink.poll_loop is called - pjlink.poll_loop() + 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 + call_list = [ + call('POWR'), + call('ERST'), + call('LAMP'), + call('AVMT'), + call('INPT'), + call('INST'), + call('INFO'), + call('INF1'), + call('INF2'), + call('NAME'), + ] - # THEN: proper calls were made to retrieve projector data - # First, call to update the timer with the next interval - self.assertTrue(mock_timer.setInterval.called) - # Next, should have called the timer to start - self.assertTrue(mock_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') + # 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 + assert pjlink.poll_timer.setInterval.called is True, 'Timer update interval should have been called' + # 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') + + def test_projector_change_status_unknown_socket_error(self): + """ + Test change_status with connection error + """ + # GIVEN: Test object and mocks + with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ + patch.object(openlp.core.projectors.pjlink.PJLink, 'changeStatus') as mock_changeStatus, \ + patch.object(openlp.core.projectors.pjlink.PJLink, 'projectorUpdateIcons') as mock_UpdateIcons: + + pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) + pjlink.projector_status = 0 + pjlink.status_connect = 0 + log_debug_calls = [ + call('({ip}) Changing status to {status} "{msg}"'.format(ip=pjlink.name, + status=STATUS_CODE[E_UNKNOWN_SOCKET_ERROR], + msg=STATUS_MSG[E_UNKNOWN_SOCKET_ERROR])), + call('({ip}) status_connect: {code}: "{msg}"'.format(ip=pjlink.name, + code=STATUS_CODE[E_NOT_CONNECTED], + msg=STATUS_MSG[E_NOT_CONNECTED])), + call('({ip}) projector_status: {code}: "{msg}"'.format(ip=pjlink.name, + code=STATUS_CODE[S_OK], + msg=STATUS_MSG[S_OK])), + call('({ip}) error_status: {code}: "{msg}"'.format(ip=pjlink.name, + code=STATUS_CODE[E_UNKNOWN_SOCKET_ERROR], + msg=STATUS_MSG[E_UNKNOWN_SOCKET_ERROR]))] + + # WHEN: change_status called with unknown socket error + pjlink.change_status(status=E_UNKNOWN_SOCKET_ERROR) + + # THEN: Proper settings should change and signals sent + mock_log.debug.assert_has_calls(log_debug_calls) + assert pjlink.projector_status == S_OK, 'Projector status should not have changed' + assert pjlink.status_connect == E_NOT_CONNECTED, 'Status connect should be NOT CONNECTED' + assert mock_UpdateIcons.emit.called is True, 'Should have called UpdateIcons' + mock_changeStatus.emit.assert_called_once_with(pjlink.ip, E_UNKNOWN_SOCKET_ERROR, + STATUS_MSG[E_UNKNOWN_SOCKET_ERROR]) + + def test_projector_change_status_connection_status_connecting(self): + """ + Test change_status with connecting status + """ + # GIVEN: Test object and mocks + with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ + patch.object(openlp.core.projectors.pjlink.PJLink, 'changeStatus') as mock_changeStatus, \ + patch.object(openlp.core.projectors.pjlink.PJLink, 'projectorUpdateIcons') as mock_UpdateIcons: + + pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) + pjlink.projector_status = 0 + pjlink.status_connect = 0 + log_debug_calls = [ + call('({ip}) Changing status to {status} "{msg}"'.format(ip=pjlink.name, + status=STATUS_CODE[S_CONNECTING], + msg=STATUS_MSG[S_CONNECTING])), + call('({ip}) status_connect: {code}: "{msg}"'.format(ip=pjlink.name, + code=STATUS_CODE[S_CONNECTING], + msg=STATUS_MSG[S_CONNECTING])), + call('({ip}) projector_status: {code}: "{msg}"'.format(ip=pjlink.name, + code=STATUS_CODE[S_OK], + msg=STATUS_MSG[S_OK])), + call('({ip}) error_status: {code}: "{msg}"'.format(ip=pjlink.name, + code=STATUS_CODE[S_OK], + msg=STATUS_MSG[S_OK]))] + + # WHEN: change_status called with CONNECTING + pjlink.change_status(status=S_CONNECTING) + + # THEN: Proper settings should change and signals sent + mock_log.debug.assert_has_calls(log_debug_calls) + mock_changeStatus.emit.assert_called_once_with(pjlink.ip, S_CONNECTING, STATUS_MSG[S_CONNECTING]) + assert pjlink.projector_status == S_OK, 'Projector status should not have changed' + assert pjlink.status_connect == S_CONNECTING, 'Status connect should be CONNECTING' + assert mock_UpdateIcons.emit.called is True, 'Should have called UpdateIcons' + + def test_projector_change_status_connection_status_connected(self): + """ + Test change_status with connected status + """ + # GIVEN: Test object and mocks + with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ + patch.object(openlp.core.projectors.pjlink.PJLink, 'changeStatus') as mock_changeStatus: + + pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) + pjlink.projector_status = 0 + pjlink.status_connect = 0 + log_debug_calls = [ + call('({ip}) Changing status to {status} "{msg}"'.format(ip=pjlink.name, + status=STATUS_CODE[S_CONNECTED], + msg=STATUS_MSG[S_CONNECTED])), + call('({ip}) status_connect: {code}: "{msg}"'.format(ip=pjlink.name, + code=STATUS_CODE[S_CONNECTED], + msg=STATUS_MSG[S_CONNECTED])), + call('({ip}) projector_status: {code}: "{msg}"'.format(ip=pjlink.name, + code=STATUS_CODE[S_OK], + msg=STATUS_MSG[S_OK])), + call('({ip}) error_status: {code}: "{msg}"'.format(ip=pjlink.name, + code=STATUS_CODE[S_OK], + msg=STATUS_MSG[S_OK]))] + + # WHEN: change_status called with CONNECTED + pjlink.change_status(status=S_CONNECTED) + + # THEN: Proper settings should change and signals sent + mock_log.debug.assert_has_calls(log_debug_calls) + mock_changeStatus.emit.assert_called_once_with(pjlink.ip, S_CONNECTED, 'Connected') + assert pjlink.projector_status == S_OK, 'Projector status should not have changed' + assert pjlink.status_connect == S_CONNECTED, 'Status connect should be CONNECTED' + + def test_projector_change_status_connection_status_with_message(self): + """ + Test change_status with connection status + """ + # GIVEN: Test object and mocks + with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ + patch.object(openlp.core.projectors.pjlink.PJLink, 'changeStatus') as mock_changeStatus: + + pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) + pjlink.projector_status = 0 + pjlink.status_connect = 0 + test_message = 'Different Status Message than default' + log_debug_calls = [ + call('({ip}) Changing status to {status} "{msg}"'.format(ip=pjlink.name, + status=STATUS_CODE[S_ON], + msg=test_message)), + call('({ip}) status_connect: {code}: "{msg}"'.format(ip=pjlink.name, + code=STATUS_CODE[S_OK], + msg=test_message)), + call('({ip}) projector_status: {code}: "{msg}"'.format(ip=pjlink.name, + code=STATUS_CODE[S_ON], + msg=test_message)), + call('({ip}) error_status: {code}: "{msg}"'.format(ip=pjlink.name, + code=STATUS_CODE[S_OK], + msg=test_message))] + + # WHEN: change_status called with projector ON status + pjlink.change_status(status=S_ON, msg=test_message) + + # THEN: Proper settings should change and signals sent + mock_log.debug.assert_has_calls(log_debug_calls) + mock_changeStatus.emit.assert_called_once_with(pjlink.ip, S_ON, test_message) + assert pjlink.projector_status == S_ON, 'Projector status should be ON' + assert pjlink.status_connect == S_OK, 'Status connect should not have changed' + + def test_projector_get_av_mute_status(self): + """ + Test sending command to retrieve shutter/audio state + """ + # GIVEN: Test object and mocks + with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ + patch.object(openlp.core.projectors.pjlink.PJLink, 'send_command') as mock_send_command: + pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) + test_data = 'AVMT' + log_debug_calls = [call('({ip}) reset_information() connect status is ' + '{state}'.format(ip=pjlink.name, state=STATUS_CODE[S_NOT_CONNECTED])), + call('({ip}) Sending {cmd} command'.format(ip=pjlink.name, cmd=test_data))] + + # 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_has_calls(log_debug_calls) + mock_send_command.assert_called_once_with(cmd=test_data) + + def test_projector_get_available_inputs(self): + """ + Test sending command to retrieve avaliable inputs + """ + # GIVEN: Test object and mocks + with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ + patch.object(openlp.core.projectors.pjlink.PJLink, 'send_command') as mock_send_command: + pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) + test_data = 'INST' + log_debug_calls = [call('({ip}) reset_information() connect status is ' + '{state}'.format(ip=pjlink.name, state=STATUS_CODE[S_NOT_CONNECTED])), + call('({ip}) Sending {cmd} command'.format(ip=pjlink.name, cmd=test_data))] + + # WHEN: get_available_inputs is called + pjlink.get_available_inputs() + + # THEN: log data and send_command should have been called + mock_log.debug.assert_has_calls(log_debug_calls) + mock_send_command.assert_called_once_with(cmd=test_data) + + def test_projector_get_error_status(self): + """ + Test sending command to retrieve projector error status + """ + # GIVEN: Test object and mocks + with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ + patch.object(openlp.core.projectors.pjlink.PJLink, 'send_command') as mock_send_command: + pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) + test_data = 'ERST' + log_debug_calls = [call('({ip}) reset_information() connect status is ' + '{state}'.format(ip=pjlink.name, state=STATUS_CODE[S_NOT_CONNECTED])), + call('({ip}) Sending {cmd} command'.format(ip=pjlink.name, cmd=test_data))] + + # WHEN: get_error_status is called + pjlink.get_error_status() + + # THEN: log data and send_command should have been called + mock_log.debug.assert_has_calls(log_debug_calls) + mock_send_command.assert_called_once_with(cmd=test_data) + + def test_projector_get_input_source(self): + """ + Test sending command to retrieve current input + """ + # GIVEN: Test object and mocks + with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ + patch.object(openlp.core.projectors.pjlink.PJLink, 'send_command') as mock_send_command: + pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) + test_data = 'INPT' + log_debug_calls = [call('({ip}) reset_information() connect status is ' + '{state}'.format(ip=pjlink.name, state=STATUS_CODE[S_NOT_CONNECTED])), + call('({ip}) Sending {cmd} command'.format(ip=pjlink.name, cmd=test_data))] + + # WHEN: get_input_source is called + pjlink.get_input_source() + + # THEN: log data and send_command should have been called + mock_log.debug.assert_has_calls(log_debug_calls) + mock_send_command.assert_called_once_with(cmd=test_data) + + def test_projector_get_lamp_status(self): + """ + Test sending command to retrieve lamp(s) status + """ + # GIVEN: Test object and mocks + with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ + patch.object(openlp.core.projectors.pjlink.PJLink, 'send_command') as mock_send_command: + pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) + test_data = 'LAMP' + log_debug_calls = [call('({ip}) reset_information() connect status is ' + '{state}'.format(ip=pjlink.name, state=STATUS_CODE[S_NOT_CONNECTED])), + call('({ip}) Sending {cmd} command'.format(ip=pjlink.name, cmd=test_data))] + + # WHEN: get_input_source is called + pjlink.get_lamp_status() + + # THEN: log data and send_command should have been called + mock_log.debug.assert_has_calls(log_debug_calls) + mock_send_command.assert_called_once_with(cmd=test_data) + + def test_projector_get_manufacturer(self): + """ + Test sending command to retrieve manufacturer name + """ + # GIVEN: Test object and mocks + with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ + patch.object(openlp.core.projectors.pjlink.PJLink, 'send_command') as mock_send_command: + pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) + test_data = 'INF1' + log_debug_calls = [call('({ip}) reset_information() connect status is ' + '{state}'.format(ip=pjlink.name, state=STATUS_CODE[S_NOT_CONNECTED])), + call('({ip}) Sending {cmd} command'.format(ip=pjlink.name, cmd=test_data))] + + # WHEN: get_input_source is called + pjlink.get_manufacturer() + + # THEN: log data and send_command should have been called + mock_log.debug.assert_has_calls(log_debug_calls) + mock_send_command.assert_called_once_with(cmd=test_data) + + def test_projector_get_model(self): + """ + Test sending command to get model information + """ + # GIVEN: Test object and mocks + with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ + patch.object(openlp.core.projectors.pjlink.PJLink, 'send_command') as mock_send_command: + pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) + test_data = 'INF2' + log_debug_calls = [call('({ip}) reset_information() connect status is ' + '{state}'.format(ip=pjlink.name, state=STATUS_CODE[S_NOT_CONNECTED])), + call('({ip}) Sending {cmd} command'.format(ip=pjlink.name, cmd=test_data))] + + # WHEN: get_input_source is called + pjlink.get_model() + + # THEN: log data and send_command should have been called + mock_log.debug.assert_has_calls(log_debug_calls) + mock_send_command.assert_called_once_with(cmd=test_data) + + def test_projector_get_name(self): + """ + Test sending command to get user-assigned name + """ + # GIVEN: Test object and mocks + with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ + patch.object(openlp.core.projectors.pjlink.PJLink, 'send_command') as mock_send_command: + pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) + test_data = 'NAME' + log_debug_calls = [call('({ip}) reset_information() connect status is ' + '{state}'.format(ip=pjlink.name, state=STATUS_CODE[S_NOT_CONNECTED])), + call('({ip}) Sending {cmd} command'.format(ip=pjlink.name, cmd=test_data))] + + # WHEN: get_input_source is called + pjlink.get_name() + + # THEN: log data and send_command should have been called + mock_log.debug.assert_has_calls(log_debug_calls) + mock_send_command.assert_called_once_with(cmd=test_data) + + def test_projector_get_other_info(self): + """ + Test sending command to retrieve other information + """ + # GIVEN: Test object and mocks + with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ + patch.object(openlp.core.projectors.pjlink.PJLink, 'send_command') as mock_send_command: + pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) + test_data = 'INFO' + log_debug_calls = [call('({ip}) reset_information() connect status is ' + '{state}'.format(ip=pjlink.name, state=STATUS_CODE[S_NOT_CONNECTED])), + call('({ip}) Sending {cmd} command'.format(ip=pjlink.name, cmd=test_data))] + + # WHEN: get_input_source is called + pjlink.get_other_info() + + # THEN: log data and send_command should have been called + mock_log.debug.assert_has_calls(log_debug_calls) + mock_send_command.assert_called_once_with(cmd=test_data) + + def test_projector_get_power_status(self): + """ + Test sending command to retrieve current power state + """ + # GIVEN: Test object and mocks + with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ + patch.object(openlp.core.projectors.pjlink.PJLink, 'send_command') as mock_send_command: + pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) + test_data = 'POWR' + log_debug_calls = [call('({ip}) reset_information() connect status is ' + '{state}'.format(ip=pjlink.name, state=STATUS_CODE[S_NOT_CONNECTED])), + call('({ip}) Sending {cmd} command'.format(ip=pjlink.name, cmd=test_data))] + + # WHEN: get_input_source is called + pjlink.get_power_status() + + # THEN: log data and send_command should have been called + mock_log.debug.assert_has_calls(log_debug_calls) + mock_send_command.assert_called_once_with(cmd=test_data) + + def test_projector_get_status_invalid(self): + """ + Test to check returned information for error code + """ + # GIVEN: Test object + pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) + test_string = 'NaN test' + + # WHEN: get_status called + code, message = pjlink._get_status(status=test_string) + + # THEN: Proper data should have been returned + assert code == -1, 'Should have returned -1 as a bad status check' + assert message is None, 'Invalid code type should have returned None for message' + + def test_projector_get_status_valid(self): + """ + Test to check returned information for status codes + """ + # GIVEN: Test object + test_message = 'Not Connected' + pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) + + # WHEN: get_status called + code, message = pjlink._get_status(status=S_NOT_CONNECTED) + + # THEN: Proper strings should have been returned + assert code == 'S_NOT_CONNECTED', 'Code returned should have been the same code that was sent' + assert 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(Projector(**TEST1_DATA), no_poll=True) + + # WHEN: get_status called + code, message = pjlink._get_status(status=9999) + + # THEN: Proper strings should have been returned + assert code is None, 'Code returned should have been the same code that was sent' + assert message is None, 'Should have returned None as message' diff --git a/tests/functional/openlp_core/projectors/test_projector_pjlink_cmd_routing.py b/tests/functional/openlp_core/projectors/test_projector_pjlink_cmd_routing.py index 74623b9c3..d6c549177 100644 --- a/tests/functional/openlp_core/projectors/test_projector_pjlink_cmd_routing.py +++ b/tests/functional/openlp_core/projectors/test_projector_pjlink_cmd_routing.py @@ -44,15 +44,16 @@ class TestPJLinkRouting(TestCase): Test not a valid command """ # GIVEN: Test object - log_warning_text = [call('(111.111.111.111) get_data(): Invalid packet - unknown command "UNK"')] - log_debug_text = [call('(111.111.111.111) get_data(ip="111.111.111.111" buffer="b\'%1UNK=Huh?\'"'), - call('(111.111.111.111) get_data(): Checking new data "%1UNK=Huh?"')] - with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ patch.object(openlp.core.projectors.pjlink.PJLink, '_trash_buffer') as mock_buffer: pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) pjlink.pjlink_functions = MagicMock() + log_warning_text = [call('({ip}) get_data(): Invalid packet - ' + 'unknown command "UNK"'.format(ip=pjlink.name))] + log_debug_text = [call('({ip}) get_data(ip="111.111.111.111" ' + 'buffer="b\'%1UNK=Huh?\'"'.format(ip=pjlink.name)), + call('({ip}) get_data(): Checking new data "%1UNK=Huh?"'.format(ip=pjlink.name))] # WHEN: get_data called with an unknown command pjlink.get_data(buff='{prefix}1UNK=Huh?'.format(prefix=PJLINK_PREFIX).encode('utf-8')) @@ -68,13 +69,12 @@ class TestPJLinkRouting(TestCase): Test process_command calls proper function """ # GIVEN: Test object and mocks - log_debug_calls = [call('(111.111.111.111) Processing command "CLSS" with data "1"'), - call('(111.111.111.111) Calling function for CLSS')] - with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ patch.object(openlp.core.projectors.pjlink.PJLink, 'process_clss') as mock_process_clss: pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) + log_debug_calls = [call('({ip}) Processing command "CLSS" with data "1"'.format(ip=pjlink.name)), + call('({ip}) Calling function for CLSS'.format(ip=pjlink.name))] # WHEN: process_command is called with valid function and data pjlink.process_command(cmd='CLSS', data='1') @@ -88,9 +88,6 @@ class TestPJLinkRouting(TestCase): Test ERRA - Authentication Error """ # GIVEN: Test object - log_error_calls = [call('(111.111.111.111) PJLINK: {msg}'.format(msg=STATUS_MSG[E_AUTHENTICATION]))] - log_debug_calls = [call('(111.111.111.111) Processing command "PJLINK" with data "ERRA"')] - with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ patch.object(openlp.core.projectors.pjlink.PJLink, 'process_pjlink') as mock_process_pjlink, \ patch.object(openlp.core.projectors.pjlink.PJLink, 'change_status') as mock_change_status, \ @@ -98,6 +95,8 @@ class TestPJLinkRouting(TestCase): patch.object(openlp.core.projectors.pjlink.PJLink, 'projectorAuthentication') as mock_authentication: pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) + log_error_calls = [call('({ip}) PJLINK: {msg}'.format(ip=pjlink.name, msg=STATUS_MSG[E_AUTHENTICATION]))] + log_debug_calls = [call('({ip}) Processing command "PJLINK" with data "ERRA"'.format(ip=pjlink.name))] # WHEN: process_command called with ERRA pjlink.process_command(cmd='PJLINK', data=PJLINK_ERRORS[E_AUTHENTICATION]) @@ -115,14 +114,13 @@ class TestPJLinkRouting(TestCase): Test ERR1 - Undefined projector function """ # GIVEN: Test object - log_error_text = [call('(111.111.111.111) CLSS: {msg}'.format(msg=STATUS_MSG[E_UNDEFINED]))] - log_debug_text = [call('(111.111.111.111) Processing command "CLSS" with data "ERR1"'), - call('(111.111.111.111) Calling function for CLSS')] - with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ patch.object(openlp.core.projectors.pjlink.PJLink, 'process_clss') as mock_process_clss: pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) + log_error_text = [call('({ip}) CLSS: {msg}'.format(ip=pjlink.name, msg=STATUS_MSG[E_UNDEFINED]))] + log_debug_text = [call('({ip}) Processing command "CLSS" with data "ERR1"'.format(ip=pjlink.name)), + call('({ip}) Calling function for CLSS'.format(ip=pjlink.name))] # WHEN: process_command called with ERR1 pjlink.process_command(cmd='CLSS', data=PJLINK_ERRORS[E_UNDEFINED]) @@ -137,14 +135,13 @@ class TestPJLinkRouting(TestCase): Test ERR2 - Parameter Error """ # GIVEN: Test object - log_error_text = [call('(111.111.111.111) CLSS: {msg}'.format(msg=STATUS_MSG[E_PARAMETER]))] - log_debug_text = [call('(111.111.111.111) Processing command "CLSS" with data "ERR2"'), - call('(111.111.111.111) Calling function for CLSS')] - with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ patch.object(openlp.core.projectors.pjlink.PJLink, 'process_clss') as mock_process_clss: pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) + log_error_text = [call('({ip}) CLSS: {msg}'.format(ip=pjlink.name, msg=STATUS_MSG[E_PARAMETER]))] + log_debug_text = [call('({ip}) Processing command "CLSS" with data "ERR2"'.format(ip=pjlink.name)), + call('({ip}) Calling function for CLSS'.format(ip=pjlink.name))] # WHEN: process_command called with ERR2 pjlink.process_command(cmd='CLSS', data=PJLINK_ERRORS[E_PARAMETER]) @@ -159,14 +156,13 @@ class TestPJLinkRouting(TestCase): Test ERR3 - Unavailable error """ # GIVEN: Test object - log_error_text = [call('(111.111.111.111) CLSS: {msg}'.format(msg=STATUS_MSG[E_UNAVAILABLE]))] - log_debug_text = [call('(111.111.111.111) Processing command "CLSS" with data "ERR3"'), - call('(111.111.111.111) Calling function for CLSS')] - with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ patch.object(openlp.core.projectors.pjlink.PJLink, 'process_clss') as mock_process_clss: pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) + log_error_text = [call('({ip}) CLSS: {msg}'.format(ip=pjlink.name, msg=STATUS_MSG[E_UNAVAILABLE]))] + log_debug_text = [call('({ip}) Processing command "CLSS" with data "ERR3"'.format(ip=pjlink.name)), + call('({ip}) Calling function for CLSS'.format(ip=pjlink.name))] # WHEN: process_command called with ERR3 pjlink.process_command(cmd='CLSS', data=PJLINK_ERRORS[E_UNAVAILABLE]) @@ -181,14 +177,13 @@ class TestPJLinkRouting(TestCase): Test ERR3 - Unavailable error """ # GIVEN: Test object - log_error_text = [call('(111.111.111.111) CLSS: {msg}'.format(msg=STATUS_MSG[E_PROJECTOR]))] - log_debug_text = [call('(111.111.111.111) Processing command "CLSS" with data "ERR4"'), - call('(111.111.111.111) Calling function for CLSS')] - with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ patch.object(openlp.core.projectors.pjlink.PJLink, 'process_clss') as mock_process_clss: pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) + log_error_text = [call('({ip}) CLSS: {msg}'.format(ip=pjlink.name, msg=STATUS_MSG[E_PROJECTOR]))] + log_debug_text = [call('({ip}) Processing command "CLSS" with data "ERR4"'.format(ip=pjlink.name)), + call('({ip}) Calling function for CLSS'.format(ip=pjlink.name))] # WHEN: process_command called with ERR4 pjlink.process_command(cmd='CLSS', data=PJLINK_ERRORS[E_PROJECTOR]) @@ -203,14 +198,13 @@ class TestPJLinkRouting(TestCase): Test command valid but no method to process yet """ # GIVEN: Test object - log_warning_text = [call('(111.111.111.111) Unable to process command="CLSS" (Future option?)')] - log_debug_text = [call('(111.111.111.111) Processing command "CLSS" with data "Huh?"')] - with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ patch.object(openlp.core.projectors.pjlink.PJLink, 'process_clss') as mock_process_clss: pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) pjlink.pjlink_functions = MagicMock() + log_warning_text = [call('({ip}) Unable to process command="CLSS" (Future option?)'.format(ip=pjlink.name))] + log_debug_text = [call('({ip}) Processing command "CLSS" with data "Huh?"'.format(ip=pjlink.name))] # WHEN: Processing a possible future command pjlink.process_command(cmd='CLSS', data="Huh?") @@ -227,14 +221,13 @@ class TestPJLinkRouting(TestCase): """ # GIVEN: Initial mocks and data # GIVEN: Test object and mocks - log_debug_calls = [call('(111.111.111.111) Processing command "CLSS" with data "OK"'), - call('(111.111.111.111) Command "CLSS" returned OK')] - with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ patch.object(openlp.core.projectors.pjlink.PJLink, 'send_command') as mock_send_command, \ patch.object(openlp.core.projectors.pjlink.PJLink, 'process_clss') as mock_process_clss: pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) + log_debug_calls = [call('({ip}) Processing command "CLSS" with data "OK"'.format(ip=pjlink.name)), + call('({ip}) Command "CLSS" returned OK'.format(ip=pjlink.name))] # WHEN: process_command is called with valid function and data pjlink.process_command(cmd='CLSS', data='OK') diff --git a/tests/functional/openlp_core/projectors/test_projector_pjlink_commands_01.py b/tests/functional/openlp_core/projectors/test_projector_pjlink_commands_01.py index c8551b594..42d46f735 100644 --- a/tests/functional/openlp_core/projectors/test_projector_pjlink_commands_01.py +++ b/tests/functional/openlp_core/projectors/test_projector_pjlink_commands_01.py @@ -26,9 +26,17 @@ from unittest import TestCase from unittest.mock import call, patch import openlp.core.projectors.pjlink -from openlp.core.projectors.constants import PJLINK_ERST_DATA, PJLINK_ERST_STATUS, PJLINK_POWR_STATUS, \ - STATUS_CODE, STATUS_MSG, E_ERROR, E_NOT_CONNECTED, E_UNKNOWN_SOCKET_ERROR, E_WARN, \ - S_CONNECTED, S_CONNECTING, S_OFF, S_OK, S_ON, S_NOT_CONNECTED, S_STANDBY +from openlp.core.projectors.constants import \ + PJLINK_ERST_DATA, \ + PJLINK_ERST_STATUS, \ + PJLINK_POWR_STATUS, \ + STATUS_CODE, \ + E_ERROR, \ + E_WARN, \ + S_OK, \ + S_ON, \ + S_NOT_CONNECTED, \ + S_STANDBY from openlp.core.projectors.db import Projector from openlp.core.projectors.pjlink import PJLink @@ -39,401 +47,6 @@ class TestPJLinkCommands(TestCase): """ Tests for the PJLinkCommands class part 1 """ - def test_projector_change_status_unknown_socket_error(self): - """ - Test change_status with connection error - """ - log_debug_calls = [ - call('(111.111.111.111) Changing status to ' - '{status} "{msg}"'.format(status=STATUS_CODE[E_UNKNOWN_SOCKET_ERROR], - msg=STATUS_MSG[E_UNKNOWN_SOCKET_ERROR])), - call('(111.111.111.111) status_connect: ' - '{code}: "{msg}"'.format(code=STATUS_CODE[E_NOT_CONNECTED], - msg=STATUS_MSG[E_NOT_CONNECTED])), - call('(111.111.111.111) projector_status: ' - '{code}: "{msg}"'.format(code=STATUS_CODE[S_OK], - msg=STATUS_MSG[S_OK])), - call('(111.111.111.111) error_status: ' - '{code}: "{msg}"'.format(code=STATUS_CODE[E_UNKNOWN_SOCKET_ERROR], - msg=STATUS_MSG[E_UNKNOWN_SOCKET_ERROR]))] - - # GIVEN: Test object and mocks - with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ - patch.object(openlp.core.projectors.pjlink.PJLink, 'changeStatus') as mock_changeStatus, \ - patch.object(openlp.core.projectors.pjlink.PJLink, 'projectorUpdateIcons') as mock_UpdateIcons: - - pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) - pjlink.projector_status = 0 - pjlink.status_connect = 0 - - # WHEN: change_status called with unknown socket error - pjlink.change_status(status=E_UNKNOWN_SOCKET_ERROR) - - # THEN: Proper settings should change and signals sent - mock_log.debug.assert_has_calls(log_debug_calls) - assert pjlink.projector_status == S_OK, 'Projector status should not have changed' - assert pjlink.status_connect == E_NOT_CONNECTED, 'Status connect should be NOT CONNECTED' - assert mock_UpdateIcons.emit.called is True, 'Should have called UpdateIcons' - mock_changeStatus.emit.assert_called_once_with(pjlink.ip, E_UNKNOWN_SOCKET_ERROR, - STATUS_MSG[E_UNKNOWN_SOCKET_ERROR]) - - def test_projector_change_status_connection_status_connecting(self): - """ - Test change_status with connecting status - """ - log_debug_calls = [ - call('(111.111.111.111) Changing status to ' - '{status} "{msg}"'.format(status=STATUS_CODE[S_CONNECTING], - msg=STATUS_MSG[S_CONNECTING])), - call('(111.111.111.111) status_connect: ' - '{code}: "{msg}"'.format(code=STATUS_CODE[S_CONNECTING], - msg=STATUS_MSG[S_CONNECTING])), - call('(111.111.111.111) projector_status: ' - '{code}: "{msg}"'.format(code=STATUS_CODE[S_OK], - msg=STATUS_MSG[S_OK])), - call('(111.111.111.111) error_status: ' - '{code}: "{msg}"'.format(code=STATUS_CODE[S_OK], - msg=STATUS_MSG[S_OK]))] - - # GIVEN: Test object and mocks - with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ - patch.object(openlp.core.projectors.pjlink.PJLink, 'changeStatus') as mock_changeStatus, \ - patch.object(openlp.core.projectors.pjlink.PJLink, 'projectorUpdateIcons') as mock_UpdateIcons: - - pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) - pjlink.projector_status = 0 - pjlink.status_connect = 0 - - # WHEN: change_status called with CONNECTING - pjlink.change_status(status=S_CONNECTING) - - # THEN: Proper settings should change and signals sent - mock_log.debug.assert_has_calls(log_debug_calls) - mock_changeStatus.emit.assert_called_once_with(pjlink.ip, S_CONNECTING, STATUS_MSG[S_CONNECTING]) - assert pjlink.projector_status == S_OK, 'Projector status should not have changed' - assert pjlink.status_connect == S_CONNECTING, 'Status connect should be CONNECTING' - assert mock_UpdateIcons.emit.called is True, 'Should have called UpdateIcons' - - def test_projector_change_status_connection_status_connected(self): - """ - Test change_status with connected status - """ - log_debug_calls = [ - call('(111.111.111.111) Changing status to ' - '{status} "{msg}"'.format(status=STATUS_CODE[S_CONNECTED], - msg=STATUS_MSG[S_CONNECTED])), - call('(111.111.111.111) status_connect: ' - '{code}: "{msg}"'.format(code=STATUS_CODE[S_CONNECTED], - msg=STATUS_MSG[S_CONNECTED])), - call('(111.111.111.111) projector_status: ' - '{code}: "{msg}"'.format(code=STATUS_CODE[S_OK], - msg=STATUS_MSG[S_OK])), - call('(111.111.111.111) error_status: ' - '{code}: "{msg}"'.format(code=STATUS_CODE[S_OK], - msg=STATUS_MSG[S_OK]))] - - # GIVEN: Test object and mocks - with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ - patch.object(openlp.core.projectors.pjlink.PJLink, 'changeStatus') as mock_changeStatus: - - pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) - pjlink.projector_status = 0 - pjlink.status_connect = 0 - - # WHEN: change_status called with CONNECTED - pjlink.change_status(status=S_CONNECTED) - - # THEN: Proper settings should change and signals sent - mock_log.debug.assert_has_calls(log_debug_calls) - mock_changeStatus.emit.assert_called_once_with(pjlink.ip, S_CONNECTED, 'Connected') - assert pjlink.projector_status == S_OK, 'Projector status should not have changed' - assert pjlink.status_connect == S_CONNECTED, 'Status connect should be CONNECTED' - - def test_projector_change_status_connection_status_with_message(self): - """ - Test change_status with connection status - """ - test_message = 'Different Status Message than default' - log_debug_calls = [ - call('(111.111.111.111) Changing status to {status} "{msg}"'.format(status=STATUS_CODE[S_ON], - msg=test_message)), - call('(111.111.111.111) status_connect: {code}: "{msg}"'.format(code=STATUS_CODE[S_OK], - msg=test_message)), - call('(111.111.111.111) projector_status: {code}: "{msg}"'.format(code=STATUS_CODE[S_ON], - msg=test_message)), - call('(111.111.111.111) error_status: {code}: "{msg}"'.format(code=STATUS_CODE[S_OK], - msg=test_message))] - - # GIVEN: Test object and mocks - with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ - patch.object(openlp.core.projectors.pjlink.PJLink, 'changeStatus') as mock_changeStatus: - - pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) - pjlink.projector_status = 0 - pjlink.status_connect = 0 - - # WHEN: change_status called with projector ON status - pjlink.change_status(status=S_ON, msg=test_message) - - # THEN: Proper settings should change and signals sent - mock_log.debug.assert_has_calls(log_debug_calls) - mock_changeStatus.emit.assert_called_once_with(pjlink.ip, S_ON, test_message) - assert pjlink.projector_status == S_ON, 'Projector status should be ON' - assert pjlink.status_connect == S_OK, 'Status connect should not have changed' - - def test_projector_get_av_mute_status(self): - """ - Test sending command to retrieve shutter/audio state - """ - test_data = 'AVMT' - log_debug_calls = [call('(111.111.111.111) reset_information() connect status is ' - '{state}'.format(state=STATUS_CODE[S_NOT_CONNECTED])), - call('(111.111.111.111) Sending {cmd} command'.format(cmd=test_data))] - - # GIVEN: Test object and mocks - with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ - patch.object(openlp.core.projectors.pjlink.PJLink, 'send_command') as mock_send_command: - pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) - - # 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_has_calls(log_debug_calls) - mock_send_command.assert_called_once_with(cmd=test_data) - - def test_projector_get_available_inputs(self): - """ - Test sending command to retrieve avaliable inputs - """ - test_data = 'INST' - log_debug_calls = [call('(111.111.111.111) reset_information() connect status is ' - '{state}'.format(state=STATUS_CODE[S_NOT_CONNECTED])), - call('(111.111.111.111) Sending {cmd} command'.format(cmd=test_data))] - - # GIVEN: Test object and mocks - with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ - patch.object(openlp.core.projectors.pjlink.PJLink, 'send_command') as mock_send_command: - pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) - - # WHEN: get_available_inputs is called - pjlink.get_available_inputs() - - # THEN: log data and send_command should have been called - mock_log.debug.assert_has_calls(log_debug_calls) - mock_send_command.assert_called_once_with(cmd=test_data) - - def test_projector_get_error_status(self): - """ - Test sending command to retrieve projector error status - """ - test_data = 'ERST' - log_debug_calls = [call('(111.111.111.111) reset_information() connect status is ' - '{state}'.format(state=STATUS_CODE[S_NOT_CONNECTED])), - call('(111.111.111.111) Sending {cmd} command'.format(cmd=test_data))] - # GIVEN: Test object and mocks - with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ - patch.object(openlp.core.projectors.pjlink.PJLink, 'send_command') as mock_send_command: - pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) - - # WHEN: get_error_status is called - pjlink.get_error_status() - - # THEN: log data and send_command should have been called - mock_log.debug.assert_has_calls(log_debug_calls) - mock_send_command.assert_called_once_with(cmd=test_data) - - def test_projector_get_input_source(self): - """ - Test sending command to retrieve current input - """ - test_data = 'INPT' - log_debug_calls = [call('(111.111.111.111) reset_information() connect status is ' - '{state}'.format(state=STATUS_CODE[S_NOT_CONNECTED])), - call('(111.111.111.111) Sending {cmd} command'.format(cmd=test_data))] - - # GIVEN: Test object and mocks - with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ - patch.object(openlp.core.projectors.pjlink.PJLink, 'send_command') as mock_send_command: - pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) - - # WHEN: get_input_source is called - pjlink.get_input_source() - - # THEN: log data and send_command should have been called - mock_log.debug.assert_has_calls(log_debug_calls) - mock_send_command.assert_called_once_with(cmd=test_data) - - def test_projector_get_lamp_status(self): - """ - Test sending command to retrieve lamp(s) status - """ - test_data = 'LAMP' - log_debug_calls = [call('(111.111.111.111) reset_information() connect status is ' - '{state}'.format(state=STATUS_CODE[S_NOT_CONNECTED])), - call('(111.111.111.111) Sending {cmd} command'.format(cmd=test_data))] - - # GIVEN: Test object and mocks - with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ - patch.object(openlp.core.projectors.pjlink.PJLink, 'send_command') as mock_send_command: - pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) - - # WHEN: get_input_source is called - pjlink.get_lamp_status() - - # THEN: log data and send_command should have been called - mock_log.debug.assert_has_calls(log_debug_calls) - mock_send_command.assert_called_once_with(cmd=test_data) - - def test_projector_get_manufacturer(self): - """ - Test sending command to retrieve manufacturer name - """ - test_data = 'INF1' - log_debug_calls = [call('(111.111.111.111) reset_information() connect status is ' - '{state}'.format(state=STATUS_CODE[S_NOT_CONNECTED])), - call('(111.111.111.111) Sending {cmd} command'.format(cmd=test_data))] - - # GIVEN: Test object and mocks - with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ - patch.object(openlp.core.projectors.pjlink.PJLink, 'send_command') as mock_send_command: - pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) - - # WHEN: get_input_source is called - pjlink.get_manufacturer() - - # THEN: log data and send_command should have been called - mock_log.debug.assert_has_calls(log_debug_calls) - mock_send_command.assert_called_once_with(cmd=test_data) - - def test_projector_get_model(self): - """ - Test sending command to get model information - """ - test_data = 'INF2' - log_debug_calls = [call('(111.111.111.111) reset_information() connect status is ' - '{state}'.format(state=STATUS_CODE[S_NOT_CONNECTED])), - call('(111.111.111.111) Sending {cmd} command'.format(cmd=test_data))] - - # GIVEN: Test object and mocks - with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ - patch.object(openlp.core.projectors.pjlink.PJLink, 'send_command') as mock_send_command: - pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) - - # WHEN: get_input_source is called - pjlink.get_model() - - # THEN: log data and send_command should have been called - mock_log.debug.assert_has_calls(log_debug_calls) - mock_send_command.assert_called_once_with(cmd=test_data) - - def test_projector_get_name(self): - """ - Test sending command to get user-assigned name - """ - test_data = 'NAME' - log_debug_calls = [call('(111.111.111.111) reset_information() connect status is ' - '{state}'.format(state=STATUS_CODE[S_NOT_CONNECTED])), - call('(111.111.111.111) Sending {cmd} command'.format(cmd=test_data))] - - # GIVEN: Test object and mocks - with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ - patch.object(openlp.core.projectors.pjlink.PJLink, 'send_command') as mock_send_command: - pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) - - # WHEN: get_input_source is called - pjlink.get_name() - - # THEN: log data and send_command should have been called - mock_log.debug.assert_has_calls(log_debug_calls) - mock_send_command.assert_called_once_with(cmd=test_data) - - def test_projector_get_other_info(self): - """ - Test sending command to retrieve other information - """ - test_data = 'INFO' - log_debug_calls = [call('(111.111.111.111) reset_information() connect status is ' - '{state}'.format(state=STATUS_CODE[S_NOT_CONNECTED])), - call('(111.111.111.111) Sending {cmd} command'.format(cmd=test_data))] - - # GIVEN: Test object and mocks - with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ - patch.object(openlp.core.projectors.pjlink.PJLink, 'send_command') as mock_send_command: - pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) - - # WHEN: get_input_source is called - pjlink.get_other_info() - - # THEN: log data and send_command should have been called - mock_log.debug.assert_has_calls(log_debug_calls) - mock_send_command.assert_called_once_with(cmd=test_data) - - def test_projector_get_power_status(self): - """ - Test sending command to retrieve current power state - """ - test_data = 'POWR' - log_debug_calls = [call('(111.111.111.111) reset_information() connect status is ' - '{state}'.format(state=STATUS_CODE[S_NOT_CONNECTED])), - call('(111.111.111.111) Sending {cmd} command'.format(cmd=test_data))] - - # GIVEN: Test object and mocks - with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ - patch.object(openlp.core.projectors.pjlink.PJLink, 'send_command') as mock_send_command: - pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) - - # WHEN: get_input_source is called - pjlink.get_power_status() - - # THEN: log data and send_command should have been called - mock_log.debug.assert_has_calls(log_debug_calls) - mock_send_command.assert_called_once_with(cmd=test_data) - - def test_projector_get_status_invalid(self): - """ - Test to check returned information for error code - """ - # GIVEN: Test object - pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) - test_string = 'NaN test' - - # WHEN: get_status called - code, message = pjlink._get_status(status=test_string) - - # THEN: Proper data should have been returned - assert code == -1, 'Should have returned -1 as a bad status check' - assert message is None, 'Invalid code type should have returned None for message' - - def test_projector_get_status_valid(self): - """ - Test to check returned information for status codes - """ - # GIVEN: Test object - test_message = 'Not Connected' - pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) - - # WHEN: get_status called - code, message = pjlink._get_status(status=S_NOT_CONNECTED) - - # THEN: Proper strings should have been returned - assert code == 'S_NOT_CONNECTED', 'Code returned should have been the same code that was sent' - assert 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(Projector(**TEST1_DATA), no_poll=True) - - # WHEN: get_status called - code, message = pjlink._get_status(status=9999) - - # THEN: Proper strings should have been returned - assert code is None, 'Code returned should have been the same code that was sent' - assert message is None, 'Should have returned None as message' - def test_projector_process_inf1(self): """ Test saving INF1 data (manufacturer) @@ -602,14 +215,14 @@ class TestPJLinkCommands(TestCase): """ Test CLSS reply has no class number """ - log_debug_calls = [call('(111.111.111.111) reset_information() connect status is ' - '{state}'.format(state=STATUS_CODE[S_NOT_CONNECTED])), - call('(111.111.111.111) Setting pjlink_class for this projector to "1"')] - log_error_calls = [call('(111.111.111.111) NAN CLSS version reply "Z" - defaulting to class "1"')] - # GIVEN: Test object with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log: pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) + log_debug_calls = [call('({ip}) reset_information() connect status is ' + '{state}'.format(ip=pjlink.name, state=STATUS_CODE[S_NOT_CONNECTED])), + call('({ip}) Setting pjlink_class for this projector to "1"'.format(ip=pjlink.name))] + log_error_calls = [call('({ip}) NAN CLSS version reply "Z" - ' + 'defaulting to class "1"'.format(ip=pjlink.name))] # WHEN: Process invalid reply pjlink.process_clss('Z') @@ -623,15 +236,14 @@ class TestPJLinkCommands(TestCase): """ Test CLSS reply has no class number """ - log_debug_calls = [call('(111.111.111.111) reset_information() connect status is ' - '{state}'.format(state=STATUS_CODE[S_NOT_CONNECTED])), - call('(111.111.111.111) Setting pjlink_class for this projector to "1"')] - log_error_calls = [call('(111.111.111.111) No numbers found in class version reply "Invalid" ' - '- defaulting to class "1"')] - # GIVEN: Test object with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log: pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) + log_debug_calls = [call('({ip}) reset_information() connect status is ' + '{state}'.format(ip=pjlink.name, state=STATUS_CODE[S_NOT_CONNECTED])), + call('({ip}) Setting pjlink_class for this projector to "1"'.format(ip=pjlink.name))] + log_error_calls = [call('({ip}) No numbers found in class version reply "Invalid" ' + '- defaulting to class "1"'.format(ip=pjlink.name))] # WHEN: Process invalid reply pjlink.process_clss('Invalid') @@ -641,6 +253,32 @@ class TestPJLinkCommands(TestCase): mock_log.error.assert_has_calls(log_error_calls) mock_log.debug.assert_has_calls(log_debug_calls) + def test_projector_process_clss_nonstandard_reply_1(self): + """ + Test CLSS request returns non-standard reply 1 + """ + # GIVEN: Test object + pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) + + # WHEN: Process non-standard reply + pjlink.process_clss('Class 1') + + # THEN: Projector class should be set with proper value + assert '1' == pjlink.pjlink_class, 'Non-standard class reply should have set class=1' + + def test_projector_process_clss_nonstandard_reply_2(self): + """ + Test CLSS request returns non-standard reply 2 + """ + # GIVEN: Test object + pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) + + # WHEN: Process non-standard reply + pjlink.process_clss('Version2') + + # THEN: Projector class should be set with proper value + assert '2' == pjlink.pjlink_class, 'Non-standard class reply should have set class=2' + def test_projector_process_erst_all_ok(self): """ Test to verify pjlink.projector_errors is set to None when no errors @@ -660,16 +298,15 @@ class TestPJLinkCommands(TestCase): """ Test test_projector_process_erst_data_invalid_length """ - chk_data = '0' * (PJLINK_ERST_DATA['DATA_LENGTH'] + 1) - log_debug_calls = [call('(111.111.111.111) reset_information() connect status is ' - '{state}'.format(state=STATUS_CODE[S_NOT_CONNECTED]))] - log_warn_calls = [call('111.111.111.111) Invalid error status response "0000000": ' - 'length != {chk}'.format(chk=PJLINK_ERST_DATA['DATA_LENGTH']))] - # GIVEN: Test object with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log: pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) pjlink.projector_errors = None + chk_data = '0' * (PJLINK_ERST_DATA['DATA_LENGTH'] + 1) + log_debug_calls = [call('({ip}) reset_information() connect status is ' + '{state}'.format(ip=pjlink.name, state=STATUS_CODE[S_NOT_CONNECTED]))] + log_warn_calls = [call('({ip}) Invalid error status response "0000000": ' + 'length != {chk}'.format(ip=pjlink.name, chk=PJLINK_ERST_DATA['DATA_LENGTH']))] # WHEN: process_erst called with invalid data (too many values pjlink.process_erst(chk_data) @@ -683,15 +320,14 @@ class TestPJLinkCommands(TestCase): """ Test test_projector_process_erst_data_invalid_nan """ - chk_data = 'Z' + ('0' * (PJLINK_ERST_DATA['DATA_LENGTH'] - 1)) - log_debug_calls = [call('(111.111.111.111) reset_information() connect status is ' - '{state}'.format(state=STATUS_CODE[S_NOT_CONNECTED]))] - log_warn_calls = [call('(111.111.111.111) Invalid error status response "Z00000"')] - # GIVEN: Test object with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log: pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) pjlink.projector_errors = None + chk_data = 'Z' + ('0' * (PJLINK_ERST_DATA['DATA_LENGTH'] - 1)) + log_debug_calls = [call('({ip}) reset_information() connect status is ' + '{state}'.format(ip=pjlink.name, state=STATUS_CODE[S_NOT_CONNECTED]))] + log_warn_calls = [call('({ip}) Invalid error status response "Z00000"'.format(ip=pjlink.name))] # WHEN: process_erst called with invalid data (too many values pjlink.process_erst(chk_data) @@ -784,14 +420,14 @@ class TestPJLinkCommands(TestCase): """ Test input source status shows current input """ - log_debug_calls = [call('(111.111.111.111) reset_information() connect status is S_NOT_CONNECTED')] - chk_source_available = ['11', '12', '21', '22', '31', '32'] - # GIVEN: Test object with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log: pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) - pjlink.source_available = chk_source_available pjlink.source = '11' + log_debug_calls = [call('({ip}) reset_information() connect status is ' + 'S_NOT_CONNECTED'.format(ip=pjlink.name))] + chk_source_available = ['11', '12', '21', '22', '31', '32'] + pjlink.source_available = chk_source_available # WHEN: Called with input source pjlink.process_inpt('21') @@ -826,15 +462,14 @@ class TestPJLinkCommands(TestCase): """ Test saving video source available information """ - log_debug_calls = [call('(111.111.111.111) Setting projector sources_available to ' - '"[\'11\', \'12\', \'21\', \'22\', \'31\', \'32\']"')] - chk_data = '21 12 11 22 32 31' # Although they should already be sorted, use unsorted to verify method - chk_test = ['11', '12', '21', '22', '31', '32'] - # GIVEN: Test object with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log: pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) pjlink.source_available = [] + log_debug_calls = [call('({ip}) Setting projector sources_available to ' + '"[\'11\', \'12\', \'21\', \'22\', \'31\', \'32\']"'.format(ip=pjlink.name))] + chk_data = '21 12 11 22 32 31' # Although they should already be sorted, use unsorted to verify method + chk_test = ['11', '12', '21', '22', '31', '32'] # WHEN: process_inst called with test data pjlink.process_inst(data=chk_data) @@ -847,13 +482,12 @@ class TestPJLinkCommands(TestCase): """ Test status multiple lamp on/off and hours """ - log_data = [call('(111.111.111.111) process_lamp(): Invalid data "11111 1 22222 0 333A3 1"')] - # GIVEN: Test object with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log: pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) pjlink.lamp = [{'Hours': 00000, 'On': True}, {'Hours': 11111, 'On': False}] + log_data = [call('({ip}) process_lamp(): Invalid data "11111 1 22222 0 333A3 1"'.format(ip=pjlink.name))] # WHEN: Call process_command with invalid lamp data pjlink.process_lamp('11111 1 22222 0 333A3 1') @@ -903,17 +537,32 @@ class TestPJLinkCommands(TestCase): assert pjlink.lamp[0]['On'] is True, 'Lamp power status should have been set to TRUE' assert 22222 == pjlink.lamp[0]['Hours'], 'Lamp hours should have been set to 22222' + def test_projector_process_lamp_single_hours_only(self): + """ + Test process lamp with 1 lamp reply hours only and no on/off status + """ + # GIVEN: Test object + pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) + pjlink.lamp = [] + + # WHEN: Process lamp command called with only hours and no lamp power state + pjlink.process_lamp("45") + + # THEN: Lamp should show hours as 45 and lamp power as Unavailable + assert 1 == len(pjlink.lamp), 'There should only be 1 lamp available' + assert 45 == pjlink.lamp[0]['Hours'], 'Lamp hours should have equalled 45' + assert pjlink.lamp[0]['On'] is None, 'Lamp power should be "None"' + def test_projector_process_name(self): """ Test saving NAME data from projector """ - chk_data = "Some Name the End-User Set IN Projector" - log_debug_calls = [call('(111.111.111.111) Setting projector PJLink name to ' - '"Some Name the End-User Set IN Projector"')] - # GIVEN: Test object with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log: pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) + chk_data = "Some Name the End-User Set IN Projector" + log_debug_calls = [call('({ip}) Setting projector PJLink name to ' + '"Some Name the End-User Set IN Projector"'.format(ip=pjlink.name))] # WHEN: process_name called with test data pjlink.process_name(data=chk_data) @@ -947,8 +596,6 @@ class TestPJLinkCommands(TestCase): """ Test process_powr invalid call """ - log_warn_calls = [call('(111.111.111.111) Unknown power response: "99"')] - # GIVEN: Test object with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \ patch.object(openlp.core.projectors.pjlink.PJLink, 'send_command') as mock_send_command, \ @@ -957,6 +604,7 @@ class TestPJLinkCommands(TestCase): pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) pjlink.power = S_STANDBY + log_warn_calls = [call('({ip}) Unknown power response: "99"'.format(ip=pjlink.name))] # WHEN: process_name called with test data pjlink.process_powr(data='99') @@ -1009,16 +657,15 @@ class TestPJLinkCommands(TestCase): """ Test saving filter type previously saved """ - filter_model = 'Filter Type Test' - log_warn_calls = [call('(111.111.111.111) Filter model already set'), - call('(111.111.111.111) Saved model: "Old filter type"'), - call('(111.111.111.111) New model: "Filter Type Test"')] - # GIVEN: Test object with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log: pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) pjlink.model_filter = 'Old filter type' + filter_model = 'Filter Type Test' + log_warn_calls = [call('({ip}) Filter model already set'.format(ip=pjlink.name)), + call('({ip}) Saved model: "Old filter type"'.format(ip=pjlink.name)), + call('({ip}) New model: "Filter Type Test"'.format(ip=pjlink.name))] # WHEN: Filter model is received pjlink.process_rfil(data=filter_model) @@ -1047,16 +694,15 @@ class TestPJLinkCommands(TestCase): """ Test saving lamp type previously saved """ - lamp_model = 'Lamp Type Test' - log_warn_calls = [call('(111.111.111.111) Lamp model already set'), - call('(111.111.111.111) Saved lamp: "Old lamp type"'), - call('(111.111.111.111) New lamp: "Lamp Type Test"')] - # GIVEN: Test object with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log: pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) pjlink.model_lamp = 'Old lamp type' + lamp_model = 'Lamp Type Test' + log_warn_calls = [call('({ip}) Lamp model already set'.format(ip=pjlink.name)), + call('({ip}) Saved lamp: "Old lamp type"'.format(ip=pjlink.name)), + call('({ip}) New lamp: "Lamp Type Test"'.format(ip=pjlink.name))] # WHEN: Filter model is received pjlink.process_rlmp(data=lamp_model) @@ -1069,14 +715,14 @@ class TestPJLinkCommands(TestCase): """ Test saving serial number from projector """ - log_debug_calls = [call('(111.111.111.111) Setting projector serial number to "Test Serial Number"')] - test_number = 'Test Serial Number' - # GIVEN: Test object with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log: pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) pjlink.serial_no = None + log_debug_calls = [call('({ip}) Setting projector serial number to ' + '"Test Serial Number"'.format(ip=pjlink.name))] + test_number = 'Test Serial Number' # WHEN: No serial number is set and we receive serial number command pjlink.process_snum(data=test_number) @@ -1089,16 +735,16 @@ class TestPJLinkCommands(TestCase): """ Test projector serial number different than saved serial number """ - log_warn_calls = [call('(111.111.111.111) Projector serial number does not match saved serial number'), - call('(111.111.111.111) Saved: "Previous serial number"'), - call('(111.111.111.111) Received: "Test Serial Number"'), - call('(111.111.111.111) NOT saving serial number')] - test_number = 'Test Serial Number' - # GIVEN: Test object with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log: pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) pjlink.serial_no = 'Previous serial number' + log_warn_calls = [call('({ip}) Projector serial number does not match ' + 'saved serial number'.format(ip=pjlink.name)), + call('({ip}) Saved: "Previous serial number"'.format(ip=pjlink.name)), + call('({ip}) Received: "Test Serial Number"'.format(ip=pjlink.name)), + call('({ip}) NOT saving serial number'.format(ip=pjlink.name))] + test_number = 'Test Serial Number' # WHEN: No serial number is set and we receive serial number command pjlink.process_snum(data=test_number) @@ -1111,14 +757,14 @@ class TestPJLinkCommands(TestCase): """ Test invalid software version information - too long """ - test_data = 'Test 1 Subtest 1' - log_debug_calls = [call('(111.111.111.111) Setting projector software version to "Test 1 Subtest 1"')] - # GIVEN: Test object with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log: pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) pjlink.sw_version = None pjlink.sw_version_received = None + test_data = 'Test 1 Subtest 1' + log_debug_calls = [call('({ip}) Setting projector software version to ' + '"Test 1 Subtest 1"'.format(ip=pjlink.name))] # WHEN: process_sver called with invalid data pjlink.process_sver(data=test_data) @@ -1131,16 +777,16 @@ class TestPJLinkCommands(TestCase): """ Test invalid software version information - Received different than saved """ - test_data_old = 'Test 1 Subtest 1' - test_data_new = 'Test 1 Subtest 2' - log_warn_calls = [call('(111.111.111.111) Projector software version does not match saved software version'), - call('(111.111.111.111) Saved: "Test 1 Subtest 1"'), - call('(111.111.111.111) Received: "Test 1 Subtest 2"'), - call('(111.111.111.111) Updating software version')] - # GIVEN: Test object with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log: pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) + test_data_old = 'Test 1 Subtest 1' + test_data_new = 'Test 1 Subtest 2' + log_warn_calls = [call('({ip}) Projector software version does not match ' + 'saved software version'.format(ip=pjlink.name)), + call('({ip}) Saved: "Test 1 Subtest 1"'.format(ip=pjlink.name)), + call('({ip}) Received: "Test 1 Subtest 2"'.format(ip=pjlink.name)), + call('({ip}) Updating software version'.format(ip=pjlink.name))] pjlink.sw_version = test_data_old # WHEN: process_sver called with invalid data @@ -1169,51 +815,3 @@ class TestPJLinkCommands(TestCase): assert pjlink.sw_version is None, 'Software version should not have changed' assert pjlink.sw_version_received is None, 'Received software version should not have changed' mock_log.warning.assert_has_calls(log_warn_calls) - - def test_projector_reset_information(self): - """ - Test reset_information() resets all information and stops timers - """ - log_debug_calls = [call('(111.111.111.111): Calling timer.stop()'), - call('(111.111.111.111): Calling socket_timer.stop()')] - - # GIVEN: Test object - with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log: - pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) - # timer and socket_timer not available until instantiation, so mock here - with patch.object(pjlink, 'socket_timer') as mock_socket_timer, \ - patch.object(pjlink, 'timer') as mock_timer: - - 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 - pjlink.reset_information() - - # THEN: All information should be reset and timers stopped - assert pjlink.power == S_OFF, 'Projector power should be OFF' - assert pjlink.pjlink_name is None, 'Projector pjlink_name should be None' - assert pjlink.manufacturer is None, 'Projector manufacturer should be None' - assert pjlink.model is None, 'Projector model should be None' - assert pjlink.shutter is None, 'Projector shutter should be None' - assert pjlink.mute is None, 'Projector shuttter should be None' - assert pjlink.lamp is None, 'Projector lamp should be None' - assert pjlink.fan is None, 'Projector fan should be None' - assert pjlink.source_available is None, 'Projector source_available should be None' - assert pjlink.source is None, 'Projector source should be None' - assert pjlink.other_info is None, 'Projector other_info should be None' - assert pjlink.send_queue == [], 'Projector send_queue should be an empty list' - assert pjlink.send_busy is False, 'Projector send_busy should be False' - assert mock_timer.stop.called is True, 'Projector timer.stop() should have been called' - assert mock_socket_timer.stop.called is True, 'Projector socket_timer.stop() should have been called' - mock_log.debug.assert_has_calls(log_debug_calls) diff --git a/tests/functional/openlp_core/projectors/test_projector_pjlink_commands_02.py b/tests/functional/openlp_core/projectors/test_projector_pjlink_commands_02.py index c9ccc2543..132a43df1 100644 --- a/tests/functional/openlp_core/projectors/test_projector_pjlink_commands_02.py +++ b/tests/functional/openlp_core/projectors/test_projector_pjlink_commands_02.py @@ -26,7 +26,7 @@ from unittest import TestCase from unittest.mock import patch, call import openlp.core.projectors.pjlink -from openlp.core.projectors.constants import S_CONNECTED +from openlp.core.projectors.constants import S_CONNECTED, S_OFF, S_ON from openlp.core.projectors.db import Projector from openlp.core.projectors.pjlink import PJLink from tests.resources.projector.data import TEST_HASH, TEST_PIN, TEST_SALT, TEST1_DATA @@ -36,17 +36,52 @@ class TestPJLinkCommands(TestCase): """ Tests for the PJLinkCommands class part 2 """ - def setUp(self): - ''' - TestPJLinkCommands part 2 initialization - ''' - self.pjlink_test = PJLink(Projector(**TEST1_DATA), no_poll=True) + def test_projector_reset_information(self): + """ + Test reset_information() resets all information and stops timers + """ + # GIVEN: Test object + with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log: + pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) + log_debug_calls = [call('({ip}): Calling poll_timer.stop()'.format(ip=pjlink.name)), + call('({ip}): Calling socket_timer.stop()'.format(ip=pjlink.name))] + # timer and socket_timer not available until instantiation, so mock here + with patch.object(pjlink, 'socket_timer') as mock_socket_timer, \ + patch.object(pjlink, 'poll_timer') as mock_timer: - def tearDown(self): - ''' - TestPJLinkCommands part 2 cleanups - ''' - self.pjlink_test = None + 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 + pjlink.reset_information() + + # THEN: All information should be reset and timers stopped + assert pjlink.power == S_OFF, 'Projector power should be OFF' + assert pjlink.pjlink_name is None, 'Projector pjlink_name should be None' + assert pjlink.manufacturer is None, 'Projector manufacturer should be None' + assert pjlink.model is None, 'Projector model should be None' + assert pjlink.shutter is None, 'Projector shutter should be None' + assert pjlink.mute is None, 'Projector shuttter should be None' + assert pjlink.lamp is None, 'Projector lamp should be None' + assert pjlink.fan is None, 'Projector fan should be None' + assert pjlink.source_available is None, 'Projector source_available should be None' + assert pjlink.source is None, 'Projector source should be None' + assert pjlink.other_info is None, 'Projector other_info should be None' + assert pjlink.send_queue == [], 'Projector send_queue should be an empty list' + assert pjlink.send_busy is False, 'Projector send_busy should be False' + assert mock_timer.stop.called is True, 'Projector timer.stop() should have been called' + assert mock_socket_timer.stop.called is True, 'Projector socket_timer.stop() should have been called' + mock_log.debug.assert_has_calls(log_debug_calls) def test_process_pjlink_normal(self): """ @@ -54,13 +89,14 @@ class TestPJLinkCommands(TestCase): """ # GIVEN: Initial mocks and data mock_log = patch.object(openlp.core.projectors.pjlink, "log").start() - mock_disconnect_from_host = patch.object(self.pjlink_test, 'disconnect_from_host').start() - mock_send_command = patch.object(self.pjlink_test, 'send_command').start() - mock_readyRead = patch.object(self.pjlink_test, 'readyRead').start() - mock_change_status = patch.object(self.pjlink_test, 'change_status').start() - pjlink = self.pjlink_test + mock_disconnect_from_host = patch('openlp.core.projectors.pjlink.PJLink.disconnect_from_host').start() + mock_send_command = patch('openlp.core.projectors.pjlink.PJLink.send_command').start() + mock_readyRead = patch('openlp.core.projectors.pjlink.PJLink.readyRead').start() + mock_change_status = patch('openlp.core.projectors.pjlink.PJLink.change_status').start() + + pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) pjlink.pin = None - log_check = [call("({111.111.111.111}) process_pjlink(): Sending 'CLSS' initial command'"), ] + log_check = [call('({ip}) process_pjlink(): Sending "CLSS" initial command'.format(ip=pjlink.name)), ] # WHEN: process_pjlink called with no authentication required pjlink.process_pjlink(data="0") @@ -68,7 +104,7 @@ class TestPJLinkCommands(TestCase): # THEN: proper processing should have occured mock_log.debug.has_calls(log_check) mock_disconnect_from_host.assert_not_called() - self.assertEqual(mock_readyRead.connect.call_count, 1, 'Should have only been called once') + assert 1 == mock_readyRead.connect.call_count, 'Should have only been called once' mock_change_status.assert_called_once_with(S_CONNECTED) mock_send_command.assert_called_with(cmd='CLSS', priority=True, salt=None) @@ -78,13 +114,14 @@ class TestPJLinkCommands(TestCase): """ # GIVEN: Initial mocks and data mock_log = patch.object(openlp.core.projectors.pjlink, "log").start() - mock_disconnect_from_host = patch.object(self.pjlink_test, 'disconnect_from_host').start() - mock_send_command = patch.object(self.pjlink_test, 'send_command').start() - mock_readyRead = patch.object(self.pjlink_test, 'readyRead').start() - mock_change_status = patch.object(self.pjlink_test, 'change_status').start() - pjlink = self.pjlink_test + mock_disconnect_from_host = patch('openlp.core.projectors.pjlink.PJLink.disconnect_from_host').start() + mock_send_command = patch('openlp.core.projectors.pjlink.PJLink.send_command').start() + mock_readyRead = patch('openlp.core.projectors.pjlink.PJLink.readyRead').start() + mock_change_status = patch('openlp.core.projectors.pjlink.PJLink.change_status').start() + + pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) pjlink.pin = TEST_PIN - log_check = [call("({111.111.111.111}) process_pjlink(): Sending 'CLSS' initial command'"), ] + log_check = [call('({ip}) process_pjlink(): Sending "CLSS" initial command'.format(ip=pjlink.name)), ] # WHEN: process_pjlink called with no authentication required pjlink.process_pjlink(data='1 {salt}'.format(salt=TEST_SALT)) @@ -92,7 +129,7 @@ class TestPJLinkCommands(TestCase): # THEN: proper processing should have occured mock_log.debug.has_calls(log_check) mock_disconnect_from_host.assert_not_called() - self.assertEqual(mock_readyRead.connect.call_count, 1, 'Should have only been called once') + assert 1 == mock_readyRead.connect.call_count, 'Should have only been called once' mock_change_status.assert_called_once_with(S_CONNECTED) mock_send_command.assert_called_with(cmd='CLSS', priority=True, salt=TEST_HASH) @@ -101,20 +138,20 @@ class TestPJLinkCommands(TestCase): Test process_pjlinnk called with no authentication but pin is set """ # GIVEN: Initial mocks and data - # GIVEN: Initial mocks and data mock_log = patch.object(openlp.core.projectors.pjlink, 'log').start() - mock_disconnect_from_host = patch.object(self.pjlink_test, 'disconnect_from_host').start() - mock_send_command = patch.object(self.pjlink_test, 'send_command').start() - pjlink = self.pjlink_test + mock_disconnect_from_host = patch('openlp.core.projectors.pjlink.PJLink.disconnect_from_host').start() + mock_send_command = patch('openlp.core.projectors.pjlink.PJLink.send_command').start() + + pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) pjlink.pin = TEST_PIN - log_check = [call('(111.111.111.111) Normal connection but PIN set - aborting'), ] + log_check = [call('({ip}) Normal connection but PIN set - aborting'.format(ip=pjlink.name)), ] # WHEN: process_pjlink called with invalid authentication scheme pjlink.process_pjlink(data='0') # THEN: Proper calls should be made mock_log.error.assert_has_calls(log_check) - self.assertEqual(mock_disconnect_from_host.call_count, 1, 'Should have only been called once') + assert 1 == mock_disconnect_from_host.call_count, 'Should have only been called once' mock_send_command.assert_not_called() def test_process_pjlink_normal_with_salt_error(self): @@ -122,20 +159,20 @@ class TestPJLinkCommands(TestCase): Test process_pjlinnk called with no authentication but pin is set """ # GIVEN: Initial mocks and data - # GIVEN: Initial mocks and data mock_log = patch.object(openlp.core.projectors.pjlink, 'log').start() - mock_disconnect_from_host = patch.object(self.pjlink_test, 'disconnect_from_host').start() - mock_send_command = patch.object(self.pjlink_test, 'send_command').start() - pjlink = self.pjlink_test + mock_disconnect_from_host = patch('openlp.core.projectors.pjlink.PJLink.disconnect_from_host').start() + mock_send_command = patch('openlp.core.projectors.pjlink.PJLink.send_command').start() + + pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) pjlink.pin = TEST_PIN - log_check = [call('(111.111.111.111) Normal connection with extra information - aborting'), ] + log_check = [call('({ip}) Normal connection with extra information - aborting'.format(ip=pjlink.name)), ] # WHEN: process_pjlink called with invalid authentication scheme pjlink.process_pjlink(data='0 {salt}'.format(salt=TEST_SALT)) # THEN: Proper calls should be made mock_log.error.assert_has_calls(log_check) - self.assertEqual(mock_disconnect_from_host.call_count, 1, 'Should have only been called once') + assert 1 == mock_disconnect_from_host.call_count, 'Should have only been called once' mock_send_command.assert_not_called() def test_process_pjlink_invalid_authentication_scheme_length_error(self): @@ -144,17 +181,18 @@ class TestPJLinkCommands(TestCase): """ # GIVEN: Initial mocks and data mock_log = patch.object(openlp.core.projectors.pjlink, 'log').start() - mock_disconnect_from_host = patch.object(self.pjlink_test, 'disconnect_from_host').start() - mock_send_command = patch.object(self.pjlink_test, 'send_command').start() - pjlink = self.pjlink_test - log_check = [call('(111.111.111.111) Invalid initial authentication scheme - aborting'), ] + mock_disconnect_from_host = patch('openlp.core.projectors.pjlink.PJLink.disconnect_from_host').start() + mock_send_command = patch('openlp.core.projectors.pjlink.PJLink.send_command').start() + + pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) + log_check = [call('({ip}) Invalid initial authentication scheme - aborting'.format(ip=pjlink.name)), ] # WHEN: process_pjlink called with invalid authentication scheme pjlink.process_pjlink(data='01') # THEN: socket should be closed and invalid data logged mock_log.error.assert_has_calls(log_check) - self.assertEqual(mock_disconnect_from_host.call_count, 1, 'Should have only been called once') + assert 1 == mock_disconnect_from_host.call_count, 'Should have only been called once' mock_send_command.assert_not_called() def test_process_pjlink_invalid_authentication_data_length_error(self): @@ -163,17 +201,18 @@ class TestPJLinkCommands(TestCase): """ # GIVEN: Initial mocks and data mock_log = patch.object(openlp.core.projectors.pjlink, 'log').start() - mock_disconnect_from_host = patch.object(self.pjlink_test, 'disconnect_from_host').start() - mock_send_command = patch.object(self.pjlink_test, 'send_command').start() - log_check = [call('(111.111.111.111) Authenticated connection but not enough info - aborting'), ] - pjlink = self.pjlink_test + mock_disconnect_from_host = patch('openlp.core.projectors.pjlink.PJLink.disconnect_from_host').start() + mock_send_command = patch('openlp.core.projectors.pjlink.PJLink.send_command').start() + + pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) + log_check = [call('({ip}) Authenticated connection but not enough info - aborting'.format(ip=pjlink.name)), ] # WHEN: process_pjlink called with no salt pjlink.process_pjlink(data='1') # THEN: socket should be closed and invalid data logged mock_log.error.assert_has_calls(log_check) - self.assertEqual(mock_disconnect_from_host.call_count, 1, 'Should have only been called once') + assert 1 == mock_disconnect_from_host.call_count, 'Should have only been called once' mock_send_command.assert_not_called() def test_process_pjlink_authenticate_pin_not_set_error(self): @@ -182,16 +221,17 @@ class TestPJLinkCommands(TestCase): """ # GIVEN: Initial mocks and data mock_log = patch.object(openlp.core.projectors.pjlink, 'log').start() - mock_disconnect_from_host = patch.object(self.pjlink_test, 'disconnect_from_host').start() - mock_send_command = patch.object(self.pjlink_test, 'send_command').start() - log_check = [call('(111.111.111.111) Authenticate connection but no PIN - aborting'), ] - pjlink = self.pjlink_test + mock_disconnect_from_host = patch('openlp.core.projectors.pjlink.PJLink.disconnect_from_host').start() + mock_send_command = patch('openlp.core.projectors.pjlink.PJLink.send_command').start() + + pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) pjlink.pin = None + log_check = [call('({ip}) Authenticate connection but no PIN - aborting'.format(ip=pjlink.name)), ] # WHEN: process_pjlink called with no salt pjlink.process_pjlink(data='1 {salt}'.format(salt=TEST_SALT)) # THEN: socket should be closed and invalid data logged mock_log.error.assert_has_calls(log_check) - self.assertEqual(mock_disconnect_from_host.call_count, 1, 'Should have only been called once') + assert 1 == mock_disconnect_from_host.call_count, 'Should have only been called once' mock_send_command.assert_not_called()