forked from openlp/openlp
PJLink 2 Update v07
bzr-revno: 2871
This commit is contained in:
commit
fa7e8f9d3c
@ -39,6 +39,7 @@ PJLINK_PREFIX = '%'
|
||||
PJLINK_PORT = 4352
|
||||
PJLINK_SUFFIX = CR
|
||||
PJLINK_TIMEOUT = 30.0
|
||||
PJLINK_TOKEN_SIZE = 8 # PJLINK 1 <token> : where <token> is 8 characters
|
||||
|
||||
# Error and status codes
|
||||
S_OK = E_OK = 0 # E_OK included since I sometimes forget
|
||||
|
@ -281,6 +281,7 @@ class PJLink(QtNetwork.QTcpSocket):
|
||||
self.pjlink_class = copy(PJLINK_CLASS)
|
||||
self.pjlink_name = None # NAME
|
||||
self.power = S_OFF # POWR
|
||||
self.projector_errors = {} # Full ERST errors
|
||||
self.serial_no = None # SNUM
|
||||
self.serial_no_received = None
|
||||
self.sw_version = None # SVER
|
||||
|
@ -29,13 +29,13 @@ NOTE: PJLink Class (version) checks are handled in the respective PJLink/PJLinkU
|
||||
|
||||
import logging
|
||||
import re
|
||||
import string
|
||||
|
||||
from openlp.core.common.i18n import translate
|
||||
from openlp.core.common.settings import Settings
|
||||
|
||||
from openlp.core.projectors.constants import E_AUTHENTICATION, PJLINK_DEFAULT_CODES, PJLINK_ERRORS, \
|
||||
PJLINK_ERST_DATA, PJLINK_ERST_STATUS, PJLINK_POWR_STATUS, S_AUTHENTICATE, S_CONNECT, S_DATA_OK, S_OFF, S_OK, S_ON, \
|
||||
S_STANDBY, STATUS_MSG
|
||||
PJLINK_ERST_DATA, PJLINK_ERST_LIST, PJLINK_ERST_STATUS, PJLINK_POWR_STATUS, PJLINK_TOKEN_SIZE, \
|
||||
E_NO_AUTHENTICATION, S_AUTHENTICATE, S_CONNECT, S_DATA_OK, S_OFF, S_OK, S_ON, S_STANDBY, STATUS_MSG
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
log.debug('Loading pjlinkcommands')
|
||||
@ -195,8 +195,7 @@ def process_erst(projector, data):
|
||||
# Bad data - ignore
|
||||
log.warning('({ip}) Invalid error status response "{data}"'.format(ip=projector.entry.name, data=data))
|
||||
return
|
||||
datacheck = int(data)
|
||||
if datacheck == 0:
|
||||
if int(data) == 0:
|
||||
projector.projector_errors = None
|
||||
# No errors
|
||||
return
|
||||
@ -209,23 +208,17 @@ def process_erst(projector, data):
|
||||
data[PJLINK_ERST_DATA['FILTER']],
|
||||
data[PJLINK_ERST_DATA['OTHER']])
|
||||
if fan != PJLINK_ERST_STATUS[S_OK]:
|
||||
projector.projector_errors[translate('OpenLP.ProjectorPJLink', 'Fan')] = \
|
||||
PJLINK_ERST_STATUS[fan]
|
||||
projector.projector_errors[PJLINK_ERST_LIST['FAN']] = PJLINK_ERST_STATUS[fan]
|
||||
if lamp != PJLINK_ERST_STATUS[S_OK]:
|
||||
projector.projector_errors[translate('OpenLP.ProjectorPJLink', 'Lamp')] = \
|
||||
PJLINK_ERST_STATUS[lamp]
|
||||
projector.projector_errors[PJLINK_ERST_LIST['LAMP']] = PJLINK_ERST_STATUS[lamp]
|
||||
if temp != PJLINK_ERST_STATUS[S_OK]:
|
||||
projector.projector_errors[translate('OpenLP.ProjectorPJLink', 'Temperature')] = \
|
||||
PJLINK_ERST_STATUS[temp]
|
||||
projector.projector_errors[PJLINK_ERST_LIST['TEMP']] = PJLINK_ERST_STATUS[temp]
|
||||
if cover != PJLINK_ERST_STATUS[S_OK]:
|
||||
projector.projector_errors[translate('OpenLP.ProjectorPJLink', 'Cover')] = \
|
||||
PJLINK_ERST_STATUS[cover]
|
||||
projector.projector_errors[PJLINK_ERST_LIST['COVER']] = PJLINK_ERST_STATUS[cover]
|
||||
if filt != PJLINK_ERST_STATUS[S_OK]:
|
||||
projector.projector_errors[translate('OpenLP.ProjectorPJLink', 'Filter')] = \
|
||||
PJLINK_ERST_STATUS[filt]
|
||||
projector.projector_errors[PJLINK_ERST_LIST['FILTER']] = PJLINK_ERST_STATUS[filt]
|
||||
if other != PJLINK_ERST_STATUS[S_OK]:
|
||||
projector.projector_errors[translate('OpenLP.ProjectorPJLink', 'Other')] = \
|
||||
PJLINK_ERST_STATUS[other]
|
||||
projector.projector_errors[PJLINK_ERST_LIST['OTHER']] = PJLINK_ERST_STATUS[other]
|
||||
return
|
||||
|
||||
|
||||
@ -389,20 +382,29 @@ def process_pjlink(projector, data):
|
||||
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=projector.entry.name))
|
||||
return E_AUTHENTICATION
|
||||
return E_NO_AUTHENTICATION
|
||||
elif projector.pin:
|
||||
log.error('({ip}) Normal connection but PIN set - aborting'.format(ip=projector.entry.name))
|
||||
return E_AUTHENTICATION
|
||||
return E_NO_AUTHENTICATION
|
||||
log.debug('({ip}) PJLINK: Returning S_CONNECT'.format(ip=projector.entry.name))
|
||||
return S_CONNECT
|
||||
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=projector.entry.name))
|
||||
return E_AUTHENTICATION
|
||||
return E_NO_AUTHENTICATION
|
||||
elif len(chk[-1]) != PJLINK_TOKEN_SIZE:
|
||||
# Bad token - incorrect size
|
||||
log.error('({ip}) Authentication token invalid (size) - aborting'.format(ip=projector.entry.name))
|
||||
return E_NO_AUTHENTICATION
|
||||
elif not all(c in string.hexdigits for c in chk[-1]):
|
||||
# Bad token - not hexadecimal
|
||||
log.error('({ip}) Authentication token invalid (not a hexadecimal number) '
|
||||
'- aborting'.format(ip=projector.entry.name))
|
||||
return E_NO_AUTHENTICATION
|
||||
elif not projector.pin:
|
||||
log.error('({ip}) Authenticate connection but no PIN - aborting'.format(ip=projector.entry.name))
|
||||
return E_AUTHENTICATION
|
||||
return E_NO_AUTHENTICATION
|
||||
log.debug('({ip}) PJLINK: Returning S_AUTHENTICATE'.format(ip=projector.entry.name))
|
||||
return S_AUTHENTICATE
|
||||
|
||||
|
@ -35,7 +35,7 @@ from tests.resources.projector.data import TEST1_DATA
|
||||
|
||||
class TestPJLinkCommands(TestCase):
|
||||
"""
|
||||
Tests PJLink get status commands part 1
|
||||
Tests PJLink commands part 1
|
||||
"""
|
||||
def setUp(self):
|
||||
"""
|
||||
@ -92,7 +92,7 @@ class TestPJLinkCommands(TestCase):
|
||||
# THEN: Shutter should be closed and mute should be True
|
||||
assert self.pjlink.shutter, 'Shutter should changed'
|
||||
assert self.pjlink.mute, 'Audio should not have changed'
|
||||
assert (not mock_UpdateIcons.emit.called), 'Update icons should NOT have been called'
|
||||
assert not mock_UpdateIcons.emit.called, 'Update icons should NOT have been called'
|
||||
mock_log.warning.assert_has_calls(log_warning_text)
|
||||
mock_log.debug.assert_has_calls(log_debug_text)
|
||||
|
||||
@ -140,8 +140,8 @@ class TestPJLinkCommands(TestCase):
|
||||
process_command(projector=self.pjlink, cmd='AVMT', data='30')
|
||||
|
||||
# THEN: Shutter should be closed and mute should be True
|
||||
assert (not self.pjlink.shutter), 'Shutter should have been set to off'
|
||||
assert (not self.pjlink.mute), 'Audio should be on'
|
||||
assert not self.pjlink.shutter, 'Shutter should have been set to off'
|
||||
assert not self.pjlink.mute, 'Audio should be on'
|
||||
assert mock_UpdateIcons.emit.called, 'Update icons should have been called'
|
||||
mock_log.warning.assert_has_calls(log_warning_text)
|
||||
mock_log.debug.assert_has_calls(log_debug_text)
|
||||
@ -193,7 +193,7 @@ class TestPJLinkCommands(TestCase):
|
||||
assert self.pjlink.shutter, 'Shutter should have been set to closed'
|
||||
assert self.pjlink.mute, 'Audio should not have changed'
|
||||
assert mock_UpdateIcons.emit.called, 'Update icons should have been called'
|
||||
assert ('AVMT' not in self.pjlink.status_timer_checks), 'Status timer list should not have AVMT callback'
|
||||
assert 'AVMT' not in self.pjlink.status_timer_checks, 'Status timer list should not have AVMT callback'
|
||||
assert mock_status_timer.stop.called, 'Projector status_timer.stop() should have been called'
|
||||
mock_log.warning.assert_has_calls(log_warning_text)
|
||||
mock_log.debug.assert_has_calls(log_debug_text)
|
||||
@ -214,7 +214,7 @@ class TestPJLinkCommands(TestCase):
|
||||
process_command(projector=self.pjlink, cmd='CLSS', data='1')
|
||||
|
||||
# THEN: Projector class should be set with proper value
|
||||
assert ('1' == self.pjlink.pjlink_class), 'Should have set class=1'
|
||||
assert '1' == self.pjlink.pjlink_class, '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)
|
||||
@ -235,7 +235,7 @@ class TestPJLinkCommands(TestCase):
|
||||
process_command(projector=self.pjlink, cmd='CLSS', data='2')
|
||||
|
||||
# THEN: Projector class should be set with proper value
|
||||
assert ('2' == self.pjlink.pjlink_class), 'Should have set class=2'
|
||||
assert '2' == self.pjlink.pjlink_class, 'Should have set class=2'
|
||||
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)
|
||||
@ -256,7 +256,7 @@ class TestPJLinkCommands(TestCase):
|
||||
process_command(projector=self.pjlink, cmd='CLSS', data='Z')
|
||||
|
||||
# THEN: Projector class should be set with default value
|
||||
assert (self.pjlink.pjlink_class == '1'), 'Invalid NaN class reply should have set class=1'
|
||||
assert self.pjlink.pjlink_class == '1', 'Invalid NaN class reply should have set class=1'
|
||||
mock_log.warning.assert_has_calls(log_warning_calls)
|
||||
mock_log.debug.assert_has_calls(log_debug_calls)
|
||||
|
||||
@ -276,7 +276,7 @@ class TestPJLinkCommands(TestCase):
|
||||
process_command(projector=self.pjlink, cmd='CLSS', data='Invalid')
|
||||
|
||||
# THEN: Projector class should be set with default value
|
||||
assert (self.pjlink.pjlink_class == '1'), 'Invalid class reply should have set class=1'
|
||||
assert self.pjlink.pjlink_class == '1', 'Invalid class reply should have set class=1'
|
||||
mock_log.warning.assert_has_calls(log_warning_calls)
|
||||
mock_log.debug.assert_has_calls(log_debug_calls)
|
||||
|
||||
@ -296,7 +296,7 @@ class TestPJLinkCommands(TestCase):
|
||||
process_command(projector=self.pjlink, cmd='CLSS', data='Class 1')
|
||||
|
||||
# THEN: Projector class should be set with proper value
|
||||
assert ('1' == self.pjlink.pjlink_class), 'Non-standard class reply should have set class=1'
|
||||
assert '1' == self.pjlink.pjlink_class, 'Non-standard 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)
|
||||
@ -316,7 +316,7 @@ class TestPJLinkCommands(TestCase):
|
||||
process_command(projector=self.pjlink, cmd='CLSS', data='Version2')
|
||||
|
||||
# THEN: Projector class should be set with proper value
|
||||
assert ('2' == self.pjlink.pjlink_class), 'Non-standard class reply should have set class=1'
|
||||
assert '2' == self.pjlink.pjlink_class, 'Non-standard class reply should have set class=1'
|
||||
mock_log.warning.assert_has_calls(log_warning_calls)
|
||||
mock_log.debug.assert_has_calls(log_debug_calls)
|
||||
|
||||
@ -348,7 +348,7 @@ class TestPJLinkCommands(TestCase):
|
||||
process_command(projector=self.pjlink, cmd='ERST', data=chk_data)
|
||||
|
||||
# THEN: PJLink instance errors should match chk_value
|
||||
assert (self.pjlink.projector_errors == chk_test), 'Projector errors should be all E_ERROR'
|
||||
assert self.pjlink.projector_errors == chk_test, 'Projector errors should be all E_ERROR'
|
||||
mock_log.warning.assert_has_calls(log_warning_calls)
|
||||
mock_log.debug.assert_has_calls(log_debug_calls)
|
||||
|
||||
@ -368,7 +368,7 @@ class TestPJLinkCommands(TestCase):
|
||||
process_command(projector=self.pjlink, cmd='ERST', data=chk_data)
|
||||
|
||||
# THEN: PJLink instance errors should be None
|
||||
assert (self.pjlink.projector_errors is None), 'projector_errors should have been set to None'
|
||||
assert self.pjlink.projector_errors is None, 'projector_errors should have been set to None'
|
||||
mock_log.warning.assert_has_calls(log_warning_calls)
|
||||
mock_log.debug.assert_has_calls(log_debug_calls)
|
||||
|
||||
@ -400,7 +400,7 @@ class TestPJLinkCommands(TestCase):
|
||||
process_command(projector=self.pjlink, cmd='ERST', data=chk_data)
|
||||
|
||||
# THEN: PJLink instance errors should match chk_value
|
||||
assert (self.pjlink.projector_errors == chk_test), 'Projector errors should be all E_WARN'
|
||||
assert self.pjlink.projector_errors == chk_test, 'Projector errors should be all E_WARN'
|
||||
mock_log.warning.assert_has_calls(log_warning_calls)
|
||||
mock_log.debug.assert_has_calls(log_debug_calls)
|
||||
|
||||
@ -423,7 +423,7 @@ class TestPJLinkCommands(TestCase):
|
||||
process_command(self.pjlink, cmd='ERST', data=chk_data)
|
||||
|
||||
# THEN: pjlink.projector_errors should be empty and warning logged
|
||||
assert (not self.pjlink.projector_errors), 'There should be no errors'
|
||||
assert not self.pjlink.projector_errors, 'There should be no errors'
|
||||
mock_log.warning.assert_has_calls(log_warn_calls)
|
||||
mock_log.debug.assert_has_calls(log_debug_calls)
|
||||
|
||||
@ -445,7 +445,7 @@ class TestPJLinkCommands(TestCase):
|
||||
process_command(self.pjlink, cmd='ERST', data=chk_data)
|
||||
|
||||
# THEN: pjlink.projector_errors should be empty and warning logged
|
||||
assert (not self.pjlink.projector_errors), 'There should be no errors'
|
||||
assert not self.pjlink.projector_errors, 'There should be no errors'
|
||||
mock_log.warning.assert_has_calls(log_warn_calls)
|
||||
mock_log.debug.assert_has_calls(log_debug_calls)
|
||||
|
||||
@ -472,10 +472,10 @@ class TestPJLinkCommands(TestCase):
|
||||
process_command(projector=self.pjlink, cmd='ERST', data=chk_data)
|
||||
|
||||
# THEN: PJLink instance errors should match only cover warning
|
||||
assert (1 == len(self.pjlink.projector_errors)), 'There should only be 1 error listed in projector_errors'
|
||||
assert ('Cover' in self.pjlink.projector_errors), '"Cover" should be the only error listed'
|
||||
assert (self.pjlink.projector_errors['Cover'] == E_WARN), '"Cover" should have E_WARN listed as error'
|
||||
assert (chk_test == self.pjlink.projector_errors), 'projector_errors should match test errors'
|
||||
assert 1 == len(self.pjlink.projector_errors), 'There should only be 1 error listed in projector_errors'
|
||||
assert 'Cover' in self.pjlink.projector_errors, '"Cover" should be the only error listed'
|
||||
assert self.pjlink.projector_errors['Cover'] == E_WARN, '"Cover" should have E_WARN listed as error'
|
||||
assert chk_test == self.pjlink.projector_errors, 'projector_errors should match test errors'
|
||||
mock_log.warning.assert_has_calls(log_warn_calls)
|
||||
mock_log.debug.assert_has_calls(log_debug_calls)
|
||||
|
||||
@ -498,7 +498,7 @@ class TestPJLinkCommands(TestCase):
|
||||
process_command(projector=self.pjlink, cmd='INF1', data=chk_data)
|
||||
|
||||
# THEN: Data should be saved
|
||||
assert (self.pjlink.manufacturer == chk_data), 'Test data should have been saved'
|
||||
assert self.pjlink.manufacturer == chk_data, 'Test data should have been saved'
|
||||
mock_log.warning.assert_has_calls(log_warn_calls)
|
||||
mock_log.debug.assert_has_calls(log_debug_calls)
|
||||
|
||||
@ -521,7 +521,7 @@ class TestPJLinkCommands(TestCase):
|
||||
process_command(projector=self.pjlink, cmd='INF2', data=chk_data)
|
||||
|
||||
# THEN: Data should be saved
|
||||
assert (self.pjlink.model == chk_data), 'Test data should have been saved'
|
||||
assert self.pjlink.model == chk_data, 'Test data should have been saved'
|
||||
mock_log.warning.assert_has_calls(log_warn_calls)
|
||||
mock_log.debug.assert_has_calls(log_debug_calls)
|
||||
|
||||
@ -544,6 +544,6 @@ class TestPJLinkCommands(TestCase):
|
||||
process_command(projector=self.pjlink, cmd='INFO', data=chk_data)
|
||||
|
||||
# THEN: Data should be saved
|
||||
assert (self.pjlink.other_info == chk_data), 'Test data should have been saved'
|
||||
assert self.pjlink.other_info == chk_data, 'Test data should have been saved'
|
||||
mock_log.warning.assert_has_calls(log_warn_calls)
|
||||
mock_log.debug.assert_has_calls(log_debug_calls)
|
||||
|
@ -35,7 +35,7 @@ from tests.resources.projector.data import TEST1_DATA
|
||||
|
||||
class TestPJLinkCommands(TestCase):
|
||||
"""
|
||||
Tests PJLink get status commands part 2
|
||||
Tests PJLink commands part 2
|
||||
"""
|
||||
def setUp(self):
|
||||
"""
|
||||
@ -67,7 +67,7 @@ class TestPJLinkCommands(TestCase):
|
||||
process_command(projector=self.pjlink, cmd='INPT', data='21')
|
||||
|
||||
# THEN: Input selected should reflect current input
|
||||
assert ('21' == self.pjlink.source), 'Input source should be set to "21"'
|
||||
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)
|
||||
|
||||
@ -88,7 +88,7 @@ class TestPJLinkCommands(TestCase):
|
||||
process_command(projector=self.pjlink, cmd='INPT', data='91')
|
||||
|
||||
# THEN: Input selected should reflect current input
|
||||
assert (not self.pjlink.source), 'Input source should not have changed'
|
||||
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)
|
||||
|
||||
@ -110,7 +110,7 @@ class TestPJLinkCommands(TestCase):
|
||||
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'
|
||||
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)
|
||||
|
||||
@ -138,7 +138,7 @@ class TestPJLinkCommands(TestCase):
|
||||
# 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"
|
||||
assert self.pjlink.source_available == chk_test, "Sources should have been sorted and saved"
|
||||
|
||||
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
|
||||
def test_projector_lamp_invalid_missing_data(self, mock_log):
|
||||
@ -158,7 +158,7 @@ class TestPJLinkCommands(TestCase):
|
||||
# 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'
|
||||
assert not self.pjlink.lamp, 'Projector lamp info should not have changed'
|
||||
|
||||
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
|
||||
def test_projector_lamp_invalid_nan(self, mock_log):
|
||||
@ -180,11 +180,11 @@ class TestPJLinkCommands(TestCase):
|
||||
# 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 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'
|
||||
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):
|
||||
@ -204,13 +204,13 @@ class TestPJLinkCommands(TestCase):
|
||||
# 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 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 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'
|
||||
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):
|
||||
@ -229,9 +229,9 @@ class TestPJLinkCommands(TestCase):
|
||||
# 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 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'
|
||||
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):
|
||||
@ -253,7 +253,7 @@ class TestPJLinkCommands(TestCase):
|
||||
# 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'
|
||||
assert self.pjlink.pjlink_name == chk_data, 'Name test data should have been saved'
|
||||
|
||||
@patch.object(openlp.core.projectors.pjlink.PJLink, 'send_command')
|
||||
@patch.object(openlp.core.projectors.pjlink.PJLink, 'change_status')
|
||||
@ -276,7 +276,7 @@ class TestPJLinkCommands(TestCase):
|
||||
# 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'
|
||||
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()
|
||||
@ -302,9 +302,9 @@ class TestPJLinkCommands(TestCase):
|
||||
# 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 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'
|
||||
assert not mock_send_command.called, 'send_command should not have been called'
|
||||
mock_change_status.assert_called_once_with(S_STANDBY)
|
||||
|
||||
@patch.object(openlp.core.projectors.pjlink.PJLink, 'send_command')
|
||||
@ -328,7 +328,7 @@ class TestPJLinkCommands(TestCase):
|
||||
# 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 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)
|
||||
@ -352,7 +352,7 @@ class TestPJLinkCommands(TestCase):
|
||||
# THEN: Filter model number should be 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'
|
||||
assert self.pjlink.model_filter == new_data, 'Filter model should have been saved'
|
||||
|
||||
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
|
||||
def test_projector_rfil_nosave(self, mock_log):
|
||||
@ -374,7 +374,7 @@ class TestPJLinkCommands(TestCase):
|
||||
process_command(projector=self.pjlink, cmd='RFIL', data=new_data)
|
||||
|
||||
# THEN: Filter model number should be saved
|
||||
assert (self.pjlink.model_filter != new_data), 'Filter model should NOT have been 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)
|
||||
|
||||
@ -397,7 +397,7 @@ class TestPJLinkCommands(TestCase):
|
||||
# THEN: Filter model number should be 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'
|
||||
assert self.pjlink.model_lamp == new_data, 'Lamp model should have been saved'
|
||||
|
||||
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
|
||||
def test_projector_rlmp_nosave(self, mock_log):
|
||||
@ -419,7 +419,7 @@ class TestPJLinkCommands(TestCase):
|
||||
process_command(projector=self.pjlink, cmd='RLMP', data=new_data)
|
||||
|
||||
# THEN: Filter model number should be saved
|
||||
assert (self.pjlink.model_lamp != new_data), 'Lamp model should NOT have been 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)
|
||||
|
||||
@ -446,7 +446,7 @@ class TestPJLinkCommands(TestCase):
|
||||
process_command(projector=self.pjlink, cmd='SNUM', data=new_data)
|
||||
|
||||
# THEN: Serial number should be set
|
||||
assert (self.pjlink.serial_no != new_data), 'Projector serial number should NOT have been 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)
|
||||
|
||||
@ -469,7 +469,7 @@ class TestPJLinkCommands(TestCase):
|
||||
process_command(projector=self.pjlink, cmd='SNUM', data=new_data)
|
||||
|
||||
# THEN: Serial number should be set
|
||||
assert (self.pjlink.serial_no == new_data), 'Projector serial number should have been 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)
|
||||
|
||||
@ -497,7 +497,7 @@ class TestPJLinkCommands(TestCase):
|
||||
process_command(self.pjlink, cmd='SVER', data=new_data)
|
||||
|
||||
# THEN: Version information should change
|
||||
assert (self.pjlink.sw_version == new_data), 'Software version should have changed'
|
||||
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)
|
||||
|
||||
@ -519,8 +519,8 @@ class TestPJLinkCommands(TestCase):
|
||||
process_command(projector=self.pjlink, cmd='SVER', data=new_data)
|
||||
|
||||
# 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'
|
||||
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)
|
||||
|
||||
@ -544,7 +544,7 @@ class TestPJLinkCommands(TestCase):
|
||||
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'
|
||||
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)
|
||||
|
@ -22,335 +22,232 @@
|
||||
"""
|
||||
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
|
||||
from openlp.core.projectors.constants import PJLINK_PORT, S_CONNECTED, S_OFF, S_ON
|
||||
from openlp.core.projectors.constants import E_NO_AUTHENTICATION, STATUS_CODE, S_AUTHENTICATE, S_CONNECT
|
||||
from openlp.core.projectors.db import Projector
|
||||
from openlp.core.projectors.pjlink import PJLink, PJLinkUDP
|
||||
from tests.resources.projector.data import TEST1_DATA, TEST2_DATA, TEST_HASH, TEST_PIN, TEST_SALT
|
||||
from openlp.core.projectors.pjlink import PJLink
|
||||
from openlp.core.projectors.pjlinkcommands import process_command
|
||||
from tests.resources.projector.data import TEST1_DATA, TEST_PIN, TEST_SALT
|
||||
|
||||
|
||||
class TestPJLinkCommands(TestCase):
|
||||
"""
|
||||
Tests for the PJLinkCommands class part 2
|
||||
Tests PJLink commands part 3
|
||||
"""
|
||||
@skip('Needs update to new setup')
|
||||
def test_projector_reset_information(self):
|
||||
def setUp(self):
|
||||
"""
|
||||
Test reset_information() resets all information and stops timers
|
||||
Initialize test state(s)
|
||||
"""
|
||||
# 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:
|
||||
# Default PJLink instance for tests
|
||||
self.pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
|
||||
|
||||
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)
|
||||
|
||||
@skip('Needs update to new setup')
|
||||
def test_process_pjlink_normal(self):
|
||||
def tearDown(self):
|
||||
"""
|
||||
Test initial connection prompt with no authentication
|
||||
Cleanup test state(s)
|
||||
"""
|
||||
# GIVEN: Initial mocks and data
|
||||
mock_log = patch.object(openlp.core.projectors.pjlink, "log").start()
|
||||
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()
|
||||
del(self.pjlink)
|
||||
|
||||
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
|
||||
pjlink.pin = None
|
||||
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")
|
||||
|
||||
# THEN: proper processing should have occured
|
||||
mock_log.debug.has_calls(log_check)
|
||||
mock_disconnect_from_host.assert_not_called()
|
||||
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)
|
||||
|
||||
@skip('Needs update to new setup')
|
||||
def test_process_pjlink_authenticate(self):
|
||||
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
|
||||
def test_process_pjlink_authenticate(self, mock_log):
|
||||
"""
|
||||
Test initial connection prompt with authentication
|
||||
"""
|
||||
# GIVEN: Initial mocks and data
|
||||
mock_log = patch.object(openlp.core.projectors.pjlink, "log").start()
|
||||
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()
|
||||
log_error_calls = []
|
||||
log_warning_calls = []
|
||||
log_debug_calls = [call('({ip}) Processing command "PJLINK" with data "1 {data}"'.format(ip=self.pjlink.name,
|
||||
data=TEST_SALT)),
|
||||
call('({ip}) Calling function for PJLINK'.format(ip=self.pjlink.name)),
|
||||
call('({ip}) Processing PJLINK command'.format(ip=self.pjlink.name)),
|
||||
call('({ip}) PJLINK: Returning {data}'.format(ip=self.pjlink.name,
|
||||
data=STATUS_CODE[S_AUTHENTICATE]))]
|
||||
|
||||
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
|
||||
pjlink.pin = TEST_PIN
|
||||
log_check = [call('({ip}) process_pjlink(): Sending "CLSS" initial command'.format(ip=pjlink.name)), ]
|
||||
self.pjlink.pin = TEST_PIN
|
||||
|
||||
# WHEN: process_pjlink called with no authentication required
|
||||
pjlink.process_pjlink(data='1 {salt}'.format(salt=TEST_SALT))
|
||||
chk = process_command(projector=self.pjlink, cmd='PJLINK', data='1 {salt}'.format(salt=TEST_SALT))
|
||||
|
||||
# THEN: proper processing should have occured
|
||||
mock_log.debug.has_calls(log_check)
|
||||
mock_disconnect_from_host.assert_not_called()
|
||||
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)
|
||||
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 chk == S_AUTHENTICATE, 'Should have returned {data}'.format(data=STATUS_CODE[S_AUTHENTICATE])
|
||||
|
||||
@skip('Needs update to new setup')
|
||||
def test_process_pjlink_normal_pin_set_error(self):
|
||||
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
|
||||
def test_process_pjlink_authenticate_pin_not_set_error(self, mock_log):
|
||||
"""
|
||||
Test initial connection prompt with authentication and no pin set
|
||||
"""
|
||||
# GIVEN: Initial mocks and data
|
||||
log_error_calls = [call('({ip}) Authenticate connection but no PIN - aborting'.format(ip=self.pjlink.name))]
|
||||
log_warning_calls = []
|
||||
log_debug_calls = [call('({ip}) Processing command "PJLINK" with data "1 {data}"'.format(ip=self.pjlink.name,
|
||||
data=TEST_SALT)),
|
||||
call('({ip}) Calling function for PJLINK'.format(ip=self.pjlink.name)),
|
||||
call('({ip}) Processing PJLINK command'.format(ip=self.pjlink.name))]
|
||||
|
||||
self.pjlink.pin = None
|
||||
|
||||
# WHEN: process_pjlink called with no authentication required
|
||||
chk = process_command(projector=self.pjlink, cmd='PJLINK', data='1 {salt}'.format(salt=TEST_SALT))
|
||||
|
||||
# THEN: proper processing should have occured
|
||||
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 chk == E_NO_AUTHENTICATION, \
|
||||
'Should have returned {data}'.format(data=STATUS_CODE[E_NO_AUTHENTICATION])
|
||||
|
||||
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
|
||||
def test_process_pjlink_authenticate_token_invalid(self, mock_log):
|
||||
"""
|
||||
Test initial connection prompt with authentication and bad token
|
||||
"""
|
||||
# GIVEN: Initial mocks and data
|
||||
bad_token = 'abcdefgh'
|
||||
log_error_calls = [call('({ip}) Authentication token invalid (not a hexadecimal number) - '
|
||||
'aborting'.format(ip=self.pjlink.name))]
|
||||
log_warning_calls = []
|
||||
log_debug_calls = [call('({ip}) Processing command "PJLINK" with data '
|
||||
'"1 {data}"'.format(ip=self.pjlink.name, data=bad_token)),
|
||||
call('({ip}) Calling function for PJLINK'.format(ip=self.pjlink.name)),
|
||||
call('({ip}) Processing PJLINK command'.format(ip=self.pjlink.name))]
|
||||
self.pjlink.pin = TEST_SALT
|
||||
|
||||
# WHEN: process_pjlink called with bad token
|
||||
chk = process_command(projector=self.pjlink, cmd='PJLINK', data='1 {data}'.format(data=bad_token))
|
||||
|
||||
# THEN: proper processing should have occured
|
||||
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 chk == E_NO_AUTHENTICATION, \
|
||||
'Should have returned {data}'.format(data=STATUS_CODE[E_NO_AUTHENTICATION])
|
||||
|
||||
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
|
||||
def test_process_pjlink_authenticate_token_length(self, mock_log):
|
||||
"""
|
||||
Test initial connection prompt with authentication and bad token
|
||||
"""
|
||||
# GIVEN: Initial mocks and data
|
||||
bad_token = '1234abcde' # Length should be 8, this is 9
|
||||
log_error_calls = [call('({ip}) Authentication token invalid (size) - '
|
||||
'aborting'.format(ip=self.pjlink.name))]
|
||||
log_warning_calls = []
|
||||
log_debug_calls = [call('({ip}) Processing command "PJLINK" with data '
|
||||
'"1 {data}"'.format(ip=self.pjlink.name, data=bad_token)),
|
||||
call('({ip}) Calling function for PJLINK'.format(ip=self.pjlink.name)),
|
||||
call('({ip}) Processing PJLINK command'.format(ip=self.pjlink.name))]
|
||||
self.pjlink.pin = TEST_SALT
|
||||
|
||||
# WHEN: process_pjlink called with bad token
|
||||
chk = process_command(projector=self.pjlink, cmd='PJLINK', data='1 {data}'.format(data=bad_token))
|
||||
|
||||
# THEN: proper processing should have occured
|
||||
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 chk == E_NO_AUTHENTICATION, \
|
||||
'Should have returned {data}'.format(data=STATUS_CODE[E_NO_AUTHENTICATION])
|
||||
|
||||
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
|
||||
def test_process_pjlink_authenticate_token_missing(self, mock_log):
|
||||
"""
|
||||
Test initial connection prompt with authentication and missing token
|
||||
"""
|
||||
# GIVEN: Initial mocks and data
|
||||
log_error_calls = [call('({ip}) Authenticated connection but not enough info - '
|
||||
'aborting'.format(ip=self.pjlink.name))]
|
||||
log_warning_calls = []
|
||||
log_debug_calls = [call('({ip}) Processing command "PJLINK" with data "1"'.format(ip=self.pjlink.name)),
|
||||
call('({ip}) Calling function for PJLINK'.format(ip=self.pjlink.name)),
|
||||
call('({ip}) Processing PJLINK command'.format(ip=self.pjlink.name))]
|
||||
|
||||
self.pjlink.pin = TEST_SALT
|
||||
|
||||
# WHEN: process_pjlink called with bad token
|
||||
chk = process_command(projector=self.pjlink, cmd='PJLINK', data='1')
|
||||
|
||||
# THEN: proper processing should have occured
|
||||
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 chk == E_NO_AUTHENTICATION, \
|
||||
'Should have returned {data}'.format(data=STATUS_CODE[E_NO_AUTHENTICATION])
|
||||
|
||||
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
|
||||
def test_process_pjlink_normal(self, mock_log):
|
||||
"""
|
||||
Test processing PJLINK initial prompt
|
||||
"""
|
||||
# GIVEN: Mocks and data
|
||||
log_error_calls = []
|
||||
log_warning_calls = []
|
||||
log_debug_calls = [call('({ip}) Processing command "PJLINK" with data "0"'.format(ip=self.pjlink.name)),
|
||||
call('({ip}) Calling function for PJLINK'.format(ip=self.pjlink.name)),
|
||||
call('({ip}) Processing PJLINK command'.format(ip=self.pjlink.name)),
|
||||
call('({ip}) PJLINK: Returning {data}'.format(ip=self.pjlink.name,
|
||||
data=STATUS_CODE[S_CONNECT]))]
|
||||
|
||||
self.pjlink.pin = None
|
||||
|
||||
# WHEN: process_pjlink called with no authentication required
|
||||
chk = process_command(projector=self.pjlink, cmd='PJLINK', data="0")
|
||||
|
||||
# THEN: proper processing should have occured
|
||||
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 chk == S_CONNECT, 'Should have returned {data}'.format(data=STATUS_CODE[S_CONNECT])
|
||||
|
||||
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
|
||||
def test_process_pjlink_normal_pin_set_error(self, mock_log):
|
||||
"""
|
||||
Test process_pjlinnk called with no authentication but pin is set
|
||||
"""
|
||||
# GIVEN: Initial mocks and data
|
||||
mock_log = patch.object(openlp.core.projectors.pjlink, 'log').start()
|
||||
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('({ip}) Normal connection but PIN set - aborting'.format(ip=pjlink.name)), ]
|
||||
log_error_calls = [call('({ip}) Normal connection but PIN set - '
|
||||
'aborting'.format(ip=self.pjlink.name))]
|
||||
log_warning_calls = []
|
||||
log_debug_calls = [call('({ip}) Processing command "PJLINK" with data "0"'.format(ip=self.pjlink.name)),
|
||||
call('({ip}) Calling function for PJLINK'.format(ip=self.pjlink.name)),
|
||||
call('({ip}) Processing PJLINK command'.format(ip=self.pjlink.name))]
|
||||
self.pjlink.pin = TEST_PIN
|
||||
|
||||
# WHEN: process_pjlink called with invalid authentication scheme
|
||||
pjlink.process_pjlink(data='0')
|
||||
chk = process_command(projector=self.pjlink, cmd='PJLINK', data='0')
|
||||
|
||||
# THEN: Proper calls should be made
|
||||
mock_log.error.assert_has_calls(log_check)
|
||||
assert 1 == mock_disconnect_from_host.call_count, 'Should have only been called once'
|
||||
mock_send_command.assert_not_called()
|
||||
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 chk == E_NO_AUTHENTICATION, \
|
||||
'Should have returned {data}'.format(data=STATUS_CODE[E_NO_AUTHENTICATION])
|
||||
|
||||
@skip('Needs update to new setup')
|
||||
def test_process_pjlink_normal_with_salt_error(self):
|
||||
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
|
||||
def test_process_pjlink_normal_with_token(self, mock_log):
|
||||
"""
|
||||
Test process_pjlinnk called with no authentication but pin is set
|
||||
"""
|
||||
# GIVEN: Initial mocks and data
|
||||
mock_log = patch.object(openlp.core.projectors.pjlink, 'log').start()
|
||||
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('({ip}) Normal connection with extra information - aborting'.format(ip=pjlink.name)), ]
|
||||
log_error_calls = [call('({ip}) Normal connection with extra information - '
|
||||
'aborting'.format(ip=self.pjlink.name))]
|
||||
log_warning_calls = []
|
||||
log_debug_calls = [call('({ip}) Processing command "PJLINK" with data '
|
||||
'"0 {data}"'.format(ip=self.pjlink.name, data=TEST_SALT)),
|
||||
call('({ip}) Calling function for PJLINK'.format(ip=self.pjlink.name)),
|
||||
call('({ip}) Processing PJLINK command'.format(ip=self.pjlink.name))]
|
||||
self.pjlink.pin = TEST_PIN
|
||||
|
||||
# WHEN: process_pjlink called with invalid authentication scheme
|
||||
pjlink.process_pjlink(data='0 {salt}'.format(salt=TEST_SALT))
|
||||
chk = process_command(projector=self.pjlink, cmd='PJLINK', data='0 {data}'.format(data=TEST_SALT))
|
||||
|
||||
# THEN: Proper calls should be made
|
||||
mock_log.error.assert_has_calls(log_check)
|
||||
assert 1 == mock_disconnect_from_host.call_count, 'Should have only been called once'
|
||||
mock_send_command.assert_not_called()
|
||||
|
||||
@skip('Needs update to new setup')
|
||||
def test_process_pjlink_invalid_authentication_scheme_length_error(self):
|
||||
"""
|
||||
Test initial connection prompt with authentication scheme longer than 1 character
|
||||
"""
|
||||
# GIVEN: Initial mocks and data
|
||||
mock_log = patch.object(openlp.core.projectors.pjlink, 'log').start()
|
||||
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)
|
||||
assert 1 == mock_disconnect_from_host.call_count, 'Should have only been called once'
|
||||
mock_send_command.assert_not_called()
|
||||
|
||||
@skip('Needs update to new setup')
|
||||
def test_process_pjlink_invalid_authentication_data_length_error(self):
|
||||
"""
|
||||
Test initial connection prompt with authentication no salt
|
||||
"""
|
||||
# GIVEN: Initial mocks and data
|
||||
mock_log = patch.object(openlp.core.projectors.pjlink, 'log').start()
|
||||
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)
|
||||
assert 1 == mock_disconnect_from_host.call_count, 'Should have only been called once'
|
||||
mock_send_command.assert_not_called()
|
||||
|
||||
@skip('Needs update to new setup')
|
||||
def test_process_pjlink_authenticate_pin_not_set_error(self):
|
||||
"""
|
||||
Test process_pjlink authentication but pin not set
|
||||
"""
|
||||
# GIVEN: Initial mocks and data
|
||||
mock_log = patch.object(openlp.core.projectors.pjlink, 'log').start()
|
||||
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)
|
||||
assert 1 == mock_disconnect_from_host.call_count, 'Should have only been called once'
|
||||
mock_send_command.assert_not_called()
|
||||
|
||||
@skip('Change to pjlink_udp.get_datagram() call')
|
||||
@patch.object(openlp.core.projectors.pjlink, 'log')
|
||||
def test_process_ackn_duplicate(self, mock_log):
|
||||
"""
|
||||
Test process_ackn method with multiple calls with same data
|
||||
"""
|
||||
# TODO: Change this to call pjlink_udp.get_datagram() so ACKN can be processed properly
|
||||
|
||||
# GIVEN: Test setup
|
||||
pjlink = PJLink(projector=self.test_list[0])
|
||||
check_list = {TEST1_DATA['ip']: {'data': TEST1_DATA['mac_adx'], 'port': PJLINK_PORT}}
|
||||
log_warn_calls = [call('(___TEST_ONE___) Host {host} already replied - '
|
||||
'ignoring'.format(host=TEST1_DATA['ip']))]
|
||||
log_debug_calls = [call('PJlinkCommands(args=() kwargs={})'),
|
||||
call('(___TEST_ONE___) reset_information() connect status is S_NOT_CONNECTED'),
|
||||
call('(___TEST_ONE___) Processing ACKN packet'),
|
||||
call('(___TEST_ONE___) Adding {host} to ACKN list'.format(host=TEST1_DATA['ip'])),
|
||||
call('(___TEST_ONE___) Processing ACKN packet')]
|
||||
|
||||
# WHEN: process_ackn called twice with same data
|
||||
pjlink.process_ackn(data=TEST1_DATA['mac_adx'], host=TEST1_DATA['ip'], port=PJLINK_PORT)
|
||||
pjlink.process_ackn(data=TEST1_DATA['mac_adx'], host=TEST1_DATA['ip'], port=PJLINK_PORT)
|
||||
|
||||
# THEN: pjlink_udp.ack_list should equal test_list
|
||||
# NOTE: This assert only returns AssertionError - does not list differences. Maybe add a compare function?
|
||||
if pjlink.ackn_list != check_list:
|
||||
# Check this way so we can print differences to stdout
|
||||
print('\nackn_list: ', pjlink.ackn_list)
|
||||
print('test_list: ', check_list, '\n')
|
||||
assert pjlink.ackn_list == check_list
|
||||
mock_log.debug.assert_has_calls(log_debug_calls)
|
||||
mock_log.warning.assert_has_calls(log_warn_calls)
|
||||
|
||||
@skip('Change to pjlink_udp.get_datagram() call')
|
||||
@patch.object(openlp.core.projectors.pjlink, 'log')
|
||||
def test_process_ackn_multiple(self, mock_log):
|
||||
"""
|
||||
Test process_ackn method with multiple calls
|
||||
"""
|
||||
# TODO: Change this to call pjlink_udp.get_datagram() so ACKN can be processed properly
|
||||
|
||||
# GIVEN: Test setup
|
||||
pjlink_udp = PJLinkUDP(projector_list=self.test_list)
|
||||
check_list = {TEST1_DATA['ip']: {'data': TEST1_DATA['mac_adx'], 'port': PJLINK_PORT},
|
||||
TEST2_DATA['ip']: {'data': TEST2_DATA['mac_adx'], 'port': PJLINK_PORT}}
|
||||
log_debug_calls = [call('(UDP) PJLinkUDP() Initialized'),
|
||||
call('(UDP) Processing ACKN packet'),
|
||||
call('(UDP) Adding {host} to ACKN list'.format(host=TEST1_DATA['ip'])),
|
||||
call('(UDP) Processing ACKN packet'),
|
||||
call('(UDP) Adding {host} to ACKN list'.format(host=TEST2_DATA['ip']))]
|
||||
|
||||
# WHEN: process_ackn called twice with different data
|
||||
pjlink_udp.process_ackn(data=TEST1_DATA['mac_adx'], host=TEST1_DATA['ip'], port=PJLINK_PORT)
|
||||
pjlink_udp.process_ackn(data=TEST2_DATA['mac_adx'], host=TEST2_DATA['ip'], port=PJLINK_PORT)
|
||||
|
||||
# THEN: pjlink_udp.ack_list should equal test_list
|
||||
# NOTE: This assert only returns AssertionError - does not list differences. Maybe add a compare function?
|
||||
if pjlink_udp.ackn_list != check_list:
|
||||
# Check this way so we can print differences to stdout
|
||||
print('\nackn_list: ', pjlink_udp.ackn_list)
|
||||
print('test_list: ', check_list)
|
||||
assert pjlink_udp.ackn_list == check_list
|
||||
mock_log.debug.assert_has_calls(log_debug_calls)
|
||||
|
||||
@skip('Change to pjlink_udp.get_datagram() call')
|
||||
@patch.object(openlp.core.projectors.pjlink, 'log')
|
||||
def test_process_ackn_single(self, mock_log):
|
||||
"""
|
||||
Test process_ackn method with single call
|
||||
"""
|
||||
# TODO: Change this to call pjlink_udp.get_datagram() so ACKN can be processed properly
|
||||
|
||||
# GIVEN: Test setup
|
||||
pjlink_udp = PJLinkUDP(projector_list=self.test_list)
|
||||
check_list = {TEST1_DATA['ip']: {'data': TEST1_DATA['mac_adx'], 'port': PJLINK_PORT}}
|
||||
log_debug_calls = [call('(UDP) PJLinkUDP() Initialized'),
|
||||
call('(UDP) Processing ACKN packet'),
|
||||
call('(UDP) Adding {host} to ACKN list'.format(host=TEST1_DATA['ip']))]
|
||||
|
||||
# WHEN: process_ackn called twice with different data
|
||||
pjlink_udp.process_ackn(data=TEST1_DATA['mac_adx'], host=TEST1_DATA['ip'], port=PJLINK_PORT)
|
||||
|
||||
# THEN: pjlink_udp.ack_list should equal test_list
|
||||
# NOTE: This assert only returns AssertionError - does not list differences. Maybe add a compare function?
|
||||
if pjlink_udp.ackn_list != check_list:
|
||||
# Check this way so we can print differences to stdout
|
||||
print('\nackn_list: ', pjlink_udp.ackn_list)
|
||||
print('test_list: ', check_list)
|
||||
assert pjlink_udp.ackn_list == check_list
|
||||
mock_log.debug.assert_has_calls(log_debug_calls)
|
||||
|
||||
@skip('Change to pjlink_udp.get_datagram() call')
|
||||
@patch.object(openlp.core.projectors.pjlink, 'log')
|
||||
def test_process_srch(self, mock_log):
|
||||
"""
|
||||
Test process_srch method
|
||||
"""
|
||||
# TODO: Change this to call pjlink_udp.get_datagram() so ACKN can be processed properly
|
||||
|
||||
# GIVEN: Test setup
|
||||
log_warn_calls = [call('(UDP) SRCH packet received from {ip} - ignoring'.format(ip=TEST1_DATA['ip'])), ]
|
||||
log_debug_calls = [call('(UDP) PJLinkUDP() Initialized'), ]
|
||||
pjlink_udp = PJLinkUDP(projector_list=self.test_list)
|
||||
|
||||
# WHEN: process_srch called
|
||||
pjlink_udp.process_srch(data=None, host=TEST1_DATA['ip'], port=PJLINK_PORT)
|
||||
|
||||
# THEN: log entries should be entered
|
||||
mock_log.warning.assert_has_calls(log_warn_calls)
|
||||
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 chk == E_NO_AUTHENTICATION, \
|
||||
'Should have returned {data}'.format(data=STATUS_CODE[E_NO_AUTHENTICATION])
|
||||
|
@ -20,7 +20,7 @@
|
||||
# along with this program. If not, see <https://www.gnu.org/licenses/>. #
|
||||
##########################################################################
|
||||
"""
|
||||
Package to test the openlp.core.projectors.pjlink base package.
|
||||
Package to test the openlp.core.projectors.pjlink base package part 1.
|
||||
"""
|
||||
from unittest import TestCase
|
||||
from unittest.mock import MagicMock, call, patch
|
||||
|
@ -20,7 +20,7 @@
|
||||
# along with this program. If not, see <https://www.gnu.org/licenses/>. #
|
||||
##########################################################################
|
||||
"""
|
||||
Package to test the openlp.core.projectors.pjlink base package.
|
||||
Package to test the openlp.core.projectors.pjlink base package part 2.
|
||||
"""
|
||||
from unittest import TestCase
|
||||
from unittest.mock import call, patch
|
||||
@ -96,10 +96,10 @@ class TestPJLinkBase(TestCase):
|
||||
mock_log.warning.assert_has_calls(log_warning_calls)
|
||||
mock_log.debug.assert_has_calls(log_debug_calls)
|
||||
mock_change_status.called_with(E_NETWORK, 'Error while sending data to projector')
|
||||
assert (not self.pjlink.send_queue), 'Send queue should be empty'
|
||||
assert (not self.pjlink.priority_queue), 'Priority queue should be empty'
|
||||
assert not self.pjlink.send_queue, 'Send queue should be empty'
|
||||
assert not self.pjlink.priority_queue, 'Priority queue should be empty'
|
||||
assert mock_timer.start.called, 'Timer should have been called'
|
||||
assert (not mock_reset.called), 'reset_information() should not should have been called'
|
||||
assert not mock_reset.called, 'reset_information() should not should have been called'
|
||||
assert mock_disconnect.called, 'disconnect_from_host() should have been called'
|
||||
assert self.pjlink.send_busy, 'send_busy should be True'
|
||||
|
||||
@ -127,10 +127,10 @@ class TestPJLinkBase(TestCase):
|
||||
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.send_queue), 'Send queue should be empty'
|
||||
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'
|
||||
assert not self.pjlink.send_queue, 'Send queue should be empty'
|
||||
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')
|
||||
@ -173,11 +173,11 @@ class TestPJLinkBase(TestCase):
|
||||
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.send_queue), 'Send queue should be empty'
|
||||
assert (not self.pjlink.priority_queue), 'Priority queue should be empty'
|
||||
assert not self.pjlink.send_queue, 'Send queue should be empty'
|
||||
assert not self.pjlink.priority_queue, 'Priority queue should be empty'
|
||||
assert mock_timer.start.called, 'Timer should have been called'
|
||||
assert (not mock_reset.called), 'reset_information() should not have been called'
|
||||
assert (not mock_disconnect.called), 'disconnect_from_host() should not have been called'
|
||||
assert not mock_reset.called, 'reset_information() should not have been called'
|
||||
assert not mock_disconnect.called, 'disconnect_from_host() should not have been called'
|
||||
assert self.pjlink.send_busy, 'send_busy flag should be True'
|
||||
|
||||
@patch.object(openlp.core.projectors.pjlink.PJLink, 'disconnect_from_host')
|
||||
@ -211,12 +211,12 @@ class TestPJLinkBase(TestCase):
|
||||
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 (self.pjlink.send_queue == [test_command]), 'Send queue should have one entry'
|
||||
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'
|
||||
assert self.pjlink.send_queue == [test_command], 'Send queue should have one entry'
|
||||
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'
|
||||
assert mock_disconnect.called, 'disconnect_from_host() should have been called'
|
||||
assert (not self.pjlink.send_busy), 'send_busy flag should be False'
|
||||
assert not self.pjlink.send_busy, 'send_busy flag should be False'
|
||||
|
||||
@patch.object(openlp.core.projectors.pjlink.PJLink, 'write')
|
||||
@patch.object(openlp.core.projectors.pjlink.PJLink, 'disconnect_from_host')
|
||||
@ -261,11 +261,11 @@ class TestPJLinkBase(TestCase):
|
||||
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.send_queue), 'Send queue should be empty'
|
||||
assert (not self.pjlink.priority_queue), 'Priority queue should be empty'
|
||||
assert not self.pjlink.send_queue, 'Send queue should be empty'
|
||||
assert not self.pjlink.priority_queue, 'Priority queue should be empty'
|
||||
assert mock_timer.start.called, 'Timer should have been called'
|
||||
assert (not mock_reset.called), 'reset_information() should not have been called'
|
||||
assert (not mock_disconnect.called), 'disconnect_from_host() should not have been called'
|
||||
assert not mock_reset.called, 'reset_information() should not have been called'
|
||||
assert not mock_disconnect.called, 'disconnect_from_host() should not have been called'
|
||||
assert self.pjlink.send_busy, 'send_busy flag should be True'
|
||||
|
||||
@patch.object(openlp.core.projectors.pjlink.PJLink, 'write')
|
||||
@ -314,10 +314,10 @@ class TestPJLinkBase(TestCase):
|
||||
mock_log.warning.assert_has_calls(log_warning_calls)
|
||||
mock_log.debug.assert_has_calls(log_debug_calls)
|
||||
assert self.pjlink.send_queue, 'Send queue should have one entry'
|
||||
assert (not self.pjlink.priority_queue), 'Priority queue should be empty'
|
||||
assert not self.pjlink.priority_queue, 'Priority queue should be empty'
|
||||
assert mock_timer.start.called, 'Timer should have been called'
|
||||
assert (not mock_reset.called), 'reset_information() should not have been called'
|
||||
assert (not mock_disconnect.called), 'disconnect_from_host() should not have been called'
|
||||
assert not mock_reset.called, 'reset_information() should not have been called'
|
||||
assert not mock_disconnect.called, 'disconnect_from_host() should not have been called'
|
||||
assert self.pjlink.send_busy, 'send_busy flag should be True'
|
||||
|
||||
@patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
|
||||
@ -358,9 +358,9 @@ class TestPJLinkBase(TestCase):
|
||||
mock_log.warning.assert_has_calls(log_warning_calls)
|
||||
mock_log.debug.assert_has_calls(log_debug_calls)
|
||||
assert self.pjlink.send_queue, 'Send queue should have one entry'
|
||||
assert (not self.pjlink.priority_queue), 'Priority queue should be empty'
|
||||
assert (not mock_timer.start.called), 'Timer should not have been called'
|
||||
assert (not mock_reset.called), 'reset_information() should not have been called'
|
||||
assert not self.pjlink.priority_queue, 'Priority queue should be empty'
|
||||
assert not mock_timer.start.called, 'Timer should not have been called'
|
||||
assert not mock_reset.called, 'reset_information() should not have been called'
|
||||
assert self.pjlink.send_busy, 'send_busy flag should be True'
|
||||
|
||||
@patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
|
||||
@ -402,10 +402,10 @@ class TestPJLinkBase(TestCase):
|
||||
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.send_queue), 'Send queue should be empty'
|
||||
assert not self.pjlink.send_queue, 'Send queue should be empty'
|
||||
assert self.pjlink.priority_queue, 'Priority queue should have one entry'
|
||||
assert (not mock_timer.start.called), 'Timer should not have been called'
|
||||
assert (not mock_reset.called), 'reset_information() should not have been called'
|
||||
assert not mock_timer.start.called, 'Timer should not have been called'
|
||||
assert not mock_reset.called, 'reset_information() should not have been called'
|
||||
assert self.pjlink.send_busy, 'send_busy flag should be True'
|
||||
|
||||
# ------------ Test PJLink.send_command ----------
|
||||
@ -440,7 +440,7 @@ class TestPJLinkBase(TestCase):
|
||||
mock_log.debug.assert_has_calls(log_debug_calls)
|
||||
mock_log.warning.assert_has_calls(log_warning_calls)
|
||||
mock_log.error.assert_has_calls(log_error_calls)
|
||||
assert (not mock_reset.called), 'reset_information() should not have been called'
|
||||
assert not mock_reset.called, 'reset_information() should not have been called'
|
||||
assert mock_send_command.called, '_underscore_send_command() should have been called'
|
||||
|
||||
@patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
|
||||
@ -473,8 +473,8 @@ class TestPJLinkBase(TestCase):
|
||||
mock_log.warning.assert_has_calls(log_warning_calls)
|
||||
mock_log.error.assert_has_calls(log_error_calls)
|
||||
mock_priority.append.assert_called_with(test_command)
|
||||
assert (not mock_send.append.called), 'send_queue should not have changed'
|
||||
assert (not mock_reset.called), 'reset_information() should not have been called'
|
||||
assert not mock_send.append.called, 'send_queue should not have changed'
|
||||
assert not mock_reset.called, 'reset_information() should not have been called'
|
||||
assert mock_send_command.called, '_underscore_send_command() should have been called'
|
||||
|
||||
@patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
|
||||
@ -506,9 +506,9 @@ class TestPJLinkBase(TestCase):
|
||||
mock_log.debug.assert_has_calls(log_debug_calls)
|
||||
mock_log.warning.assert_has_calls(log_warning_calls)
|
||||
mock_log.error.assert_has_calls(log_error_calls)
|
||||
assert (self.pjlink.send_queue == [test_command]), 'Send queue should have one entry'
|
||||
assert (not self.pjlink.priority_queue), 'Priority queue should be empty'
|
||||
assert (not mock_reset.called), 'reset_information() should not have been called'
|
||||
assert self.pjlink.send_queue == [test_command], 'Send queue should have one entry'
|
||||
assert not self.pjlink.priority_queue, 'Priority queue should be empty'
|
||||
assert not mock_reset.called, 'reset_information() should not have been called'
|
||||
assert mock_send_command.called, '_underscore_send_command() should have been called'
|
||||
|
||||
@patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
|
||||
@ -540,9 +540,9 @@ class TestPJLinkBase(TestCase):
|
||||
mock_log.debug.assert_has_calls(log_debug_calls)
|
||||
mock_log.warning.assert_has_calls(log_warning_calls)
|
||||
mock_log.error.assert_has_calls(log_error_calls)
|
||||
assert (not self.pjlink.send_queue), 'Send queue should be empty'
|
||||
assert (self.pjlink.priority_queue == [test_command]), 'Priority queue should have one entry'
|
||||
assert (not mock_reset.called), 'reset_information() should not have been called'
|
||||
assert not self.pjlink.send_queue, 'Send queue should be empty'
|
||||
assert self.pjlink.priority_queue == [test_command], 'Priority queue should have one entry'
|
||||
assert not mock_reset.called, 'reset_information() should not have been called'
|
||||
assert mock_send_command.called, '_underscore_send_command() should have been called'
|
||||
|
||||
@patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
|
||||
@ -569,10 +569,10 @@ class TestPJLinkBase(TestCase):
|
||||
mock_log.debug.assert_has_calls(log_debug_calls)
|
||||
mock_log.warning.assert_has_calls(log_warning_calls)
|
||||
mock_log.error.assert_has_calls(log_error_calls)
|
||||
assert (not self.pjlink.send_queue), 'Send queue should be empty'
|
||||
assert (not self.pjlink.priority_queue), 'Priority queue should be empty'
|
||||
assert (not mock_reset.called), 'reset_information() should not have been called'
|
||||
assert (not mock_send_command.called), '_underscore_send_command() should not have been called'
|
||||
assert not self.pjlink.send_queue, 'Send queue should be empty'
|
||||
assert not self.pjlink.priority_queue, 'Priority queue should be empty'
|
||||
assert not mock_reset.called, 'reset_information() should not have been called'
|
||||
assert not mock_send_command.called, '_underscore_send_command() should not have been called'
|
||||
|
||||
@patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
|
||||
@patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
|
||||
@ -602,8 +602,8 @@ class TestPJLinkBase(TestCase):
|
||||
mock_log.warning.assert_has_calls(log_warning_calls)
|
||||
mock_log.error.assert_has_calls(log_error_calls)
|
||||
assert self.pjlink.send_queue, 'Send queue should have one entry'
|
||||
assert (not self.pjlink.priority_queue), 'Priority queue should be empty'
|
||||
assert (not mock_reset.called), 'reset_information() should not have been called'
|
||||
assert not self.pjlink.priority_queue, 'Priority queue should be empty'
|
||||
assert not mock_reset.called, 'reset_information() should not have been called'
|
||||
assert mock_send_command.called, '_underscore_send_command() should have been called'
|
||||
|
||||
@patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
|
||||
@ -633,9 +633,9 @@ class TestPJLinkBase(TestCase):
|
||||
mock_log.debug.assert_has_calls(log_debug_calls)
|
||||
mock_log.warning.assert_has_calls(log_warning_calls)
|
||||
mock_log.error.assert_has_calls(log_error_calls)
|
||||
assert (not self.pjlink.send_queue), 'Send queue should be empty'
|
||||
assert not self.pjlink.send_queue, 'Send queue should be empty'
|
||||
assert self.pjlink.priority_queue, 'Priority queue should have one entry'
|
||||
assert (not mock_reset.called), 'reset_information() should not have been called'
|
||||
assert not mock_reset.called, 'reset_information() should not have been called'
|
||||
assert mock_send_command.called, '_underscore_send_command() should have been called'
|
||||
|
||||
@patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
|
||||
@ -661,7 +661,7 @@ class TestPJLinkBase(TestCase):
|
||||
mock_log.debug.assert_has_calls(log_debug_calls)
|
||||
mock_log.warning.assert_has_calls(log_warning_calls)
|
||||
mock_log.error.assert_has_calls(log_error_calls)
|
||||
assert (not self.pjlink.send_queue), 'Send queue should be empty'
|
||||
assert (not self.pjlink.priority_queue), 'Priority queue should be empty'
|
||||
assert not self.pjlink.send_queue, 'Send queue should be empty'
|
||||
assert not self.pjlink.priority_queue, 'Priority queue should be empty'
|
||||
assert mock_reset.called, 'reset_information() should have been called'
|
||||
assert (not mock_send_command.called), '_underscore_send_command() should not have been called'
|
||||
assert not mock_send_command.called, '_underscore_send_command() should not have been called'
|
||||
|
130
tests/openlp_core/projectors/test_projector_pjlink_base_03.py
Normal file
130
tests/openlp_core/projectors/test_projector_pjlink_base_03.py
Normal file
@ -0,0 +1,130 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
# vim: autoindent shiftwidth=4 expandtab textwidth=120 tabstop=4 softtabstop=4
|
||||
|
||||
##########################################################################
|
||||
# OpenLP - Open Source Lyrics Projection #
|
||||
# ---------------------------------------------------------------------- #
|
||||
# Copyright (c) 2008-2019 OpenLP Developers #
|
||||
# ---------------------------------------------------------------------- #
|
||||
# This program is free software: you can redistribute it and/or modify #
|
||||
# it under the terms of the GNU General Public License as published by #
|
||||
# the Free Software Foundation, either version 3 of the License, or #
|
||||
# (at your option) any later version. #
|
||||
# #
|
||||
# This program is distributed in the hope that it will be useful, #
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of #
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the #
|
||||
# GNU General Public License for more details. #
|
||||
# #
|
||||
# You should have received a copy of the GNU General Public License #
|
||||
# along with this program. If not, see <https://www.gnu.org/licenses/>. #
|
||||
##########################################################################
|
||||
"""
|
||||
Package to test the openlp.core.projectors.pjlink base package part 3.
|
||||
"""
|
||||
from unittest import TestCase
|
||||
from unittest.mock import call, patch
|
||||
|
||||
import openlp.core.projectors.pjlink
|
||||
from openlp.core.projectors.constants import PJLINK_CLASS, STATUS_CODE, \
|
||||
S_NOT_CONNECTED, S_OFF, 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
|
||||
|
||||
|
||||
class TestPJLinkBase(TestCase):
|
||||
"""
|
||||
Tests for the PJLink module
|
||||
"""
|
||||
def setUp(self):
|
||||
"""
|
||||
Initialize test state(s)
|
||||
"""
|
||||
# Default PJLink instance for tests
|
||||
self.pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
|
||||
|
||||
def tearDown(self):
|
||||
"""
|
||||
Cleanup test state(s)
|
||||
"""
|
||||
del(self.pjlink)
|
||||
|
||||
@patch.object(openlp.core.projectors.pjlink, 'log')
|
||||
def test_projector_reset_information(self, mock_log):
|
||||
"""
|
||||
Test reset_information() resets all information and stops timers
|
||||
"""
|
||||
# GIVEN: Test object
|
||||
log_debug_calls = [call('({ip}) reset_information() connect status is '
|
||||
'S_NOT_CONNECTED'.format(ip=self.pjlink.name)),
|
||||
call('({ip}): Calling poll_timer.stop()'.format(ip=self.pjlink.name)),
|
||||
call('({ip}): Calling socket_timer.stop()'.format(ip=self.pjlink.name)),
|
||||
call('({ip}): Calling status_timer.stop()'.format(ip=self.pjlink.name))]
|
||||
|
||||
# Attributes not available until instantiation, so mock here
|
||||
with patch.object(self.pjlink, 'socket_timer') as mock_socket_timer, \
|
||||
patch.object(self.pjlink, 'status_timer') as mock_status_timer, \
|
||||
patch.object(self.pjlink, 'poll_timer') as mock_poll_timer, \
|
||||
patch.object(self.pjlink, 'state') as mock_state:
|
||||
mock_state.return_value = QSOCKET_STATE[S_NOT_CONNECTED]
|
||||
# Set attributes to something other than None or {} or []
|
||||
self.pjlink.fan = True
|
||||
self.pjlink.filter_time = True
|
||||
self.pjlink.lamp = True
|
||||
self.pjlink.mac_adx_received = 'Some random MAC'
|
||||
self.pjlink.manufacturer = 'PJLINK'
|
||||
self.pjlink.model = '1'
|
||||
self.pjlink.model_filter = 'Filter'
|
||||
self.pjlink.model_lamp = 'Lamp'
|
||||
self.pjlink.mute = True
|
||||
self.pjlink.other_info = 'Another Test'
|
||||
self.pjlink.pjlink_class = 2
|
||||
self.pjlink.pjlink_name = 'OPENLPTEST'
|
||||
self.pjlink.power = S_ON
|
||||
self.pjlink.projector_errors = {'test1': True, 'test2': False}
|
||||
self.pjlink.serial_no = 'Some Number'
|
||||
self.pjlink.serial_no_received = 'Some Other Number'
|
||||
self.pjlink.sw_version = 'Some Version'
|
||||
self.pjlink.sw_version_received = 'Some Other Version'
|
||||
self.pjlink.shutter = True
|
||||
self.pjlink.source_available = True
|
||||
self.pjlink.source = True
|
||||
self.pjlink.status_timer_checks = {'test1': object(), 'test2': object()}
|
||||
self.pjlink.send_busy = False
|
||||
self.pjlink.send_queue = ['test1', 'test2']
|
||||
self.pjlink.priority_queue = ['test1', 'test2']
|
||||
|
||||
# WHEN: reset_information() is called
|
||||
self.pjlink.reset_information()
|
||||
|
||||
# THEN: All information should be reset and timers stopped
|
||||
mock_log.debug.assert_has_calls(log_debug_calls)
|
||||
assert self.pjlink.fan is None, 'fan should be None'
|
||||
assert self.pjlink.filter_time is None, 'filter_time should be None'
|
||||
assert self.pjlink.lamp is None, 'lamp should be None'
|
||||
assert self.pjlink.mac_adx_received is None, 'mac_adx_received should be None'
|
||||
assert self.pjlink.manufacturer is None, 'manufacturer should be None'
|
||||
assert self.pjlink.model is None, 'model should be None'
|
||||
assert self.pjlink.model_filter is None, 'model_filter should be None'
|
||||
assert self.pjlink.model_lamp is None, 'model_lamp should be None'
|
||||
assert not self.pjlink.mute, 'mute should be False'
|
||||
assert self.pjlink.other_info is None, 'other should be None'
|
||||
assert self.pjlink.pjlink_class == PJLINK_CLASS, 'pjlink_class should be {cls}'.format(cls=PJLINK_CLASS)
|
||||
assert self.pjlink.pjlink_name is None, 'pjlink_name should be None'
|
||||
assert self.pjlink.power == S_OFF, 'power should be {data}'.format(data=STATUS_CODE[S_OFF])
|
||||
assert self.pjlink.projector_errors == {}, 'projector_errors should be an empty dict'
|
||||
assert self.pjlink.serial_no is None, 'serial_no should be None'
|
||||
assert self.pjlink.serial_no_received is None, 'serial_no_received should be None'
|
||||
assert self.pjlink.sw_version is None, 'sw_version should be None'
|
||||
assert self.pjlink.sw_version_received is None, 'sw_version_received should be None'
|
||||
assert not self.pjlink.shutter, 'shutter should be False'
|
||||
assert self.pjlink.source_available is None, 'source_available should be None'
|
||||
assert self.pjlink.source is None, 'source should be None'
|
||||
assert self.pjlink.status_timer_checks == {}, 'status_timer_checks should be an empty dict'
|
||||
assert not self.pjlink.send_busy, 'send_busy should be False'
|
||||
assert self.pjlink.send_queue == [], 'send_queue should be an empty list'
|
||||
assert self.pjlink.priority_queue == [], 'priority_queue should be an empty list'
|
||||
assert mock_socket_timer.stop.called, 'socket_timer.stop() should have been called'
|
||||
assert mock_status_timer.stop.called, 'status_timer.stop() should have been called'
|
||||
assert mock_poll_timer.stop.called, 'poll_timer.stop() should have been called'
|
Loading…
Reference in New Issue
Block a user