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

PJLink2 Update v06 

bzr-revno: 2869
This commit is contained in:
Ken Roberts 2019-05-07 17:57:35 +01:00 committed by Tim Bentley
parent 984e79faa9 855828939f
commit b4d5425887
7 changed files with 465 additions and 406 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -520,14 +520,23 @@ class PJLink(QtNetwork.QTcpSocket):
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.entry.name, buff=read))
if read == -1:
self.socket_timer.start()
while self.bytesAvailable() >= 1:
data = self.readLine(1024)
data = data.strip()
if not data:
log.warning('({ip}) get_socket(): Ignoring empty packet'.format(ip=self.entry.name))
if self.bytesAvailable() < 1:
break
self.socket_timer.stop()
if data:
log.debug('({ip}) get_socket(): "{buff}"'.format(ip=self.entry.name, buff=data))
if data == -1:
# No data available
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)
return
return self.get_data(buff=data)
def get_data(self, buff, *args, **kwargs):
"""
@ -540,21 +549,22 @@ class PJLink(QtNetwork.QTcpSocket):
# NOTE: Class2 has changed to some values being UTF-8
data_in = decode(buff, 'utf-8') if isinstance(buff, bytes) else buff
data = data_in.strip()
self.receive_data_signal()
# Initial packet checks
if (len(data) < 7):
self._trash_buffer(msg='get_data(): Invalid packet - length')
return self.receive_data_signal()
return
elif len(data) > self.max_size:
self._trash_buffer(msg='get_data(): Invalid packet - too long ({length} bytes)'.format(length=len(data)))
return self.receive_data_signal()
return
elif not data.startswith(PJLINK_PREFIX):
self._trash_buffer(msg='get_data(): Invalid packet - PJLink prefix missing')
return self.receive_data_signal()
return
elif data[6] != '=' and data[8] != '=':
# data[6] = standard command packet
# data[8] = initial PJLink connection (after mangling)
self._trash_buffer(msg='get_data(): Invalid reply - Does not have "="')
return self.receive_data_signal()
return
log.debug('({ip}) get_data(): Checking new data "{data}"'.format(ip=self.entry.name, data=data))
header, data = data.split('=')
log.debug('({ip}) get_data() header="{header}" data="{data}"'.format(ip=self.entry.name,
@ -572,20 +582,20 @@ class PJLink(QtNetwork.QTcpSocket):
data=data))
if cmd not in PJLINK_VALID_CMD:
self._trash_buffer('get_data(): Invalid packet - unknown command "{data}"'.format(data=cmd))
return self.receive_data_signal()
return
elif version not in PJLINK_VALID_CMD[cmd]['version']:
self._trash_buffer(msg='get_data() Command reply version does not match a valid command version')
return self.receive_data_signal()
return
elif int(self.pjlink_class) < int(version):
if not ignore_class:
log.warning('({ip}) get_data(): Projector returned class reply higher '
'than projector stated class'.format(ip=self.entry.name))
return self.receive_data_signal()
return
chk = process_command(self, cmd, data)
if chk is None:
# Command processed normally and not initial connection, so skip other checks
return self.receive_data_signal()
return
# PJLink initial connection checks
elif chk == S_DATA_OK:
# Previous command returned OK
@ -612,7 +622,7 @@ class PJLink(QtNetwork.QTcpSocket):
self.projectorAuthentication.emit(self.entry.name)
self.change_status(status=E_AUTHENTICATION)
return self.receive_data_signal()
return
@QtCore.pyqtSlot(QtNetwork.QAbstractSocket.SocketError)
def get_error(self, err):
@ -728,17 +738,17 @@ class PJLink(QtNetwork.QTcpSocket):
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.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.entry.name))
else:
if not self.priority_queue and not self.send_queue:
# No data to send
log.warning('({ip}) _send_command(): No data to send'.format(ip=self.entry.name))
self.send_busy = False
return
elif self.priority_queue:
out = self.priority_queue.pop(0)
log.debug('({ip}) _send_command(): Getting priority queued packet'.format(ip=self.entry.name))
elif self.send_queue:
out = self.send_queue.pop(0)
log.debug('({ip}) _send_command(): Getting normal queued packet'.format(ip=self.entry.name))
self.send_busy = True
log.debug('({ip}) _send_command(): Sending "{data}"'.format(ip=self.entry.name, data=out.strip()))
self.socket_timer.start()

46
openlp/core/projectors/pjlinkcommands.py
View File

@ -153,14 +153,14 @@ def process_clss(projector, data):
# fix the class reply is to just remove all non-digit characters.
chk = re.findall(r'\d', data)
if len(chk) < 1:
log.error('({ip}) No numbers found in class version reply "{data}" - '
'defaulting to class "1"'.format(ip=projector.entry.name, data=data))
log.warning('({ip}) No numbers found in class version reply "{data}" - '
'defaulting to class "1"'.format(ip=projector.entry.name, data=data))
clss = '1'
else:
clss = chk[0] # Should only be the first match
elif not data.isdigit():
log.error('({ip}) NAN CLSS version reply "{data}" - '
'defaulting to class "1"'.format(ip=projector.entry.name, data=data))
log.warning('({ip}) NAN CLSS version reply "{data}" - '
'defaulting to class "1"'.format(ip=projector.entry.name, data=data))
clss = '1'
else:
clss = data
@ -282,12 +282,13 @@ def process_inpt(projector, data):
if projector.source_available is not None:
# We have available inputs, so verify it's in the list
if data not in projector.source_available:
log.warn('({ip}) Input source not listed in available sources - ignoring'.format(ip=projector.entry.name))
log.warning('({ip}) Input source not listed in available sources - '
'ignoring'.format(ip=projector.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=projector.entry.name))
log.warning('({ip}) Input source not listed as a PJLink valid source '
'- ignoring'.format(ip=projector.entry.name))
return
projector.source = data
log.debug('({ip}) Setting current source to "{data}"'.format(ip=projector.entry.name, data=projector.source))
@ -326,7 +327,10 @@ def process_lamp(projector, data):
lamps = []
lamp_list = data.split()
if len(lamp_list) < 2:
lamps.append({'Hours': int(lamp_list[0]), 'On': None})
# Invalid data - not enough information
log.warning('({ip}) process_lamp(): Invalid data "{data}" - '
'Missing data'.format(ip=projector.entry.name, data=data))
return
else:
while lamp_list:
if not lamp_list[0].isnumeric() or not lamp_list[1].isnumeric():
@ -411,20 +415,22 @@ def process_powr(projector, data):
:param projector: Projector instance
:param data: Power status
"""
log.debug('({ip}: Processing POWR command'.format(ip=projector.entry.name))
if data in PJLINK_POWR_STATUS:
power = PJLINK_POWR_STATUS[data]
update_icons = projector.power != power
projector.power = power
projector.change_status(PJLINK_POWR_STATUS[data])
if update_icons:
projector.projectorUpdateIcons.emit()
# Update the input sources available
if power == S_ON:
projector.send_command('INST')
else:
log.debug('({ip}) Processing POWR command'.format(ip=projector.entry.name))
if data not in PJLINK_POWR_STATUS:
# Log unknown status response
log.warning('({ip}) Unknown power response: "{data}"'.format(ip=projector.entry.name, data=data))
return
power = PJLINK_POWR_STATUS[data]
update_icons = projector.power != power
if update_icons:
projector.power = power
projector.change_status(PJLINK_POWR_STATUS[data])
projector.projectorUpdateIcons.emit()
if power == S_ON:
# Input sources list should only be available after power on, so update here
projector.send_command('INST')
if projector.power in [S_ON, S_STANDBY, S_OFF] and 'POWR' in projector.status_timer_checks:
projector.status_timer_delete(cmd='POWR')
return

12
tests/openlp_core/projectors/test_projector_commands_01.py
View File

@ -246,8 +246,8 @@ class TestPJLinkCommands(TestCase):
Test CLSS reply has no class number
"""
# GIVEN: Test setup
log_error_calls = [call('({ip}) NAN CLSS version reply "Z" - '
'defaulting to class "1"'.format(ip=self.pjlink.name))]
log_warning_calls = [call('({ip}) NAN CLSS version reply "Z" - '
'defaulting to class "1"'.format(ip=self.pjlink.name))]
log_debug_calls = [call('({ip}) Processing command "CLSS" with data "Z"'.format(ip=self.pjlink.name)),
call('({ip}) Calling function for CLSS'.format(ip=self.pjlink.name)),
call('({ip}) Setting pjlink_class for this projector to "1"'.format(ip=self.pjlink.name))]
@ -257,7 +257,7 @@ class TestPJLinkCommands(TestCase):
# THEN: Projector class should be set with default value
assert (self.pjlink.pjlink_class == '1'), 'Invalid NaN class reply should have set class=1'
mock_log.error.assert_has_calls(log_error_calls)
mock_log.warning.assert_has_calls(log_warning_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
@ -266,8 +266,8 @@ class TestPJLinkCommands(TestCase):
Test CLSS reply has no class number
"""
# GIVEN: Test object
log_error_calls = [call('({ip}) No numbers found in class version reply "Invalid" '
'- defaulting to class "1"'.format(ip=self.pjlink.name))]
log_warning_calls = [call('({ip}) No numbers found in class version reply "Invalid" '
'- defaulting to class "1"'.format(ip=self.pjlink.name))]
log_debug_calls = [call('({ip}) Processing command "CLSS" with data "Invalid"'.format(ip=self.pjlink.name)),
call('({ip}) Calling function for CLSS'.format(ip=self.pjlink.name)),
call('({ip}) Setting pjlink_class for this projector to "1"'.format(ip=self.pjlink.name))]
@ -277,7 +277,7 @@ class TestPJLinkCommands(TestCase):
# THEN: Projector class should be set with default value
assert (self.pjlink.pjlink_class == '1'), 'Invalid class reply should have set class=1'
mock_log.error.assert_has_calls(log_error_calls)
mock_log.warning.assert_has_calls(log_warning_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')

708
tests/openlp_core/projectors/test_projector_commands_02.py
View File

@ -22,7 +22,7 @@
"""
Package to test the openlp.core.projectors.pjlink commands package.
"""
from unittest import TestCase, skip
from unittest import TestCase
from unittest.mock import call, patch
import openlp.core.projectors.pjlink
@ -49,428 +49,502 @@ class TestPJLinkCommands(TestCase):
"""
del(self.pjlink)
@skip('Needs update to new setup')
def test_projector_process_inpt_valid(self):
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_inpt_good(self, mock_log):
"""
Test input source status shows current input
"""
# 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
self.pjlink.source = '11'
log_warning_calls = []
log_debug_calls = [call('({ip}) Processing command "INPT" with data "21"'.format(ip=self.pjlink.name)),
call('({ip}) Calling function for INPT'.format(ip=self.pjlink.name)),
call('({ip}) Setting current source to "21"'.format(ip=self.pjlink.name))]
chk_source_available = ['11', '12', '21', '22', '31', '32']
self.pjlink.source_available = chk_source_available
# WHEN: Called with input source
process_command(projector=self.pjlink, cmd='INPT', data='21')
# WHEN: Called with input source
process_command(projector=self.pjlink, cmd='INPT', data='21')
# THEN: Input selected should reflect current input
assert pjlink.source == '21', 'Input source should be set to "21"'
mock_log.debug.assert_has_calls(log_debug_calls)
# THEN: Input selected should reflect current input
assert ('21' == self.pjlink.source), 'Input source should be set to "21"'
mock_log.warning.assert_has_calls(log_warning_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
@skip('Needs update to new setup')
def test_projector_process_input_not_in_list(self):
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_inpt_invalid(self, mock_log):
"""
Test setting input outside of available inputs
TODO: Future test
Test input source returned not valid according to standard
"""
pass
# GIVEN: Test object
log_warning_calls = [call('({ip}) Input source not listed as a PJLink valid source - '
'ignoring'.format(ip=self.pjlink.name))]
log_debug_calls = [call('({ip}) Processing command "INPT" with data "91"'.format(ip=self.pjlink.name)),
call('({ip}) Calling function for INPT'.format(ip=self.pjlink.name))]
self.pjlink.source = None
self.pjlink.source_available = None
@skip('Needs update to new setup')
def test_projector_process_input_not_in_default(self):
"""
Test setting input with no sources available
TODO: Future test
"""
pass
# WHEN: Called with input source
process_command(projector=self.pjlink, cmd='INPT', data='91')
@skip('Needs update to new setup')
def test_projector_process_input_invalid(self):
"""
Test setting input with an invalid value
# THEN: Input selected should reflect current input
assert (not self.pjlink.source), 'Input source should not have changed'
mock_log.warning.assert_has_calls(log_warning_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
TODO: Future test
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_inpt_not_in_list(self, mock_log):
"""
Test input source not listed in available sources
"""
# GIVEN: Test object
log_warning_calls = [call('({ip}) Input source not listed in available sources - '
'ignoring'.format(ip=self.pjlink.name))]
log_debug_calls = [call('({ip}) Processing command "INPT" with data "25"'.format(ip=self.pjlink.name)),
call('({ip}) Calling function for INPT'.format(ip=self.pjlink.name))]
self.pjlink.source = '11'
chk_source_available = ['11', '12', '21', '22', '31', '32']
self.pjlink.source_available = chk_source_available
@skip('Needs update to new setup')
def test_projector_process_inst_class_1(self):
# WHEN: Called with input source
process_command(projector=self.pjlink, cmd='INPT', data='25')
# THEN: Input selected should reflect current input
assert ('11' == self.pjlink.source), 'Input source should not have changed'
mock_log.warning.assert_has_calls(log_warning_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_inst_class_1(self, mock_log):
"""
Test saving video source available information
"""
# 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}) reset_information() connect status is '
'S_NOT_CONNECTED'.format(ip=pjlink.name)),
call('({ip}) Setting projector source_available to '
'"[\'11\', \'12\', \'21\', \'22\', \'31\', \'32\']"'.format(ip=pjlink.name))]
self.pjlink.source_available = []
log_warning_calls = []
log_debug_calls = [call('({ip}) Processing command "INST" with data '
'"21 12 11 22 32 31"'.format(ip=self.pjlink.name)),
call('({ip}) Calling function for INST'.format(ip=self.pjlink.name)),
call('({ip}) Setting projector source_available to '
'"[\'11\', \'12\', \'21\', \'22\', \'31\', \'32\']"'.format(ip=self.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']
chk_data = '21 12 11 22 32 31' # Although they should already be sorted, use unsorted to check method
chk_test = ['11', '12', '21', '22', '31', '32']
# WHEN: process_inst called with test data
pjlink.process_inst(data=chk_data)
# WHEN: process_inst called with test data
process_command(projector=self.pjlink, cmd='INST', data=chk_data)
# THEN: Data should have been sorted and saved properly
assert pjlink.source_available == chk_test, "Sources should have been sorted and saved"
mock_log.debug.assert_has_calls(log_debug_calls)
# THEN: Data should have been sorted and saved properly
mock_log.warning.assert_has_calls(log_warning_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
assert (self.pjlink.source_available == chk_test), "Sources should have been sorted and saved"
@skip('Needs update to new setup')
def test_projector_process_lamp_invalid(self):
"""
Test status multiple lamp on/off and hours
"""
# 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')
# THEN: lamps should not have changed
assert 2 == len(pjlink.lamp), 'Projector should have kept 2 lamps specified'
assert pjlink.lamp[0]['On'] is True, 'Lamp 1 power status should have stayed TRUE'
assert 00000 == pjlink.lamp[0]['Hours'], 'Lamp 1 hours should have been left at 00000'
assert pjlink.lamp[1]['On'] is False, 'Lamp 2 power status should have stayed FALSE'
assert 11111 == pjlink.lamp[1]['Hours'], 'Lamp 2 hours should have been left at 11111'
mock_log.warning.assert_has_calls(log_data)
@skip('Needs update to new setup')
def test_projector_process_lamp_multiple(self):
"""
Test status multiple lamp on/off and hours
"""
# GIVEN: Test object
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
pjlink.lamp = []
# WHEN: Call process_command with invalid lamp data
pjlink.process_lamp('11111 1 22222 0 33333 1')
# THEN: Lamp should have been set with proper lamp status
assert 3 == len(pjlink.lamp), 'Projector should have 3 lamps specified'
assert pjlink.lamp[0]['On'] is True, 'Lamp 1 power status should have been set to TRUE'
assert 11111 == pjlink.lamp[0]['Hours'], 'Lamp 1 hours should have been set to 11111'
assert pjlink.lamp[1]['On'] is False, 'Lamp 2 power status should have been set to FALSE'
assert 22222 == pjlink.lamp[1]['Hours'], 'Lamp 2 hours should have been set to 22222'
assert pjlink.lamp[2]['On'] is True, 'Lamp 3 power status should have been set to TRUE'
assert 33333 == pjlink.lamp[2]['Hours'], 'Lamp 3 hours should have been set to 33333'
@skip('Needs update to new setup')
def test_projector_process_lamp_single(self):
"""
Test status lamp on/off and hours
"""
# GIVEN: Test object
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
pjlink.lamp = []
# WHEN: Call process_command with invalid lamp data
pjlink.process_lamp('22222 1')
# THEN: Lamp should have been set with status=ON and hours=22222
assert 1 == len(pjlink.lamp), 'Projector should have only 1 lamp'
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'
@skip('Needs update to new setup')
def test_projector_process_lamp_single_hours_only(self):
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_lamp_invalid_missing_data(self, mock_log):
"""
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 = []
log_warning_calls = [call('({ip}) process_lamp(): Invalid data "45" - '
'Missing data'.format(ip=self.pjlink.name))]
log_debug_calls = [call('({ip}) Processing command "LAMP" with data "45"'.format(ip=self.pjlink.name)),
call('({ip}) Calling function for LAMP'.format(ip=self.pjlink.name))]
self.pjlink.lamp = None
# WHEN: Process lamp command called with only hours and no lamp power state
pjlink.process_lamp("45")
# WHEN: Call process_command with 3 lamps
process_command(projector=self.pjlink, cmd='LAMP', data='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"'
# THEN: Lamp should have been set with proper lamp status
mock_log.warning.assert_has_calls(log_warning_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
assert (not self.pjlink.lamp), 'Projector lamp info should not have changed'
@skip('Needs update to new setup')
def test_projector_process_name(self):
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_lamp_invalid_nan(self, mock_log):
"""
Test status multiple lamp on/off and hours
"""
# GIVEN: Test object
self.pjlink.lamp = [{'Hours': 00000, 'On': True},
{'Hours': 11111, 'On': False}]
log_warning_calls = [call('({ip}) process_lamp(): Invalid data "11111 1 22222 0 '
'333A3 1"'.format(ip=self.pjlink.name))]
log_debug_calls = [call('({ip}) Processing command "LAMP" with data "11111 1 22222 0 '
'333A3 1"'.format(ip=self.pjlink.name)),
call('({ip}) Calling function for LAMP'.format(ip=self.pjlink.name))]
# WHEN: Call process_command with invalid lamp data
process_command(projector=self.pjlink, cmd='LAMP', data='11111 1 22222 0 333A3 1')
# THEN: lamps should not have changed
mock_log.warning.assert_has_calls(log_warning_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
assert (2 == len(self.pjlink.lamp)), 'Projector lamp list should not have changed'
assert self.pjlink.lamp[0]['On'], 'Lamp 1 power status should not have changed'
assert (0 == self.pjlink.lamp[0]['Hours']), 'Lamp 1 hours should not have changed'
assert (not self.pjlink.lamp[1]['On']), 'Lamp 2 power status should not have changed'
assert (11111 == self.pjlink.lamp[1]['Hours']), 'Lamp 2 hours should not have changed'
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_lamp_multiple(self, mock_log):
"""
Test status multiple lamp on/off and hours
"""
# GIVEN: Test object
log_warning_calls = []
log_debug_calls = [call('({ip}) Processing command "LAMP" with data "11111 1 22222 0 '
'33333 1"'.format(ip=self.pjlink.name)),
call('({ip}) Calling function for LAMP'.format(ip=self.pjlink.name))]
self.pjlink.lamp = None
# WHEN: Call process_command with 3 lamps
process_command(projector=self.pjlink, cmd='LAMP', data='11111 1 22222 0 33333 1')
# THEN: Lamp should have been set with proper lamp status
mock_log.warning.assert_has_calls(log_warning_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
assert (3 == len(self.pjlink.lamp)), 'Projector should have 3 lamps specified'
assert self.pjlink.lamp[0]['On'], 'Lamp 1 power status should have been set to TRUE'
assert (11111 == self.pjlink.lamp[0]['Hours']), 'Lamp 1 hours should have been set to 11111'
assert (not self.pjlink.lamp[1]['On']), 'Lamp 2 power status should have been set to FALSE'
assert (22222 == self.pjlink.lamp[1]['Hours']), 'Lamp 2 hours should have been set to 22222'
assert self.pjlink.lamp[2]['On'], 'Lamp 3 power status should have been set to TRUE'
assert (33333 == self.pjlink.lamp[2]['Hours']), 'Lamp 3 hours should have been set to 33333'
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_lamp_single(self, mock_log):
"""
Test status lamp on/off and hours
"""
# GIVEN: Test object
log_warning_calls = []
log_debug_calls = [call('({ip}) Processing command "LAMP" with data "11111 1"'.format(ip=self.pjlink.name)),
call('({ip}) Calling function for LAMP'.format(ip=self.pjlink.name))]
self.pjlink.lamp = None
# WHEN: Call process_command with 3 lamps
process_command(projector=self.pjlink, cmd='LAMP', data='11111 1')
# THEN: Lamp should have been set with proper lamp status
mock_log.warning.assert_has_calls(log_warning_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
assert (1 == len(self.pjlink.lamp)), 'Projector should have 1 lamp specified'
assert self.pjlink.lamp[0]['On'], 'Lamp 1 power status should have been set to TRUE'
assert (11111 == self.pjlink.lamp[0]['Hours']), 'Lamp 1 hours should have been set to 11111'
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_name(self, mock_log):
"""
Test saving NAME data from 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))]
chk_data = "Some Name the End-User Set IN Projector"
log_warning_calls = []
log_debug_calls = [call('({ip}) Processing command "NAME" with data '
'"Some Name the End-User Set IN Projector"'.format(ip=self.pjlink.name)),
call('({ip}) Calling function for NAME'.format(ip=self.pjlink.name)),
call('({ip}) Setting projector PJLink name to '
'"Some Name the End-User Set IN Projector"'.format(ip=self.pjlink.name))]
# WHEN: process_name called with test data
pjlink.process_name(data=chk_data)
# WHEN: process_name called with test data
process_command(projector=self.pjlink, cmd='NAME', data=chk_data)
# THEN: name should be set and logged
assert pjlink.pjlink_name == chk_data, 'Name test data should have been saved'
mock_log.debug.assert_has_calls(log_debug_calls)
# THEN: name should be set and logged
mock_log.warning.assert_has_calls(log_warning_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
assert (self.pjlink.pjlink_name == chk_data), 'Name test data should have been saved'
@skip('Needs update to new setup')
def test_projector_process_powr_on(self):
"""
Test status power to ON
"""
# GIVEN: Test object
with patch.object(openlp.core.projectors.pjlink.PJLink, 'send_command') as mock_send_command, \
patch.object(openlp.core.projectors.pjlink.PJLink, 'change_status') as mock_change_status, \
patch.object(openlp.core.projectors.pjlink.PJLink, 'projectorUpdateIcons') as mock_UpdateIcons:
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
pjlink.power = S_STANDBY
# WHEN: process_name called with test data
pjlink.process_powr(data=PJLINK_POWR_STATUS[S_ON])
# THEN: Power should be set to ON
assert pjlink.power == S_ON, 'Power should have been set to ON'
assert mock_UpdateIcons.emit.called is True, 'projectorUpdateIcons should have been called'
mock_send_command.assert_called_once_with('INST')
mock_change_status.assert_called_once_with(S_ON)
@skip('Needs update to new setup')
def test_projector_process_powr_invalid(self):
@patch.object(openlp.core.projectors.pjlink.PJLink, 'send_command')
@patch.object(openlp.core.projectors.pjlink.PJLink, 'change_status')
@patch.object(openlp.core.projectors.pjlink.PJLink, 'projectorUpdateIcons')
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_powr_invalid(self, mock_log, mock_UpdateIcons, mock_change_status, mock_send_command):
"""
Test process_powr invalid call
"""
# 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, \
patch.object(openlp.core.projectors.pjlink.PJLink, 'change_status') as mock_change_status, \
patch.object(openlp.core.projectors.pjlink.PJLink, 'projectorUpdateIcons') as mock_UpdateIcons:
self.pjlink.power = S_STANDBY
log_warning_calls = [call('({ip}) Unknown power response: "99"'.format(ip=self.pjlink.name))]
log_debug_calls = [call('({ip}) Processing command "POWR" with data "99"'.format(ip=self.pjlink.name)),
call('({ip}) Calling function for POWR'.format(ip=self.pjlink.name)),
call('({ip}) Processing POWR command'.format(ip=self.pjlink.name))]
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_command called with test data
process_command(projector=self.pjlink, cmd='POWR', data='99')
# WHEN: process_name called with test data
pjlink.process_powr(data='99')
# THEN: Projector power should not have changed
mock_log.warning.assert_has_calls(log_warning_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
assert (S_STANDBY == self.pjlink.power), 'Power should not have changed'
mock_UpdateIcons.emit.assert_not_called()
mock_change_status.assert_not_called()
mock_send_command.assert_not_called()
# THEN: Power should be set to ON
assert pjlink.power == S_STANDBY, 'Power should not have changed'
mock_UpdateIcons.emit.assert_not_called()
mock_change_status.assert_not_called()
mock_send_command.assert_not_called()
mock_log.warning.assert_has_calls(log_warn_calls)
@skip('Needs update to new setup')
def test_projector_process_powr_off(self):
@patch.object(openlp.core.projectors.pjlink.PJLink, 'send_command')
@patch.object(openlp.core.projectors.pjlink.PJLink, 'change_status')
@patch.object(openlp.core.projectors.pjlink.PJLink, 'projectorUpdateIcons')
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_powr_off(self, mock_log, mock_UpdateIcons, mock_change_status, mock_send_command):
"""
Test status power to STANDBY
Test status power to OFF
"""
# GIVEN: Test object
with patch.object(openlp.core.projectors.pjlink.PJLink, 'send_command') as mock_send_command, \
patch.object(openlp.core.projectors.pjlink.PJLink, 'change_status') as mock_change_status, \
patch.object(openlp.core.projectors.pjlink.PJLink, 'projectorUpdateIcons') as mock_UpdateIcons:
log_warning_calls = []
log_debug_calls = [call('({ip}) Processing command "POWR" with data "0"'.format(ip=self.pjlink.name)),
call('({ip}) Calling function for POWR'.format(ip=self.pjlink.name)),
call('({ip}) Processing POWR command'.format(ip=self.pjlink.name))]
self.pjlink.power = S_ON
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
pjlink.power = S_ON
# WHEN: process_name called with test data
process_command(projector=self.pjlink, cmd='POWR', data=PJLINK_POWR_STATUS[S_STANDBY])
# WHEN: process_name called with test data
pjlink.process_powr(data=PJLINK_POWR_STATUS[S_STANDBY])
# THEN: Power should be set to ON
mock_log.warning.assert_has_calls(log_warning_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
assert (S_STANDBY == self.pjlink.power), 'Power should have been set to OFF'
assert mock_UpdateIcons.emit.called, 'projectorUpdateIcons should have been called'
assert (not mock_send_command.called), 'send_command should not have been called'
mock_change_status.assert_called_once_with(S_STANDBY)
# THEN: Power should be set to ON
assert pjlink.power == S_STANDBY, 'Power should have changed to S_STANDBY'
mock_UpdateIcons.emit.assert_called_with()
mock_change_status.assert_called_with(313)
mock_send_command.assert_not_called()
@patch.object(openlp.core.projectors.pjlink.PJLink, 'send_command')
@patch.object(openlp.core.projectors.pjlink.PJLink, 'change_status')
@patch.object(openlp.core.projectors.pjlink.PJLink, 'projectorUpdateIcons')
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_powr_on(self, mock_log, mock_UpdateIcons, mock_change_status, mock_send_command):
"""
Test status power to ON
"""
# GIVEN: Test object
log_warning_calls = []
log_debug_calls = [call('({ip}) Processing command "POWR" with data "1"'.format(ip=self.pjlink.name)),
call('({ip}) Calling function for POWR'.format(ip=self.pjlink.name)),
call('({ip}) Processing POWR command'.format(ip=self.pjlink.name))]
self.pjlink.power = S_STANDBY
@skip('Needs update to new setup')
def test_projector_process_rfil_save(self):
# WHEN: process_name called with test data
process_command(projector=self.pjlink, cmd='POWR', data=PJLINK_POWR_STATUS[S_ON])
# THEN: Power should be set to ON
mock_log.warning.assert_has_calls(log_warning_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
assert (S_ON == self.pjlink.power), 'Power should have been set to ON'
assert mock_UpdateIcons.emit.called, 'projectorUpdateIcons should have been called'
mock_send_command.assert_called_once_with('INST')
mock_change_status.assert_called_once_with(S_ON)
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_rfil_save(self, mock_log):
"""
Test saving filter type
"""
filter_model = 'Filter Type Test'
# GIVEN: Test object
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
pjlink.model_filter = None
new_data = 'Filter Type Test'
log_warning_calls = []
log_debug_calls = [call('({ip}) Processing command "RFIL" with data '
'"{data}"'.format(ip=self.pjlink.name, data=new_data)),
call('({ip}) Calling function for RFIL'.format(ip=self.pjlink.name))]
self.pjlink.model_filter = None
# WHEN: Filter model is received
pjlink.process_rfil(data=filter_model)
process_command(projector=self.pjlink, cmd='RFIL', data=new_data)
# THEN: Filter model number should be saved
assert pjlink.model_filter == filter_model, 'Filter type should have been saved'
mock_log.warning.assert_has_calls(log_warning_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
assert (self.pjlink.model_filter == new_data), 'Filter model should have been saved'
@skip('Needs update to new setup')
def test_projector_process_rfil_nosave(self):
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_rfil_nosave(self, mock_log):
"""
Test saving filter type previously saved
"""
# GIVEN: Test object
with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log:
old_data = 'Old filter type'
new_data = 'Filter Type Test'
log_warning_calls = [call('({ip}) Filter model already set'.format(ip=self.pjlink.name)),
call('({ip}) Saved model: "{data}"'.format(ip=self.pjlink.name, data=old_data)),
call('({ip}) New model: "{data}"'.format(ip=self.pjlink.name, data=new_data))]
log_debug_calls = [call('({ip}) Processing command "RFIL" with data '
'"{data}"'.format(ip=self.pjlink.name, data=new_data)),
call('({ip}) Calling function for RFIL'.format(ip=self.pjlink.name))]
self.pjlink.model_filter = old_data
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
process_command(projector=self.pjlink, cmd='RFIL', data=new_data)
# WHEN: Filter model is received
pjlink.process_rfil(data=filter_model)
# THEN: Filter model number should be saved
assert (self.pjlink.model_filter != new_data), 'Filter model should NOT have been saved'
mock_log.warning.assert_has_calls(log_warning_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
# THEN: Filter model number should be saved
assert pjlink.model_filter != filter_model, 'Filter type should NOT have been saved'
mock_log.warning.assert_has_calls(log_warn_calls)
@skip('Needs update to new setup')
def test_projector_process_rlmp_save(self):
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_rlmp_save(self, mock_log):
"""
Test saving lamp type
"""
# GIVEN: Test object
# GIVEN: Test object
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
pjlink.model_lamp = None
lamp_model = 'Lamp Type Test'
new_data = 'Lamp Type Test'
log_warning_calls = []
log_debug_calls = [call('({ip}) Processing command "RLMP" with data '
'"{data}"'.format(ip=self.pjlink.name, data=new_data)),
call('({ip}) Calling function for RLMP'.format(ip=self.pjlink.name))]
self.pjlink.model_lamp = None
# WHEN: Filter model is received
pjlink.process_rlmp(data=lamp_model)
process_command(projector=self.pjlink, cmd='RLMP', data=new_data)
# THEN: Filter model number should be saved
assert pjlink.model_lamp == lamp_model, 'Lamp type should have been saved'
mock_log.warning.assert_has_calls(log_warning_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
assert (self.pjlink.model_lamp == new_data), 'Lamp model should have been saved'
@skip('Needs update to new setup')
def test_projector_process_rlmp_nosave(self):
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_rlmp_nosave(self, mock_log):
"""
Test saving lamp type previously saved
"""
# GIVEN: Test object
with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log:
old_data = 'Old filter type'
new_data = 'Filter Type Test'
log_warning_calls = [call('({ip}) Lamp model already set'.format(ip=self.pjlink.name)),
call('({ip}) Saved lamp: "{data}"'.format(ip=self.pjlink.name, data=old_data)),
call('({ip}) New lamp: "{data}"'.format(ip=self.pjlink.name, data=new_data))]
log_debug_calls = [call('({ip}) Processing command "RLMP" with data '
'"{data}"'.format(ip=self.pjlink.name, data=new_data)),
call('({ip}) Calling function for RLMP'.format(ip=self.pjlink.name))]
self.pjlink.model_lamp = old_data
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
process_command(projector=self.pjlink, cmd='RLMP', data=new_data)
# WHEN: Filter model is received
pjlink.process_rlmp(data=lamp_model)
# THEN: Filter model number should be saved
assert (self.pjlink.model_lamp != new_data), 'Lamp model should NOT have been saved'
mock_log.warning.assert_has_calls(log_warning_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
# THEN: Filter model number should be saved
assert pjlink.model_lamp != lamp_model, 'Lamp type should NOT have been saved'
mock_log.warning.assert_has_calls(log_warn_calls)
@skip('Needs update to new setup')
def test_projector_process_snum_set(self):
"""
Test saving serial number from projector
"""
# 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)
# THEN: Serial number should be set
assert pjlink.serial_no == test_number, 'Projector serial number should have been set'
mock_log.debug.assert_has_calls(log_debug_calls)
@skip('Needs update to new setup')
def test_projector_process_snum_different(self):
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_snum_different(self, mock_log):
"""
Test projector serial number different than saved 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'
new_data = 'Test Serial Number'
old_data = 'Previous serial number'
log_warning_calls = [call('({ip}) Projector serial number does not match '
'saved serial number'.format(ip=self.pjlink.name)),
call('({ip}) Saved: "{data}"'.format(ip=self.pjlink.name, data=old_data)),
call('({ip}) Received: "{data}"'.format(ip=self.pjlink.name, data=new_data)),
call('({ip}) NOT saving serial number'.format(ip=self.pjlink.name))]
# WHEN: No serial number is set and we receive serial number command
pjlink.process_snum(data=test_number)
log_debug_calls = [call('({ip}) Processing command "SNUM" with data '
'"{data}"'.format(ip=self.pjlink.name, data=new_data)),
call('({ip}) Calling function for SNUM'.format(ip=self.pjlink.name))]
self.pjlink.serial_no = old_data
# THEN: Serial number should be set
assert pjlink.serial_no != test_number, 'Projector serial number should NOT have been set'
mock_log.warning.assert_has_calls(log_warn_calls)
# WHEN: No serial number is set and we receive serial number command
process_command(projector=self.pjlink, cmd='SNUM', data=new_data)
@skip('Needs update to new setup')
def test_projector_process_sver(self):
# THEN: Serial number should be set
assert (self.pjlink.serial_no != new_data), 'Projector serial number should NOT have been set'
mock_log.warning.assert_has_calls(log_warning_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_snum_set(self, mock_log):
"""
Test invalid software version information - too long
Test saving serial number from projector
"""
# 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))]
new_data = 'Test Serial Number'
self.pjlink.serial_no = None
log_warning_calls = []
log_debug_calls = [call('({ip}) Processing command "SNUM" with data "{data}"'.format(ip=self.pjlink.name,
data=new_data)),
call('({ip}) Calling function for SNUM'.format(ip=self.pjlink.name)),
call('({ip}) Setting projector serial number to '
'"{data}"'.format(ip=self.pjlink.name, data=new_data))]
# WHEN: process_sver called with invalid data
pjlink.process_sver(data=test_data)
# WHEN: No serial number is set and we receive serial number command
process_command(projector=self.pjlink, cmd='SNUM', data=new_data)
# THEN: Version information should not change
assert pjlink.sw_version == test_data, 'Software version should have been updated'
mock_log.debug.assert_has_calls(log_debug_calls)
# THEN: Serial number should be set
assert (self.pjlink.serial_no == new_data), 'Projector serial number should have been set'
mock_log.warning.assert_has_calls(log_warning_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
@skip('Needs update to new setup')
def test_projector_process_sver_changed(self):
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_sver_changed(self, mock_log):
"""
Test invalid software version information - Received different than saved
"""
# 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
old_data = 'Test 1 Subtest 1'
new_data = 'Test 1 Subtest 2'
log_warning_calls = [call('({ip}) Projector software version does not match '
'saved software version'.format(ip=self.pjlink.name)),
call('({ip}) Saved: "{data}"'.format(ip=self.pjlink.name, data=old_data)),
call('({ip}) Received: "{data}"'.format(ip=self.pjlink.name, data=new_data)),
call('({ip}) Updating software version'.format(ip=self.pjlink.name))]
log_debug_calls = [call('({ip}) Processing command "SVER" with data '
'"{data}"'.format(ip=self.pjlink.name, data=new_data)),
call('({ip}) Calling function for SVER'.format(ip=self.pjlink.name)),
call('({ip}) Setting projector software version to '
'"{data}"'.format(ip=self.pjlink.name, data=new_data))]
self.pjlink.sw_version = old_data
# WHEN: process_sver called with invalid data
pjlink.process_sver(data=test_data_new)
# WHEN: process_sver called with invalid data
process_command(self.pjlink, cmd='SVER', data=new_data)
# THEN: Version information should not change
assert pjlink.sw_version == test_data_new, 'Software version should have changed'
mock_log.warning.assert_has_calls(log_warn_calls)
# THEN: Version information should change
assert (self.pjlink.sw_version == new_data), 'Software version should have changed'
mock_log.warning.assert_has_calls(log_warning_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
@skip('Needs update to new setup')
def test_projector_process_sver_invalid(self):
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_sver_invalid(self, mock_log):
"""
Test invalid software version information - too long
"""
test_data = 'This is a test software version line that is too long based on PJLink version 2 specs'
log_warn_calls = [call('Invalid software version - too long')]
# 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
new_data = 'This is a test software version line that is too long based on PJLink version 2 specs'
log_warning_calls = [call('Invalid software version - too long')]
log_debug_calls = [call('({ip}) Processing command "SVER" with data "{data}"'.format(ip=self.pjlink.name,
data=new_data)),
call('({ip}) Calling function for SVER'.format(ip=self.pjlink.name))]
self.pjlink.sw_version = None
self.pjlink.sw_version_received = None
# WHEN: process_sver called with invalid data
pjlink.process_sver(data=test_data)
# WHEN: process_sver called with invalid data
process_command(projector=self.pjlink, cmd='SVER', data=new_data)
# THEN: Version information should not change
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)
# THEN: Version information should not change
assert (not self.pjlink.sw_version), 'Software version should not have changed'
assert (not self.pjlink.sw_version_received), 'Received software version should not have changed'
mock_log.warning.assert_has_calls(log_warning_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_sver_save(self, mock_log):
"""
Test invalid software version information - too long
"""
# GIVEN: Test object
new_data = 'Test 1 Subtest 1'
log_warning_calls = []
log_debug_calls = [call('({ip}) Processing command "SVER" with data '
'"{data}"'.format(ip=self.pjlink.name, data=new_data)),
call('({ip}) Calling function for SVER'.format(ip=self.pjlink.name)),
call('({ip}) Setting projector software version to '
'"{data}"'.format(ip=self.pjlink.name, data=new_data))]
self.pjlink.sw_version = None
self.pjlink.sw_version_received = None
# WHEN: process_sver called with invalid data
process_command(projector=self.pjlink, cmd='SVER', data=new_data)
# THEN: Version information should not change
assert (self.pjlink.sw_version == new_data), 'Software version should have been updated'
assert (not self.pjlink.sw_version_received), 'Received version field should not have changed'
mock_log.warning.assert_has_calls(log_warning_calls)
mock_log.debug.assert_has_calls(log_debug_calls)

0
tests/openlp_core/projectors/test_projector_pjlink_commands_02.py → tests/openlp_core/projectors/test_projector_commands_03.py
View File

31
tests/openlp_core/projectors/test_projector_pjlink_base_02.py
View File

@ -132,37 +132,6 @@ class TestPJLinkBase(TestCase):
assert (not mock_timer.called), 'Timer should not have been called'
assert (not mock_reset.called), 'reset_information() should not have been called'
@patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
@patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
@patch.object(openlp.core.projectors.pjlink, 'log')
def test_local_send_command_no_data_queue_check(self, mock_log, mock_reset, mock_state):
"""
Test _underscore_send_command last queue length check
"""
# GIVEN: Test object
log_error_calls = []
log_warning_calls = [call('({ip}) _send_command(): No data to send'.format(ip=self.pjlink.name))]
log_debug_calls = []
mock_state.return_value = QSOCKET_STATE[S_CONNECTED]
self.pjlink.priority_queue = []
# WHEN: _send_command called with no data and queue's emtpy
# Patch some attributes here since they are not available until after instantiation
with patch.object(self.pjlink, 'socket_timer') as mock_timer, \
patch.object(self.pjlink, 'send_queue') as mock_queue:
# Unlikely case of send_queue not really empty, but len(send_queue) returns 0
mock_queue.return_value = ['test']
mock_queue.__len__.return_value = 0
self.pjlink._send_command(data=None)
# THEN:
mock_log.error.assert_has_calls(log_error_calls)
mock_log.warning.assert_has_calls(log_warning_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
assert (not self.pjlink.priority_queue), 'Priority queue should be empty'
assert (not mock_timer.called), 'Timer should not have been called'
assert (not mock_reset.called), 'reset_information() should not have been called'
@patch.object(openlp.core.projectors.pjlink.PJLink, 'write')
@patch.object(openlp.core.projectors.pjlink.PJLink, 'disconnect_from_host')
@patch.object(openlp.core.projectors.pjlink.PJLink, 'state')

20
tests/openlp_core/projectors/test_projector_pjlink_cmd_routing.py
View File

@ -56,9 +56,9 @@ class TestPJLinkRouting(TestCase):
Test projector received valid command invalid version
"""
# GIVEN: Test object
log_warning_text = [call('({ip}) get_data() Command reply version does not match '
'a valid command version'.format(ip=self.pjlink.name)),
call('({ip}) _send_command(): Nothing to send - returning'.format(ip=self.pjlink.name))]
log_warning_text = [call('({ip}) _send_command(): Nothing to send - returning'.format(ip=self.pjlink.name)),
call('({ip}) get_data() Command reply version does not match '
'a valid command version'.format(ip=self.pjlink.name))]
log_debug_text = [call('({ip}) get_data(buffer="{pre}XCLSS=X"'.format(ip=self.pjlink.name, pre=PJLINK_PREFIX)),
call('({ip}) get_data(): Checking new data "{pre}XCLSS=X"'.format(ip=self.pjlink.name,
pre=PJLINK_PREFIX)),
@ -83,10 +83,10 @@ class TestPJLinkRouting(TestCase):
Test projector receiving invalid command
"""
# GIVEN: Test object
log_warning_text = [call('({ip}) get_data(): Invalid packet - '
'unknown command "UNKN"'.format(ip=self.pjlink.name)),
call('({ip}) _send_command(): Nothing to send - '
'returning'.format(ip=self.pjlink.name))]
log_warning_text = [call('({ip}) _send_command(): Nothing to send - '
'returning'.format(ip=self.pjlink.name)),
call('({ip}) get_data(): Invalid packet - '
'unknown command "UNKN"'.format(ip=self.pjlink.name))]
log_debug_text = [call('({ip}) get_data(buffer="{pre}1UNKN=Huh?"'.format(ip=self.pjlink.name,
pre=PJLINK_PREFIX)),
call('({ip}) get_data(): Checking new data "{pre}1UNKN=Huh?"'.format(ip=self.pjlink.name,
@ -113,9 +113,9 @@ class TestPJLinkRouting(TestCase):
Test projector received valid command with command version higher than projector
"""
# GIVEN: Test object
log_warning_text = [call('({ip}) get_data(): Projector returned class reply higher than projector '
'stated class'.format(ip=self.pjlink.name)),
call('({ip}) _send_command(): Nothing to send - returning'.format(ip=self.pjlink.name))]
log_warning_text = [call('({ip}) _send_command(): Nothing to send - returning'.format(ip=self.pjlink.name)),
call('({ip}) get_data(): Projector returned class reply higher than projector '
'stated class'.format(ip=self.pjlink.name))]
log_debug_text = [call('({ip}) get_data(buffer="{pre}2ACKN=X"'.format(ip=self.pjlink.name,
pre=PJLINK_PREFIX)),