openlp-core team mailing list archive
-
openlp-core team
-
Mailing list archive
-
Message #33900
[Merge] lp:~alisonken1/openlp/pjlink2_v04 into lp:openlp
Ken Roberts has proposed merging lp:~alisonken1/openlp/pjlink2_v04 into lp:openlp.
Commit message:
PJLink2 update V04
Requested reviews:
OpenLP Core (openlp-core)
For more details, see:
https://code.launchpad.net/~alisonken1/openlp/pjlink2_v04/+merge/366946
NOTE: Part 4 of a multi-part merge.
v[1..n] merges are to fix tests
- Fixed/Added routing tests for PJLink.get_data() method
- Move process_command() routing tests to separate module
- Refactor/add process_command() tests
- Refactor process_pjlink() command and tests
- Fix PJLink socket state check causing loss of connection (regression)
- Added 3 new status codes for initial connection processing
--------------------------------------------------------------------------------
lp:~alisonken1/openlp/pjlink2_v04 (revision 2865)
https://ci.openlp.io/job/Branch-01-Pull/2729/ [SUCCESS]
https://ci.openlp.io/job/Branch-02a-Linux-Tests/2623/ [SUCCESS]
https://ci.openlp.io/job/Branch-02b-macOS-Tests/398/ [SUCCESS]
https://ci.openlp.io/job/Branch-03a-Build-Source/220/ [SUCCESS]
https://ci.openlp.io/job/Branch-03b-Build-macOS/199/ [SUCCESS]
https://ci.openlp.io/job/Branch-04a-Code-Lint/1682/ [SUCCESS]
https://ci.openlp.io/job/Branch-04b-Test-Coverage/1495/ [SUCCESS]
https://ci.openlp.io/job/Branch-05-AppVeyor-Tests/375/ [SUCCESS]
All builds passed
--
Your team OpenLP Core is requested to review the proposed merge of lp:~alisonken1/openlp/pjlink2_v04 into lp:openlp.
=== modified file 'openlp/core/projectors/constants.py'
--- openlp/core/projectors/constants.py 2019-04-13 13:00:22 +0000
+++ openlp/core/projectors/constants.py 2019-05-04 05:43:42 +0000
@@ -106,6 +106,9 @@
S_ON = 315
S_COOLDOWN = 316
S_INFO = 317
+S_CONNECT = 318 # Initial connection, connected
+S_AUTHENTICATE = 319 # Initial connection, send pin hash
+S_DATA_OK = 320 # Previous command returned OK
# Information that does not affect status
S_NETWORK_IDLE = 400
@@ -369,11 +372,14 @@
E_UNKNOWN_SOCKET_ERROR: 'E_UNKNOWN_SOCKET_ERROR',
E_UNSUPPORTED_SOCKET_OPERATION: 'E_UNSUPPORTED_SOCKET_OPERATION',
E_WARN: 'E_WARN',
+ S_AUTHENTICATE: 'S_AUTHENTICATE',
S_BOUND: 'S_BOUND',
+ S_CONNECT: 'S_CONNECT',
S_COOLDOWN: 'S_COOLDOWN',
S_CLOSING: 'S_CLOSING',
S_CONNECTED: 'S_CONNECTED',
S_CONNECTING: 'S_CONNECTING',
+ S_DATA_OK: 'S_DATA_OK',
S_HOST_LOOKUP: 'S_HOST_LOOKUP',
S_INFO: 'S_INFO',
S_INITIALIZE: 'S_INITIALIZE',
@@ -387,7 +393,7 @@
S_ON: 'S_ON',
S_STANDBY: 'S_STANDBY',
S_STATUS: 'S_STATUS',
- S_WARMUP: 'S_WARMUP',
+ S_WARMUP: 'S_WARMUP'
}
# Map status codes to message strings
@@ -459,11 +465,14 @@
'The requested socket operation is not supported by the local '
'operating system (e.g., lack of IPv6 support)'),
E_WARN: translate('OpenLP.ProjectorConstants', 'Warning condition detected'),
+ S_AUTHENTICATE: translate('OpenLP.ProjectorConstants', 'Connection initializing with pin'),
S_BOUND: translate('OpenLP.ProjectorConstants', 'Socket is bount to an address or port'),
+ S_CONNECT: translate('OpenLP.ProjectorConstants', 'Connection initializing'),
S_CLOSING: translate('OpenLP.ProjectorConstants', 'Socket is about to close'),
S_CONNECTED: translate('OpenLP.ProjectorConstants', 'Connected'),
S_CONNECTING: translate('OpenLP.ProjectorConstants', 'Connecting'),
S_COOLDOWN: translate('OpenLP.ProjectorConstants', 'Cooldown in progress'),
+ S_DATA_OK: translate('OpenLP.ProjectorConstants', 'Command returned with OK'),
S_HOST_LOOKUP: translate('OpenLP.ProjectorConstants', 'Performing a host name lookup'),
S_INFO: translate('OpenLP.ProjectorConstants', 'Projector Information available'),
S_INITIALIZE: translate('OpenLP.ProjectorConstants', 'Initialize in progress'),
=== modified file 'openlp/core/projectors/pjlink.py'
--- openlp/core/projectors/pjlink.py 2019-04-28 19:21:23 +0000
+++ openlp/core/projectors/pjlink.py 2019-05-04 05:43:42 +0000
@@ -52,14 +52,15 @@
from PyQt5 import QtCore, QtNetwork
+from openlp.core.common import qmd5_hash
from openlp.core.common.i18n import translate
from openlp.core.common.settings import Settings
from openlp.core.projectors.pjlinkcommands import process_command
-from openlp.core.projectors.constants import CONNECTION_ERRORS, E_CONNECTION_REFUSED, E_GENERAL, \
+from openlp.core.projectors.constants import CONNECTION_ERRORS, E_AUTHENTICATION, E_CONNECTION_REFUSED, E_GENERAL, \
E_NETWORK, E_NOT_CONNECTED, E_SOCKET_TIMEOUT, PJLINK_CLASS, \
PJLINK_MAX_PACKET, PJLINK_PORT, PJLINK_PREFIX, PJLINK_SUFFIX, \
- PJLINK_VALID_CMD, PROJECTOR_STATE, QSOCKET_STATE, S_CONNECTED, S_CONNECTING, S_NOT_CONNECTED, S_OFF, S_OK, S_ON, \
- STATUS_CODE, STATUS_MSG
+ PJLINK_VALID_CMD, PROJECTOR_STATE, QSOCKET_STATE, S_AUTHENTICATE, S_CONNECT, S_CONNECTED, S_CONNECTING, \
+ S_DATA_OK, S_NOT_CONNECTED, S_OFF, S_OK, S_ON, STATUS_CODE, STATUS_MSG
log = logging.getLogger(__name__)
@@ -565,21 +566,10 @@
# V = PJLink class or version
# C = PJLink command
version, cmd = header[1], header[2:].upper()
- log.debug('({ip}) get_data() version="{version}" cmd="{cmd}"'.format(ip=self.entry.name,
- version=version, cmd=cmd))
- # TODO: Below commented for now since it seems to cause issues with testing some invalid data.
- # Revisit after more refactoring is finished.
- '''
- try:
- version, cmd = header[1], header[2:].upper()
- log.debug('({ip}) get_data() version="{version}" cmd="{cmd}"'.format(ip=self.entry.name,
- version=version, cmd=cmd))
- except ValueError as e:
- self.change_status(E_INVALID_DATA)
- log.warning('({ip}) get_data(): Received data: "{data}"'.format(ip=self.entry.name, data=data_in))
- self._trash_buffer('get_data(): Expected header + command + data')
- return self.receive_data_signal()
- '''
+ log.debug('({ip}) get_data() version="{version}" cmd="{cmd}" data="{data}"'.format(ip=self.entry.name,
+ version=version,
+ cmd=cmd,
+ data=data))
if cmd not in PJLINK_VALID_CMD:
self._trash_buffer('get_data(): Invalid packet - unknown command "{data}"'.format(data=cmd))
return self.receive_data_signal()
@@ -590,7 +580,38 @@
if not ignore_class:
log.warning('({ip}) get_data(): Projector returned class reply higher '
'than projector stated class'.format(ip=self.entry.name))
- process_command(self, cmd, data)
+ return self.receive_data_signal()
+
+ chk = process_command(self, cmd, data)
+ if chk is None:
+ # Command processed normally and not initial connection, so skip other checks
+ return self.receive_data_signal()
+ # PJLink initial connection checks
+ elif chk == S_DATA_OK:
+ # Previous command returned OK
+ log.debug('({ip}) OK returned - resending command'.format(ip=self.entry.name))
+ self.send_command(cmd=cmd, priority=True)
+ elif chk == S_CONNECT:
+ # Normal connection
+ log.debug('({ip}) Connecting normal'.format(ip=self.entry.name))
+ self.change_status(S_CONNECTED)
+ self.send_command(cmd='CLSS', priority=True)
+ self.readyRead.connect(self.get_socket)
+ elif chk == S_AUTHENTICATE:
+ # Connection with pin
+ log.debug('({ip}) Connecting with pin'.format(ip=self.entry.name))
+ data_hash = str(qmd5_hash(salt=chk[1].encode('utf-8'), data=self.pin.encode('utf-8')),
+ encoding='ascii')
+ self.change_status(S_CONNECTED)
+ self.readyRead.connect(self.get_socket)
+ self.send_command(cmd='CLSS', salt=data_hash, priority=True)
+ elif chk == E_AUTHENTICATION:
+ # Projector did not like our pin
+ log.warning('({ip}) Failed authentication - disconnecting'.format(ip=self.entry.name))
+ self.disconnect_from_host()
+ self.projectorAuthentication.emit(self.entry.name)
+ self.change_status(status=E_AUTHENTICATION)
+
return self.receive_data_signal()
@QtCore.pyqtSlot(QtNetwork.QAbstractSocket.SocketError)
@@ -631,7 +652,9 @@
:param salt: Optional salt for md5 hash initial authentication
:param priority: Option to send packet now rather than queue it up
"""
- if QSOCKET_STATE[self.state()] != QSOCKET_STATE[S_CONNECTED]:
+ log.debug('({ip}) send_command(cmd="{cmd}" opts="{opts}" salt="{salt}" '
+ 'priority={pri}'.format(ip=self.entry.name, cmd=cmd, opts=opts, salt=salt, pri=priority))
+ if QSOCKET_STATE[self.state()] != S_CONNECTED:
log.warning('({ip}) send_command(): Not connected - returning'.format(ip=self.entry.name))
return self.reset_information()
if cmd not in PJLINK_VALID_CMD:
@@ -640,11 +663,11 @@
# Just in case there's already something to send
return self._send_command()
return
- log.debug('({ip}) send_command(): Building cmd="{command}" opts="{data}"{salt}'.format(ip=self.entry.name,
- command=cmd,
- data=opts,
- salt='' if salt is None
- else ' with hash'))
+ log.debug('({ip}) send_command(): Building cmd="{command}" opts="{data}" '
+ '{salt}'.format(ip=self.entry.name,
+ command=cmd,
+ data=opts,
+ salt='' if salt is None else 'with hash'))
# Until we absolutely have to start doing version checks, use the default
# for PJLink class
header = PJLINK_HEADER.format(linkclass=PJLINK_VALID_CMD[cmd]['default'])
=== modified file 'openlp/core/projectors/pjlinkcommands.py'
--- openlp/core/projectors/pjlinkcommands.py 2019-04-28 19:22:52 +0000
+++ openlp/core/projectors/pjlinkcommands.py 2019-05-04 05:43:42 +0000
@@ -30,14 +30,12 @@
import logging
import re
-from openlp.core.common import qmd5_hash
-
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_CONNECTED, S_OFF, S_OK, S_ON, S_STANDBY, \
- STATUS_MSG
+ 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
log = logging.getLogger(__name__)
log.debug('Loading pjlinkcommands')
@@ -68,19 +66,18 @@
elif _data == 'OK':
log.debug('({ip}) Command "{cmd}" returned OK'.format(ip=projector.entry.name, cmd=cmd))
# A command returned successfully, so do a query on command to verify status
- return projector.send_command(cmd=cmd, priority=True)
+ return S_DATA_OK
+
elif _data in PJLINK_ERRORS:
# Oops - projector error
log.error('({ip}) {cmd}: {err}'.format(ip=projector.entry.name,
cmd=cmd,
err=STATUS_MSG[PJLINK_ERRORS[_data]]))
- if PJLINK_ERRORS[_data] == E_AUTHENTICATION:
- projector.disconnect_from_host()
- projector.projectorAuthentication.emit(projector.name)
- return projector.change_status(status=E_AUTHENTICATION)
+ return PJLINK_ERRORS[_data]
+
# Command checks already passed
log.debug('({ip}) Calling function for {cmd}'.format(ip=projector.entry.name, cmd=cmd))
- pjlink_functions[cmd](projector=projector, data=data)
+ return pjlink_functions[cmd](projector=projector, data=data)
def process_ackn(projector, data):
@@ -376,35 +373,28 @@
if len(chk[0]) != 1:
# Invalid - after splitting, first field should be 1 character, either '0' or '1' only
log.error('({ip}) Invalid initial authentication scheme - aborting'.format(ip=projector.entry.name))
- return projector.disconnect_from_host()
+ return E_AUTHENTICATION
elif chk[0] == '0':
# Normal connection no authentication
if len(chk) > 1:
# Invalid data - there should be nothing after a normal authentication scheme
log.error('({ip}) Normal connection with extra information - aborting'.format(ip=projector.entry.name))
- return projector.disconnect_from_host()
+ return E_AUTHENTICATION
elif projector.pin:
log.error('({ip}) Normal connection but PIN set - aborting'.format(ip=projector.entry.name))
- return projector.disconnect_from_host()
- else:
- data_hash = None
+ return E_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 projector.disconnect_from_host()
+ return E_AUTHENTICATION
elif not projector.pin:
log.error('({ip}) Authenticate connection but no PIN - aborting'.format(ip=projector.entry.name))
- return projector.disconnect_from_host()
- else:
- data_hash = str(qmd5_hash(salt=chk[1].encode('utf-8'), data=projector.pin.encode('utf-8')),
- encoding='ascii')
- # Passed basic checks, so start connection
- projector.readyRead.connect(projector.get_socket)
- projector.change_status(S_CONNECTED)
- log.debug('({ip}) process_pjlink(): Sending "CLSS" initial command'.format(ip=projector.entry.name))
- # Since this is an initial connection, make it a priority just in case
- return projector.send_command(cmd="CLSS", salt=data_hash, priority=True)
+ return E_AUTHENTICATION
+ log.debug('({ip}) PJLINK: Returning S_AUTHENTICATE'.format(ip=projector.entry.name))
+ return S_AUTHENTICATE
def process_powr(projector, data):
=== added file 'tests/openlp_core/projectors/test_projector_command_routing.py'
--- tests/openlp_core/projectors/test_projector_command_routing.py 1970-01-01 00:00:00 +0000
+++ tests/openlp_core/projectors/test_projector_command_routing.py 2019-05-04 05:43:42 +0000
@@ -0,0 +1,132 @@
+# -*- 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 command routing.
+"""
+
+from unittest import TestCase
+from unittest.mock import call, patch
+
+import openlp.core.projectors.pjlink
+from openlp.core.projectors.pjlinkcommands import process_command
+from openlp.core.projectors.constants import E_UNDEFINED, S_DATA_OK
+from openlp.core.projectors.db import Projector
+from openlp.core.projectors.pjlink import PJLink
+from tests.resources.projector.data import TEST1_DATA
+
+
+class TestPJLinkRouting(TestCase):
+ """
+ Tests for the PJLink module command routing
+ """
+ def setUp(self):
+ """
+ Setup test environment
+ """
+ self.pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
+
+ def tearDown(self):
+ """
+ Reset test environment
+ """
+ del(self.pjlink)
+
+ @patch.object(openlp.core.projectors.pjlinkcommands, 'log')
+ def test_routing_command(self, mock_log):
+ """
+ Test process_command receiving command not in function map
+ """
+ # GIVEN: Test setup
+ log_warning_text = []
+ log_debug_text = [call('({ip}) Processing command "CLSS" with data "1"'.format(ip=self.pjlink.name)),
+ call('({ip}) Calling function for CLSS'.format(ip=self.pjlink.name)),
+ call('({ip}) Setting pjlink_class for this projector to "1"'.format(ip=self.pjlink.name))]
+
+ # WHEN: called with valid command and data
+ chk = process_command(projector=self.pjlink, cmd='CLSS', data='1')
+
+ # THEN: Appropriate log entries should have been made and methods called/not called
+ mock_log.warning.assert_has_calls(log_warning_text)
+ mock_log.debug.assert_has_calls(log_debug_text)
+ assert (chk is None), 'process_clss() should have returned None'
+
+ @patch.object(openlp.core.projectors.pjlinkcommands, 'process_clss')
+ @patch.object(openlp.core.projectors.pjlinkcommands, 'pjlink_functions')
+ @patch.object(openlp.core.projectors.pjlinkcommands, 'log')
+ def test_routing_command_unknown(self, mock_log, mock_functions, mock_clss):
+ """
+ Test process_command receiving command not in function map
+ """
+ # GIVEN: Test setup
+ log_warning_text = [call('({ip}) Unable to process command="CLSS" '
+ '(Future option?)'.format(ip=self.pjlink.name))]
+ log_debug_text = [call('({ip}) Processing command "CLSS" with data "?"'.format(ip=self.pjlink.name))]
+ # Fake CLSS command is not in list
+ mock_functions.__contains__.return_value = False
+
+ # WHEN: called with unknown command
+ process_command(projector=self.pjlink, cmd='CLSS', data='?')
+
+ # THEN: Appropriate log entries should have been made and methods called/not called
+ mock_log.warning.assert_has_calls(log_warning_text)
+ mock_log.debug.assert_has_calls(log_debug_text)
+ assert (mock_functions.__contains__.call_count == 1), 'pjlink_functions should have been accessed only once'
+ assert (not mock_clss.called), 'Should not have called process_clss'
+
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'send_command')
+ @patch.object(openlp.core.projectors.pjlinkcommands, 'process_clss')
+ @patch.object(openlp.core.projectors.pjlinkcommands, 'log')
+ def test_routing_data_ok(self, mock_log, mock_clss, mock_send):
+ """
+ Test process_command calls function and sets appropriate value(s) in projector instance
+ """
+ # GIVEN: Test setup
+ log_debug_text = [call('({ip}) Processing command "CLSS" with data "OK"'.format(ip=self.pjlink.name)),
+ call('({ip}) Command "CLSS" returned OK'.format(ip=self.pjlink.name))
+ ]
+
+ # WHEN: Command called with OK
+ chk = process_command(projector=self.pjlink, cmd='CLSS', data='OK')
+
+ # THEN: Appropriate log entries should have been made and methods called/not called
+ mock_log.debug.asset_has_calls(log_debug_text)
+ mock_send.assert_not_called()
+ mock_clss.assert_not_called()
+ assert (chk == S_DATA_OK), 'Should have returned S_DATA_OK'
+
+ @patch.object(openlp.core.projectors.pjlinkcommands, 'log')
+ def test_routing_pjink_errors(self, mock_log):
+ """
+ Test rouing when PJLink error received (err1, err2, err3, err4, erra)
+ """
+ # GIVEN: Test setup
+ log_error_text = [call('({ip}) CLSS: PJLink returned "ERR1: Undefined Command"'.format(ip=self.pjlink.name))]
+ log_debug_text = [call('({ip}) Processing command "CLSS" with data "ERR1"'.format(ip=self.pjlink.name))]
+ err_code = E_UNDEFINED
+
+ # WHEN: routing called
+ chk = process_command(projector=self.pjlink, cmd='CLSS', data='ERR1')
+
+ # THEN: Appropriate log entries should have been made and methods called/not called
+ mock_log.error.assert_has_calls(log_error_text)
+ mock_log.debug.assert_has_calls(log_debug_text)
+ assert (chk == err_code), 'Should have returned E_UNDEFINED'
=== modified file 'tests/openlp_core/projectors/test_projector_constants.py'
--- tests/openlp_core/projectors/test_projector_constants.py 2019-04-13 13:00:22 +0000
+++ tests/openlp_core/projectors/test_projector_constants.py 2019-05-04 05:43:42 +0000
@@ -40,11 +40,12 @@
missing_str = []
# GIVEN: List of defined E_* and S_* items defined in constants
+
for item in constants.__dict__:
if item.startswith('E_') or item.startswith('S_'):
check.append(item)
- # WHEN: Verify defined list against STATUS_STR
+ # WHEN: Verify items were addeded to check
for item in check:
if constants.__dict__[item] not in STATUS_CODE:
missing_str.append(item)
=== modified file 'tests/openlp_core/projectors/test_projector_pjlink_base_02.py'
--- tests/openlp_core/projectors/test_projector_pjlink_base_02.py 2019-04-28 19:21:23 +0000
+++ tests/openlp_core/projectors/test_projector_pjlink_base_02.py 2019-05-04 05:43:42 +0000
@@ -454,9 +454,9 @@
suff=PJLINK_SUFFIX)
log_error_calls = []
log_warning_calls = []
- log_debug_calls = [call('({ip}) send_command(): Building cmd="CLSS" opts="?"'.format(ip=self.pjlink.name)),
+ log_debug_calls = [call('({ip}) send_command(): Building cmd="CLSS" opts="?" '.format(ip=self.pjlink.name)),
call('({ip}) send_command(): Adding to normal queue'.format(ip=self.pjlink.name))]
- mock_state.return_value = S_CONNECTED
+ mock_state.return_value = QSOCKET_STATE[S_CONNECTED]
# Patch here since pjlink does not have priority or send queue's until instantiated
with patch.object(self.pjlink, 'send_queue') as mock_send, \
@@ -488,9 +488,9 @@
suff=PJLINK_SUFFIX)
log_error_calls = []
log_warning_calls = []
- log_debug_calls = [call('({ip}) send_command(): Building cmd="CLSS" opts="?"'.format(ip=self.pjlink.name)),
+ log_debug_calls = [call('({ip}) send_command(): Building cmd="CLSS" opts="?" '.format(ip=self.pjlink.name)),
call('({ip}) send_command(): Adding to priority queue'.format(ip=self.pjlink.name))]
- mock_state.return_value = S_CONNECTED
+ mock_state.return_value = QSOCKET_STATE[S_CONNECTED]
# Patch here since pjlink does not have priority or send queue's until instantiated
with patch.object(self.pjlink, 'send_queue') as mock_send, \
@@ -523,8 +523,10 @@
log_error_calls = []
log_warning_calls = [call('({ip}) send_command(): Already in normal queue - '
'skipping'.format(ip=self.pjlink.name))]
- log_debug_calls = [call('({ip}) send_command(): Building cmd="CLSS" opts="?"'.format(ip=self.pjlink.name))]
- mock_state.return_value = S_CONNECTED
+ log_debug_calls = [call('({ip}) send_command(cmd="CLSS" opts="?" salt="None" '
+ 'priority=False'.format(ip=self.pjlink.name)),
+ call('({ip}) send_command(): Building cmd="CLSS" opts="?" '.format(ip=self.pjlink.name))]
+ mock_state.return_value = QSOCKET_STATE[S_CONNECTED]
self.pjlink.send_queue = [test_command]
self.pjlink.priority_queue = []
@@ -555,8 +557,10 @@
log_error_calls = []
log_warning_calls = [call('({ip}) send_command(): Already in priority queue - '
'skipping'.format(ip=self.pjlink.name))]
- log_debug_calls = [call('({ip}) send_command(): Building cmd="CLSS" opts="?"'.format(ip=self.pjlink.name))]
- mock_state.return_value = S_CONNECTED
+ log_debug_calls = [call('({ip}) send_command(cmd="CLSS" opts="?" salt="None" '
+ 'priority=True'.format(ip=self.pjlink.name)),
+ call('({ip}) send_command(): Building cmd="CLSS" opts="?" '.format(ip=self.pjlink.name))]
+ mock_state.return_value = QSOCKET_STATE[S_CONNECTED]
self.pjlink.send_queue = []
self.pjlink.priority_queue = [test_command]
@@ -585,7 +589,7 @@
'ignoring.'.format(ip=self.pjlink.name))]
log_warning_calls = []
log_debug_calls = []
- mock_state.return_value = S_CONNECTED
+ mock_state.return_value = QSOCKET_STATE[S_CONNECTED]
self.pjlink.send_queue = []
self.pjlink.priority_queue = []
@@ -617,7 +621,7 @@
'ignoring.'.format(ip=self.pjlink.name))]
log_warning_calls = []
log_debug_calls = []
- mock_state.return_value = S_CONNECTED
+ mock_state.return_value = QSOCKET_STATE[S_CONNECTED]
self.pjlink.send_queue = [test_command]
self.pjlink.priority_queue = []
@@ -649,7 +653,7 @@
'ignoring.'.format(ip=self.pjlink.name))]
log_warning_calls = []
log_debug_calls = []
- mock_state.return_value = S_CONNECTED
+ mock_state.return_value = QSOCKET_STATE[S_CONNECTED]
self.pjlink.send_queue = []
self.pjlink.priority_queue = [test_command]
=== modified file 'tests/openlp_core/projectors/test_projector_pjlink_cmd_routing.py'
--- tests/openlp_core/projectors/test_projector_pjlink_cmd_routing.py 2019-04-28 19:21:23 +0000
+++ tests/openlp_core/projectors/test_projector_pjlink_cmd_routing.py 2019-05-04 05:43:42 +0000
@@ -24,10 +24,9 @@
"""
from unittest import TestCase, skip
-from unittest.mock import MagicMock, call, patch
+from unittest.mock import call, patch
import openlp.core.projectors.pjlink
-from openlp.core.projectors.pjlinkcommands import process_command
from openlp.core.projectors.constants import E_AUTHENTICATION, E_PARAMETER, E_PROJECTOR, E_UNAVAILABLE, E_UNDEFINED, \
PJLINK_ERRORS, PJLINK_PREFIX, STATUS_MSG
from openlp.core.projectors.db import Projector
@@ -51,82 +50,93 @@
"""
del(self.pjlink)
- @patch.object(openlp.core.projectors.pjlink, 'log')
- def test_get_data_unknown_command(self, mock_log):
- """
- Test not a valid command
- """
- # GIVEN: Test object
- self.pjlink.pjlink_functions = MagicMock()
+ @patch.object(openlp.core.projectors.pjlinkcommands, 'process_command')
+ @patch.object(openlp.core.projectors.pjlink, 'log')
+ def test_projector_get_data_invalid_version(self, mock_log, mock_process_cmd):
+ """
+ Test projector received valid command invalid version
+ """
+ # GIVEN: Test object
+ log_warning_text = [call('({ip}) get_data() Command reply version does not match '
+ 'a valid command version'.format(ip=self.pjlink.name)),
+ call('({ip}) _send_command(): Nothing to send - returning'.format(ip=self.pjlink.name))]
+ log_debug_text = [call('({ip}) get_data(buffer="{pre}XCLSS=X"'.format(ip=self.pjlink.name, pre=PJLINK_PREFIX)),
+ call('({ip}) get_data(): Checking new data "{pre}XCLSS=X"'.format(ip=self.pjlink.name,
+ pre=PJLINK_PREFIX)),
+ call('({ip}) get_data() header="{pre}XCLSS" data="X"'.format(ip=self.pjlink.name,
+ pre=PJLINK_PREFIX)),
+ call('({ip}) get_data() version="X" cmd="CLSS" data="X"'.format(ip=self.pjlink.name)),
+ call('({ip}) Cleaning buffer - msg = "get_data() Command reply version does '
+ 'not match a valid command version"'.format(ip=self.pjlink.name)),
+ call('({ip}) Finished cleaning buffer - 0 bytes dropped'.format(ip=self.pjlink.name))]
+ # WHEN: get_data called with an unknown command
+ self.pjlink.get_data(buff='{prefix}XCLSS=X'.format(prefix=PJLINK_PREFIX))
+
+ # THEN: Appropriate log entries should have been made and methods called/not called
+ mock_log.warning.assert_has_calls(log_warning_text)
+ mock_log.debug.assert_has_calls(log_debug_text)
+ assert (mock_process_cmd.call_count == 0), 'process_command should not have been called'
+
+ @patch.object(openlp.core.projectors.pjlinkcommands, 'process_command')
+ @patch.object(openlp.core.projectors.pjlink, 'log')
+ def test_projector_get_data_unknown_command(self, mock_log, mock_process_cmd):
+ """
+ Test projector receiving invalid command
+ """
+ # GIVEN: Test object
log_warning_text = [call('({ip}) get_data(): Invalid packet - '
'unknown command "UNKN"'.format(ip=self.pjlink.name)),
call('({ip}) _send_command(): Nothing to send - '
'returning'.format(ip=self.pjlink.name))]
- log_debug_text = [call('(___TEST_ONE___) get_data(buffer="%1UNKN=Huh?"'),
- call('(___TEST_ONE___) get_data(): Checking new data "%1UNKN=Huh?"'),
- call('(___TEST_ONE___) get_data() header="%1UNKN" data="Huh?"'),
- call('(___TEST_ONE___) get_data() version="1" cmd="UNKN"'),
- call('(___TEST_ONE___) Cleaning buffer - msg = "get_data(): '
- 'Invalid packet - unknown command "UNKN""'),
- call('(___TEST_ONE___) Finished cleaning buffer - 0 bytes dropped')]
+ log_debug_text = [call('({ip}) get_data(buffer="{pre}1UNKN=Huh?"'.format(ip=self.pjlink.name,
+ pre=PJLINK_PREFIX)),
+ call('({ip}) get_data(): Checking new data "{pre}1UNKN=Huh?"'.format(ip=self.pjlink.name,
+ pre=PJLINK_PREFIX)),
+ call('({ip}) get_data() header="{pre}1UNKN" data="Huh?"'.format(ip=self.pjlink.name,
+ pre=PJLINK_PREFIX)),
+ call('({ip}) get_data() version="1" cmd="UNKN" data="Huh?"'.format(ip=self.pjlink.name)),
+ call('({ip}) Cleaning buffer - msg = "get_data(): Invalid packet - '
+ 'unknown command "UNKN""'.format(ip=self.pjlink.name)),
+ call('({ip}) Finished cleaning buffer - 0 bytes dropped'.format(ip=self.pjlink.name))]
+
# WHEN: get_data called with an unknown command
self.pjlink.get_data(buff='{prefix}1UNKN=Huh?'.format(prefix=PJLINK_PREFIX))
# THEN: Appropriate log entries should have been made and methods called/not called
mock_log.warning.assert_has_calls(log_warning_text)
mock_log.debug.assert_has_calls(log_debug_text)
- assert self.pjlink.pjlink_functions.called is False, 'Should not have accessed pjlink_functions'
-
- @skip('Needs update to new setup')
- @patch("openlp.core.projectors.pjlink.log")
- def test_process_command_call_clss(self, mock_log):
- """
- Test process_command calls proper function
- """
- # GIVEN: Test object and mocks
- with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \
- patch.object(openlp.core.projectors.pjlink.PJLink, 'process_clss') as mock_process_clss:
-
- log_debug_calls = [call('({ip}) Processing command "CLSS" with data "1"'.format(ip=self.pjlink.name)),
- call('({ip}) Calling function for CLSS'.format(ip=self.pjlink.name))]
-
- # WHEN: process_command is called with valid function and data
- process_command(projector=self.pjlink, cmd='CLSS', data='1')
-
- # THEN: Appropriate log entries should have been made and methods called
- mock_log.debug.assert_has_calls(log_debug_calls)
- mock_process_clss.assert_called_once_with(data='1')
-
- @skip('Needs update to new setup')
- def test_process_command_erra(self):
- """
- Test ERRA - Authentication Error
+ assert (mock_process_cmd.call_count == 0), 'process_command should not have been called'
+
+ @patch.object(openlp.core.projectors.pjlinkcommands, 'process_command')
+ @patch.object(openlp.core.projectors.pjlink, 'log')
+ def test_projector_get_data_version_mismatch(self, mock_log, mock_process_cmd):
+ """
+ Test projector received valid command with command version higher than projector
"""
# GIVEN: Test object
- with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \
- patch.object(openlp.core.projectors.pjlink.PJLink, 'process_pjlink') as mock_process_pjlink, \
- patch.object(openlp.core.projectors.pjlink.PJLink, 'change_status') as mock_change_status, \
- patch.object(openlp.core.projectors.pjlink.PJLink, 'disconnect_from_host') as mock_disconnect, \
- patch.object(openlp.core.projectors.pjlink.PJLink, 'projectorAuthentication') as mock_authentication:
-
- pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
- log_error_calls = [call('({ip}) PJLINK: {msg}'.format(ip=self.pjlink.name,
- msg=STATUS_MSG[E_AUTHENTICATION]))]
- log_debug_calls = [call('({ip}) Processing command "PJLINK" with data "ERRA"'.format(ip=self.pjlink.name))]
-
- # WHEN: process_command called with ERRA
- pjlink.process_command(cmd='PJLINK', data=PJLINK_ERRORS[E_AUTHENTICATION])
-
- # THEN: Appropriate log entries should have been made and methods called/not called
- assert mock_disconnect.called is True, 'disconnect_from_host should have been called'
- mock_log.error.assert_has_calls(log_error_calls)
- mock_log.debug.assert_has_calls(log_debug_calls)
- mock_change_status.assert_called_once_with(status=E_AUTHENTICATION)
- mock_authentication.emit.assert_called_once_with(pjlink.name)
- mock_process_pjlink.assert_not_called()
+ log_warning_text = [call('({ip}) get_data(): Projector returned class reply higher than projector '
+ 'stated class'.format(ip=self.pjlink.name)),
+ call('({ip}) _send_command(): Nothing to send - returning'.format(ip=self.pjlink.name))]
+
+ log_debug_text = [call('({ip}) get_data(buffer="{pre}2ACKN=X"'.format(ip=self.pjlink.name,
+ pre=PJLINK_PREFIX)),
+ call('({ip}) get_data(): Checking new data "{pre}2ACKN=X"'.format(ip=self.pjlink.name,
+ pre=PJLINK_PREFIX)),
+ call('({ip}) get_data() header="{pre}2ACKN" data="X"'.format(ip=self.pjlink.name,
+ pre=PJLINK_PREFIX)),
+ call('({ip}) get_data() version="2" cmd="ACKN" data="X"'.format(ip=self.pjlink.name))]
+ self.pjlink.pjlink_class = '1'
+
+ # WHEN: get_data called with an unknown command
+ self.pjlink.get_data(buff='{prefix}2ACKN=X'.format(prefix=PJLINK_PREFIX))
+
+ # THEN: Appropriate log entries should have been made and methods called/not called
+ mock_log.warning.assert_has_calls(log_warning_text)
+ mock_log.debug.assert_has_calls(log_debug_text)
+ assert (mock_process_cmd.call_count == 0), 'process_command should not have been called'
@skip('Needs update to new setup')
- def test_process_command_err1(self):
+ def test_routing_err1(self):
"""
Test ERR1 - Undefined projector function
"""
@@ -148,7 +158,7 @@
mock_process_clss.assert_called_once_with(data=PJLINK_ERRORS[E_UNDEFINED])
@skip('Needs update to new setup')
- def test_process_command_err2(self):
+ def test_routing_err2(self):
"""
Test ERR2 - Parameter Error
"""
@@ -170,7 +180,7 @@
mock_process_clss.assert_called_once_with(data=PJLINK_ERRORS[E_PARAMETER])
@skip('Needs update to new setup')
- def test_process_command_err3(self):
+ def test_routing_err3(self):
"""
Test ERR3 - Unavailable error
"""
@@ -192,7 +202,7 @@
mock_process_clss.assert_called_once_with(data=PJLINK_ERRORS[E_UNAVAILABLE])
@skip('Needs update to new setup')
- def test_process_command_err4(self):
+ def test_routing_err4(self):
"""
Test ERR3 - Unavailable error
"""
@@ -214,48 +224,29 @@
mock_process_clss.assert_called_once_with(data=PJLINK_ERRORS[E_PROJECTOR])
@skip('Needs update to new setup')
- def test_process_command_future(self):
+ def test_routing_erra(self):
"""
- Test command valid but no method to process yet
+ Test ERRA - Authentication Error
"""
# GIVEN: Test object
with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \
- patch.object(openlp.core.projectors.pjlink.PJLink, 'process_clss') as mock_process_clss:
+ patch.object(openlp.core.projectors.pjlink.PJLink, 'process_pjlink') as mock_process_pjlink, \
+ patch.object(openlp.core.projectors.pjlink.PJLink, 'change_status') as mock_change_status, \
+ patch.object(openlp.core.projectors.pjlink.PJLink, 'disconnect_from_host') as mock_disconnect, \
+ patch.object(openlp.core.projectors.pjlink.PJLink, 'projectorAuthentication') as mock_authentication:
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
- pjlink.pjlink_functions = MagicMock()
- log_warning_text = [call('({ip}) Unable to process command="CLSS" '
- '(Future option?)'.format(ip=self.pjlink.name))]
- log_debug_text = [call('({ip}) Processing command "CLSS" '
- 'with data "Huh?"'.format(ip=self.pjlink.name))]
+ log_error_calls = [call('({ip}) PJLINK: {msg}'.format(ip=self.pjlink.name,
+ msg=STATUS_MSG[E_AUTHENTICATION]))]
+ log_debug_calls = [call('({ip}) Processing command "PJLINK" with data "ERRA"'.format(ip=self.pjlink.name))]
- # WHEN: Processing a possible future command
- pjlink.process_command(cmd='CLSS', data="Huh?")
+ # WHEN: process_command called with ERRA
+ pjlink.process_command(cmd='PJLINK', data=PJLINK_ERRORS[E_AUTHENTICATION])
# THEN: Appropriate log entries should have been made and methods called/not called
- mock_log.debug.assert_has_calls(log_debug_text)
- mock_log.warning.assert_has_calls(log_warning_text)
- assert pjlink.pjlink_functions.called is False, 'Should not have accessed pjlink_functions'
- assert mock_process_clss.called is False, 'Should not have called process_clss'
-
- @skip('Needs update to new setup')
- def test_process_command_ok(self):
- """
- Test command returned success
- """
- # GIVEN: Test object and mocks
- with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \
- patch.object(openlp.core.projectors.pjlink.PJLink, 'send_command') as mock_send_command, \
- patch.object(openlp.core.projectors.pjlink.PJLink, 'process_clss') as mock_process_clss:
-
- pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
- log_debug_calls = [call('({ip}) Processing command "CLSS" with data "OK"'.format(ip=self.pjlink.name)),
- call('({ip}) Command "CLSS" returned OK'.format(ip=self.pjlink.name))]
-
- # WHEN: process_command is called with valid function and data
- pjlink.process_command(cmd='CLSS', data='OK')
-
- # THEN: Appropriate log entries should have been made and methods called
+ assert mock_disconnect.called is True, 'disconnect_from_host should have been called'
+ mock_log.error.assert_has_calls(log_error_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
- mock_send_command.assert_called_once_with(cmd='CLSS')
- mock_process_clss.assert_not_called()
+ mock_change_status.assert_called_once_with(status=E_AUTHENTICATION)
+ mock_authentication.emit.assert_called_once_with(pjlink.name)
+ mock_process_pjlink.assert_not_called()
Follow ups
