openlp-core team mailing list archive
-
openlp-core team
-
Mailing list archive
-
Message #33977
[Merge] lp:~alisonken1/openlp/pjlink2-v07 into lp:openlp
Ken Roberts has proposed merging lp:~alisonken1/openlp/pjlink2-v07 into lp:openlp with lp:~alisonken1/openlp/pjlink2-v06 as a prerequisite.
Commit message:
PJLink 2 Update v07
Requested reviews:
OpenLP Core (openlp-core)
For more details, see:
https://code.launchpad.net/~alisonken1/openlp/pjlink2-v07/+merge/367304
NOTE: Part 7 of a multi-part merge.
v[1..n] merges are to fix tests
- Move reset_information() test from commands_03 module to pjlink_base_03 module
- Add missing attribute pjlink.projector_errors {} for ERST data
- Remove extraneous translates from process_erst()
- Added import string to pjlinkcommands to validate authentication token
- Fix return codes for PJLINK command
- Update/add tests for PJLINK command
- Remove tests for commands not handled yet
NOTE: MAC offline, all tests except 2 MAC tests passed
--
Your team OpenLP Core is requested to review the proposed merge of lp:~alisonken1/openlp/pjlink2-v07 into lp:openlp.
=== modified file 'openlp/core/projectors/constants.py'
--- openlp/core/projectors/constants.py 2019-05-04 05:25:07 +0000
+++ openlp/core/projectors/constants.py 2019-05-11 09:24:01 +0000
@@ -39,6 +39,7 @@
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
=== modified file 'openlp/core/projectors/pjlink.py'
--- openlp/core/projectors/pjlink.py 2019-05-05 04:08:32 +0000
+++ openlp/core/projectors/pjlink.py 2019-05-11 09:24:01 +0000
@@ -281,6 +281,7 @@
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
=== modified file 'openlp/core/projectors/pjlinkcommands.py'
--- openlp/core/projectors/pjlinkcommands.py 2019-05-05 04:08:32 +0000
+++ openlp/core/projectors/pjlinkcommands.py 2019-05-11 09:24:01 +0000
@@ -29,13 +29,13 @@
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 @@
# 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 @@
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 @@
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
=== modified file 'tests/openlp_core/projectors/test_projector_commands_01.py'
--- tests/openlp_core/projectors/test_projector_commands_01.py 2019-05-05 04:08:32 +0000
+++ tests/openlp_core/projectors/test_projector_commands_01.py 2019-05-11 09:24:01 +0000
@@ -35,7 +35,7 @@
class TestPJLinkCommands(TestCase):
"""
- Tests PJLink get status commands part 1
+ Tests PJLink commands part 1
"""
def setUp(self):
"""
=== modified file 'tests/openlp_core/projectors/test_projector_commands_02.py'
--- tests/openlp_core/projectors/test_projector_commands_02.py 2019-05-05 04:08:32 +0000
+++ tests/openlp_core/projectors/test_projector_commands_02.py 2019-05-11 09:24:01 +0000
@@ -35,7 +35,7 @@
class TestPJLinkCommands(TestCase):
"""
- Tests PJLink get status commands part 2
+ Tests PJLink commands part 2
"""
def setUp(self):
"""
=== modified file 'tests/openlp_core/projectors/test_projector_commands_03.py'
--- tests/openlp_core/projectors/test_projector_commands_03.py 2019-05-05 04:08:32 +0000
+++ tests/openlp_core/projectors/test_projector_commands_03.py 2019-05-11 09:24:01 +0000
@@ -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):
- """
- Test reset_information() resets all information and stops timers
- """
- # GIVEN: Test object
- with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log:
- pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
- log_debug_calls = [call('({ip}): Calling poll_timer.stop()'.format(ip=pjlink.name)),
- call('({ip}): Calling socket_timer.stop()'.format(ip=pjlink.name))]
- # timer and socket_timer not available until instantiation, so mock here
- with patch.object(pjlink, 'socket_timer') as mock_socket_timer, \
- patch.object(pjlink, 'poll_timer') as mock_timer:
-
- pjlink.power = S_ON
- pjlink.pjlink_name = 'OPENLPTEST'
- pjlink.manufacturer = 'PJLINK'
- pjlink.model = '1'
- pjlink.shutter = True
- pjlink.mute = True
- pjlink.lamp = True
- pjlink.fan = True
- pjlink.source_available = True
- pjlink.other_info = 'ANOTHER TEST'
- pjlink.send_queue = True
- pjlink.send_busy = True
-
- # WHEN: reset_information() is called
- pjlink.reset_information()
-
- # THEN: All information should be reset and timers stopped
- assert pjlink.power == S_OFF, 'Projector power should be OFF'
- assert pjlink.pjlink_name is None, 'Projector pjlink_name should be None'
- assert pjlink.manufacturer is None, 'Projector manufacturer should be None'
- assert pjlink.model is None, 'Projector model should be None'
- assert pjlink.shutter is None, 'Projector shutter should be None'
- assert pjlink.mute is None, 'Projector shuttter should be None'
- assert pjlink.lamp is None, 'Projector lamp should be None'
- assert pjlink.fan is None, 'Projector fan should be None'
- assert pjlink.source_available is None, 'Projector source_available should be None'
- assert pjlink.source is None, 'Projector source should be None'
- assert pjlink.other_info is None, 'Projector other_info should be None'
- assert pjlink.send_queue == [], 'Projector send_queue should be an empty list'
- assert pjlink.send_busy is False, 'Projector send_busy should be False'
- assert mock_timer.stop.called is True, 'Projector timer.stop() should have been called'
- assert mock_socket_timer.stop.called is True, 'Projector socket_timer.stop() should have been called'
- mock_log.debug.assert_has_calls(log_debug_calls)
-
- @skip('Needs update to new setup')
- def test_process_pjlink_normal(self):
- """
- Test initial connection prompt with no 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()
-
- 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):
+ 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.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()
-
- 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)), ]
-
- # WHEN: process_pjlink called with no authentication required
- pjlink.process_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)
-
- @skip('Needs update to new setup')
- def test_process_pjlink_normal_pin_set_error(self):
- """
- 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)), ]
-
- # WHEN: process_pjlink called with invalid authentication scheme
- pjlink.process_pjlink(data='0')
-
- # THEN: Proper calls should be made
- mock_log.error.assert_has_calls(log_check)
- 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_normal_with_salt_error(self):
- """
- 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)), ]
-
- # WHEN: process_pjlink called with invalid authentication scheme
- pjlink.process_pjlink(data='0 {salt}'.format(salt=TEST_SALT))
-
- # THEN: Proper calls should be made
- mock_log.error.assert_has_calls(log_check)
- 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.debug.assert_has_calls(log_debug_calls)
+ 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]))]
+
+ self.pjlink.pin = TEST_PIN
+
+ # 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 == S_AUTHENTICATE), 'Should have returned {data}'.format(data=STATUS_CODE[S_AUTHENTICATE])
+
+ @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
+ 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
+ chk = process_command(projector=self.pjlink, cmd='PJLINK', data='0')
+
+ # THEN: Proper calls should be made
+ 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_with_token(self, mock_log):
+ """
+ Test process_pjlinnk called with no authentication but pin is set
+ """
+ # GIVEN: Initial mocks and data
+ 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
+ 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_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])
=== modified file 'tests/openlp_core/projectors/test_projector_pjlink_base_01.py'
--- tests/openlp_core/projectors/test_projector_pjlink_base_01.py 2019-04-28 19:21:23 +0000
+++ tests/openlp_core/projectors/test_projector_pjlink_base_01.py 2019-05-11 09:24:01 +0000
@@ -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
=== modified file 'tests/openlp_core/projectors/test_projector_pjlink_base_02.py'
--- tests/openlp_core/projectors/test_projector_pjlink_base_02.py 2019-05-05 04:08:32 +0000
+++ tests/openlp_core/projectors/test_projector_pjlink_base_02.py 2019-05-11 09:24:01 +0000
@@ -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
=== added file 'tests/openlp_core/projectors/test_projector_pjlink_base_03.py'
--- tests/openlp_core/projectors/test_projector_pjlink_base_03.py 1970-01-01 00:00:00 +0000
+++ tests/openlp_core/projectors/test_projector_pjlink_base_03.py 2019-05-11 09:24:01 +0000
@@ -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'
Follow ups
