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

PJLink2 Update v07

This commit is contained in:
Ken Roberts 2019-05-11 02:07:40 -07:00
parent b4d5425887
commit 0c14ddf31f
9 changed files with 345 additions and 314 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
openlp/core/projectors/constants.py
View File

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

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

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

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

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

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

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

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

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

475
tests/openlp_core/projectors/test_projector_commands_03.py
View File

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

2
tests/openlp_core/projectors/test_projector_pjlink_base_01.py
View File

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

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

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

130
tests/openlp_core/projectors/test_projector_pjlink_base_03.py Normal file
View 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'