raoul/openlp
1
0
Fork 0
forked from openlp/openlp
Code Issues Pull Requests Projects Releases Wiki Activity

- Ongoing conversions from self.asserts to plain asserts

Browse Source 
- Remove test_ part of method name for bugfix tests no longer used
- Move tests *_change_status_* from pjlink_commands_01 to pjlink_base
- Move test projector_reset_information from pjlink_commands_01 to pjlink_commands_02
- Move lamp_nonstandard_reply from bugfix to pjlink_commands_01
- Move clss_nonstandard_reply 1/2 from bugfix to pjlink_commands_01
- Move test_projector_get_* tests from pjlink_commands_01 to pjlink_base
- Added pjli...

bzr-revno: 2809
This commit is contained in:
Ken Roberts 2018-01-13 17:44:36 -07:00 committed by Raoul Snyman
commit d6087813ae
10 changed files with 920 additions and 1018 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
openlp/core/projectors/db.py
View File

@ -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(5)) # 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(5))
manufacturer = Column(String(128))
model = Column(String(128))
other = Column(String(128))

30
openlp/core/projectors/manager.py
View File

@ -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()

388
openlp/core/projectors/pjlink.py
View File

@ -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()

22
openlp/core/projectors/upgrade.py
View File

@ -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(5), server_default=null()))
log.debug('{status} projector DB upgrade to version 3'.format(status='Updated' if upgrade_db else 'Skipping'))

68
tests/functional/openlp_core/projectors/test_projector_bugfixes_01.py
View File

@ -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

78
tests/functional/openlp_core/projectors/test_projector_db.py
View File

@ -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(name="OpenLP Test")>',
'Manufacturer.__repr__() should have returned a proper representation string')
assert str(manufacturer) == '<Manufacturer(name="OpenLP Test")>', \
'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(name='"OpenLP Test"')>',
'Model.__repr__() should have returned a proper representation string')
assert str(model) == '<Model(name='"OpenLP Test"')>', \
'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(pjlink_name="Test object", pjlink_code="11", text="Input text")>',
'Source.__repr__() should have returned a proper representation string')
assert str(source) == '<Source(pjlink_name="Test object", pjlink_code="11", text="Input text")>', \
'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(id="1", code="11", text="First RGB source", projector_id="1")>',
'ProjectorSource.__repr__)_ should have returned a proper representation string')
assert str(source) == '<ProjectorSource(id="1", code="11", text="First RGB source", projector_id="1")>', \
'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'

520
tests/functional/openlp_core/projectors/test_projector_pjlink_base.py
View File

@ -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'

57
tests/functional/openlp_core/projectors/test_projector_pjlink_cmd_routing.py
View File

@ -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')

626
tests/functional/openlp_core/projectors/test_projector_pjlink_commands_01.py
View File

@ -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)

144
tests/functional/openlp_core/projectors/test_projector_pjlink_commands_02.py
View File

@ -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()