openlp-core team mailing list archive
-
openlp-core team
-
Mailing list archive
-
Message #32620
[Merge] lp:~alisonken1/openlp/pjlink2-o into lp:openlp
Ken Roberts has proposed merging lp:~alisonken1/openlp/pjlink2-o into lp:openlp.
Commit message:
PJLink2 update o
Requested reviews:
Raoul Snyman (raoul-snyman)
For more details, see:
https://code.launchpad.net/~alisonken1/openlp/pjlink2-o/+merge/336067
- Ongoing conversions from self.asserts to plain asserts
- 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 pjlink_class to Projector() table in preps for UDP update
- Fix db tests
- Replace log strings from ip=self.ip to ip=self.entry.name
- Convert pjlink from threading to non-threading
- Updates to tests for new testing format and log strings
- Update db Projector.pjlink_class field size
--------------------------------------------------------------------------------
lp:~alisonken1/openlp/pjlink2-o (revision 2809)
https://ci.openlp.io/job/Branch-01-Pull/2420/ [SUCCESS]
https://ci.openlp.io/job/Branch-02a-Linux-Tests/2321/ [SUCCESS]
https://ci.openlp.io/job/Branch-02b-macOS-Tests/116/ [SUCCESS]
https://ci.openlp.io/job/Branch-03a-Build-Source/38/ [SUCCESS]
https://ci.openlp.io/job/Branch-03b-Build-macOS/37/ [SUCCESS]
https://ci.openlp.io/job/Branch-04a-Code-Analysis/1500/ [SUCCESS]
https://ci.openlp.io/job/Branch-04b-Test-Coverage/1313/ [SUCCESS]
https://ci.openlp.io/job/Branch-05-AppVeyor-Tests/264/ [FAILURE]
--
Your team OpenLP Core is subscribed to branch lp:openlp.
=== modified file 'openlp/core/projectors/db.py'
--- openlp/core/projectors/db.py 2017-12-29 09:15:48 +0000
+++ openlp/core/projectors/db.py 2018-01-13 07:34:12 +0000
@@ -152,6 +152,7 @@
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 @@
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 @@
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 @@
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))
=== modified file 'openlp/core/projectors/manager.py'
--- openlp/core/projectors/manager.py 2018-01-04 06:10:20 +0000
+++ openlp/core/projectors/manager.py 2018-01-13 07:34:12 +0000
@@ -522,8 +522,8 @@
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 @@
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 @@
"""
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()
=== modified file 'openlp/core/projectors/pjlink.py'
--- openlp/core/projectors/pjlink.py 2018-01-04 06:10:20 +0000
+++ openlp/core/projectors/pjlink.py 2018-01-13 07:34:12 +0000
@@ -58,8 +58,8 @@
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 @@
'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 @@
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 @@
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 @@
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 @@
: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 @@
_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 @@
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 @@
'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 @@
# : 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 @@
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 @@
"""
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 @@
: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 @@
: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 @@
: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 @@
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 @@
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 @@
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 @@
: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 @@
: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 @@
: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 @@
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 @@
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 @@
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 @@
: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 @@
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 @@
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 @@
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 @@
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 @@
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 @@
"""
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 @@
# 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 @@
: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 @@
# '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 @@
: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 @@
"""
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 @@
# 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 @@
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 @@
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 @@
: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 @@
: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 @@
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 @@
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 @@
: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 @@
"""
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 @@
"""
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 @@
"""
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 @@
: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 @@
"""
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 @@
"""
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 @@
"""
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 @@
"""
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()
=== modified file 'openlp/core/projectors/upgrade.py'
--- openlp/core/projectors/upgrade.py 2017-12-29 09:15:48 +0000
+++ openlp/core/projectors/upgrade.py 2018-01-13 07:34:12 +0000
@@ -33,7 +33,7 @@
log = logging.getLogger(__name__)
# Initial projector DB was unversioned
-__version__ = 2
+__version__ = 3
log.debug('Projector DB upgrade module loading')
@@ -71,3 +71,23 @@
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'))
=== modified file 'tests/functional/openlp_core/projectors/test_projector_bugfixes_01.py'
--- tests/functional/openlp_core/projectors/test_projector_bugfixes_01.py 2017-12-28 08:22:55 +0000
+++ tests/functional/openlp_core/projectors/test_projector_bugfixes_01.py 2018-01-13 07:34:12 +0000
@@ -24,57 +24,32 @@
"""
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
-
- # 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):
+ # Test now part of test_projector_pjlink_commands_01
+ # Keeping here for bug reference
+ pass
+
+ 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
-
- # 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):
+ # Test now part of test_projector_pjlink_commands_01
+ # Keeping here for bug reference
+ pass
+
+ def bug_1593882_no_pin_authenticated_connection(self):
"""
Test bug 1593882 no pin and authenticated request exception
"""
@@ -82,7 +57,7 @@
# 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 @@
# Keeping here for bug reference
pass
- def test_bug_1734275_process_lamp_nonstandard_reply(self):
- """
- Test bugfix 17342785 non-standard LAMP response
- """
- # 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"')
+ def bug_1734275_process_lamp_nonstandard_reply(self):
+ """
+ Test bugfix 17342785 non-standard LAMP response with one lamp hours only
+ """
+ # Test now part of test_projector_pjlink_commands_01
+ # Keeping here for bug reference
+ pass
=== modified file 'tests/functional/openlp_core/projectors/test_projector_db.py'
--- tests/functional/openlp_core/projectors/test_projector_db.py 2017-12-28 08:22:55 +0000
+++ tests/functional/openlp_core/projectors/test_projector_db.py 2018-01-13 07:34:12 +0000
@@ -107,7 +107,7 @@
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 @@
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 @@
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 @@
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 @@
# 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 @@
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 @@
# 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 @@
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 @@
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 @@
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 @@
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 @@
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 @@
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 @@
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 @@
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 @@
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 @@
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 @@
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 @@
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 @@
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 @@
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'
=== modified file 'tests/functional/openlp_core/projectors/test_projector_pjlink_base.py'
--- tests/functional/openlp_core/projectors/test_projector_pjlink_base.py 2018-01-03 00:35:14 +0000
+++ tests/functional/openlp_core/projectors/test_projector_pjlink_base.py 2018-01-13 07:34:12 +0000
@@ -25,7 +25,19 @@
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 @@
"""
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
-
- # WHEN: process_command is called with "ERR2" status from projector
- pjlink.process_command('POWR', 'ERR2')
-
- # THEN: change_status should have called change_status with E_UNDEFINED
- # as first parameter
- mock_change_status.called_with(E_PARAMETER,
- 'change_status should have been called with "{}"'.format(
- STATUS_CODE[E_PARAMETER]))
-
- @patch.object(pjlink_test, 'disconnect_from_host')
- def test_socket_abort(self, mock_disconnect):
+ with patch('openlp.core.projectors.pjlink.PJLink.changeStatus') as mock_changeStatus:
+
+ pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
+
+ # WHEN: process_command is called with "ERR2" status from projector
+ pjlink.process_command('POWR', 'ERR2')
+
+ # THEN: change_status should have called change_status with E_UNDEFINED
+ # as first parameter
+ mock_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
-
- # 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')
+ 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()
+
+ # 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 @@
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()
-
- 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'),
- ]
-
- # WHEN: PJLink.poll_loop is called
- pjlink.poll_loop()
-
- # THEN: proper calls were made to retrieve projector data
- # First, call to update the timer with the next interval
- self.assertTrue(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')
+ with patch('openlp.core.projectors.pjlink.PJLink.send_command') as mock_send_command:
+
+ 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
+
+ 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'),
+ ]
+
+ # 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'
=== modified file 'tests/functional/openlp_core/projectors/test_projector_pjlink_cmd_routing.py'
--- tests/functional/openlp_core/projectors/test_projector_pjlink_cmd_routing.py 2017-12-26 04:14:39 +0000
+++ tests/functional/openlp_core/projectors/test_projector_pjlink_cmd_routing.py 2018-01-13 07:34:12 +0000
@@ -44,15 +44,16 @@
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 @@
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 @@
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 @@
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 @@
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 @@
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 @@
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 @@
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 @@
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 @@
"""
# 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')
=== modified file 'tests/functional/openlp_core/projectors/test_projector_pjlink_commands_01.py'
--- tests/functional/openlp_core/projectors/test_projector_pjlink_commands_01.py 2018-01-03 00:35:14 +0000
+++ tests/functional/openlp_core/projectors/test_projector_pjlink_commands_01.py 2018-01-13 07:34:12 +0000
@@ -26,9 +26,17 @@
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 @@
"""
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 @@
"""
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 @@
"""
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 @@
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 @@
"""
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 @@
"""
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 @@
"""
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 = '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
- pjlink.source = '11'
# WHEN: Called with input source
pjlink.process_inpt('21')
@@ -826,15 +462,14 @@
"""
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 @@
"""
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 @@
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 @@
"""
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 @@
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 @@
"""
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 @@
"""
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 @@
"""
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 @@
"""
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 @@
"""
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 @@
"""
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 @@
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)
=== modified file 'tests/functional/openlp_core/projectors/test_projector_pjlink_commands_02.py'
--- tests/functional/openlp_core/projectors/test_projector_pjlink_commands_02.py 2017-12-28 08:22:55 +0000
+++ tests/functional/openlp_core/projectors/test_projector_pjlink_commands_02.py 2018-01-13 07:34:12 +0000
@@ -26,7 +26,7 @@
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 @@
"""
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 tearDown(self):
- '''
- TestPJLinkCommands part 2 cleanups
- '''
- self.pjlink_test = None
+ 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:
+
+ 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 @@
"""
# 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 @@
# 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 @@
"""
# 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 @@
# 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 @@
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 @@
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 @@
"""
# 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 @@
"""
# 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 @@
"""
# 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()
Follow ups
