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

Head

Browse Source
This commit is contained in:
Tim Bentley 2019-05-04 20:29:20 +01:00
commit 3526c3de64
7 changed files with 373 additions and 223 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -106,6 +106,9 @@ S_WARMUP = 314
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 @@ STATUS_CODE = {
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 @@ STATUS_CODE = {
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 @@ STATUS_MSG = {
'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'),

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

@ -52,14 +52,15 @@ from copy import copy
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 @@ class PJLink(QtNetwork.QTcpSocket):
# 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 @@ class PJLink(QtNetwork.QTcpSocket):
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 @@ class PJLink(QtNetwork.QTcpSocket):
: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 @@ class PJLink(QtNetwork.QTcpSocket):
# 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'])

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

@ -30,14 +30,12 @@ NOTE: PJLink Class (version) checks are handled in the respective PJLink/PJLinkU
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 @@ def process_command(projector, cmd, data):
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 @@ def process_pjlink(projector, data):
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):

132
tests/openlp_core/projectors/test_projector_command_routing.py Normal file
View File

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

3
tests/openlp_core/projectors/test_projector_constants.py
View File

@ -40,11 +40,12 @@ class TestProjectorConstants(TestCase):
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)

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

@ -454,9 +454,9 @@ class TestPJLinkBase(TestCase):
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 @@ class TestPJLinkBase(TestCase):
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 @@ class TestPJLinkBase(TestCase):
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 @@ class TestPJLinkBase(TestCase):
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 @@ class TestPJLinkBase(TestCase):
'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 @@ class TestPJLinkBase(TestCase):
'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 @@ class TestPJLinkBase(TestCase):
'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]

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

@ -24,10 +24,9 @@ Package to test the openlp.core.projectors.pjlink command routing.
"""
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,54 +50,181 @@ class TestPJLinkRouting(TestCase):
"""
del(self.pjlink)
@patch.object(openlp.core.projectors.pjlinkcommands, 'process_command')
@patch.object(openlp.core.projectors.pjlink, 'log')
def test_get_data_unknown_command(self, mock_log):
def test_projector_get_data_invalid_version(self, mock_log, mock_process_cmd):
"""
Test not a valid command
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
self.pjlink.pjlink_functions = MagicMock()
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'
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
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')
@patch("openlp.core.projectors.pjlink.log")
def test_process_command_call_clss(self, mock_log):
def test_routing_err1(self):
"""
Test process_command calls proper function
Test ERR1 - Undefined projector function
"""
# GIVEN: Test object and mocks
# 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:
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))]
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
log_error_text = [call('({ip}) CLSS: {msg}'.format(ip=self.pjlink.name, msg=STATUS_MSG[E_UNDEFINED]))]
log_debug_text = [call('({ip}) Processing command "CLSS" with data "ERR1"'.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')
# WHEN: process_command called with ERR1
pjlink.process_command(cmd='CLSS', data=PJLINK_ERRORS[E_UNDEFINED])
# 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')
mock_log.error.assert_has_calls(log_error_text)
mock_log.debug.assert_has_calls(log_debug_text)
mock_process_clss.assert_called_once_with(data=PJLINK_ERRORS[E_UNDEFINED])
@skip('Needs update to new setup')
def test_process_command_erra(self):
def test_routing_err2(self):
"""
Test ERR2 - Parameter 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:
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
log_error_text = [call('({ip}) CLSS: {msg}'.format(ip=self.pjlink.name, msg=STATUS_MSG[E_PARAMETER]))]
log_debug_text = [call('({ip}) Processing command "CLSS" with data "ERR2"'.format(ip=self.pjlink.name)),
call('({ip}) Calling function for CLSS'.format(ip=self.pjlink.name))]
# WHEN: process_command called with ERR2
pjlink.process_command(cmd='CLSS', data=PJLINK_ERRORS[E_PARAMETER])
# 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)
mock_process_clss.assert_called_once_with(data=PJLINK_ERRORS[E_PARAMETER])
@skip('Needs update to new setup')
def test_routing_err3(self):
"""
Test ERR3 - Unavailable 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:
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
log_error_text = [call('({ip}) CLSS: {msg}'.format(ip=self.pjlink.name, msg=STATUS_MSG[E_UNAVAILABLE]))]
log_debug_text = [call('({ip}) Processing command "CLSS" with data "ERR3"'.format(ip=self.pjlink.name)),
call('({ip}) Calling function for CLSS'.format(ip=self.pjlink.name))]
# WHEN: process_command called with ERR3
pjlink.process_command(cmd='CLSS', data=PJLINK_ERRORS[E_UNAVAILABLE])
# THEN: Appropriate log entries should have been made and methods called
mock_log.error.assert_has_calls(log_error_text)
mock_log.debug.assert_has_calls(log_debug_text)
mock_process_clss.assert_called_once_with(data=PJLINK_ERRORS[E_UNAVAILABLE])
@skip('Needs update to new setup')
def test_routing_err4(self):
"""
Test ERR3 - Unavailable 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:
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
log_error_text = [call('({ip}) CLSS: {msg}'.format(ip=self.pjlink.name, msg=STATUS_MSG[E_PROJECTOR]))]
log_debug_text = [call('({ip}) Processing command "CLSS" with data "ERR4"'.format(ip=self.pjlink.name)),
call('({ip}) Calling function for CLSS'.format(ip=self.pjlink.name))]
# WHEN: process_command called with ERR4
pjlink.process_command(cmd='CLSS', data=PJLINK_ERRORS[E_PROJECTOR])
# THEN: Appropriate log entries should have been made and methods called
mock_log.error.assert_has_calls(log_error_text)
mock_log.debug.assert_has_calls(log_debug_text)
mock_process_clss.assert_called_once_with(data=PJLINK_ERRORS[E_PROJECTOR])
@skip('Needs update to new setup')
def test_routing_erra(self):
"""
Test ERRA - Authentication Error
"""
@ -124,138 +250,3 @@ class TestPJLinkRouting(TestCase):
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()
@skip('Needs update to new setup')
def test_process_command_err1(self):
"""
Test ERR1 - Undefined projector function
"""
# 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:
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
log_error_text = [call('({ip}) CLSS: {msg}'.format(ip=self.pjlink.name, msg=STATUS_MSG[E_UNDEFINED]))]
log_debug_text = [call('({ip}) Processing command "CLSS" with data "ERR1"'.format(ip=self.pjlink.name)),
call('({ip}) Calling function for CLSS'.format(ip=self.pjlink.name))]
# WHEN: process_command called with ERR1
pjlink.process_command(cmd='CLSS', data=PJLINK_ERRORS[E_UNDEFINED])
# THEN: Appropriate log entries should have been made and methods called
mock_log.error.assert_has_calls(log_error_text)
mock_log.debug.assert_has_calls(log_debug_text)
mock_process_clss.assert_called_once_with(data=PJLINK_ERRORS[E_UNDEFINED])
@skip('Needs update to new setup')
def test_process_command_err2(self):
"""
Test ERR2 - Parameter 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:
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
log_error_text = [call('({ip}) CLSS: {msg}'.format(ip=self.pjlink.name, msg=STATUS_MSG[E_PARAMETER]))]
log_debug_text = [call('({ip}) Processing command "CLSS" with data "ERR2"'.format(ip=self.pjlink.name)),
call('({ip}) Calling function for CLSS'.format(ip=self.pjlink.name))]
# WHEN: process_command called with ERR2
pjlink.process_command(cmd='CLSS', data=PJLINK_ERRORS[E_PARAMETER])
# 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)
mock_process_clss.assert_called_once_with(data=PJLINK_ERRORS[E_PARAMETER])
@skip('Needs update to new setup')
def test_process_command_err3(self):
"""
Test ERR3 - Unavailable 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:
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
log_error_text = [call('({ip}) CLSS: {msg}'.format(ip=self.pjlink.name, msg=STATUS_MSG[E_UNAVAILABLE]))]
log_debug_text = [call('({ip}) Processing command "CLSS" with data "ERR3"'.format(ip=self.pjlink.name)),
call('({ip}) Calling function for CLSS'.format(ip=self.pjlink.name))]
# WHEN: process_command called with ERR3
pjlink.process_command(cmd='CLSS', data=PJLINK_ERRORS[E_UNAVAILABLE])
# THEN: Appropriate log entries should have been made and methods called
mock_log.error.assert_has_calls(log_error_text)
mock_log.debug.assert_has_calls(log_debug_text)
mock_process_clss.assert_called_once_with(data=PJLINK_ERRORS[E_UNAVAILABLE])
@skip('Needs update to new setup')
def test_process_command_err4(self):
"""
Test ERR3 - Unavailable 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:
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
log_error_text = [call('({ip}) CLSS: {msg}'.format(ip=self.pjlink.name, msg=STATUS_MSG[E_PROJECTOR]))]
log_debug_text = [call('({ip}) Processing command "CLSS" with data "ERR4"'.format(ip=self.pjlink.name)),
call('({ip}) Calling function for CLSS'.format(ip=self.pjlink.name))]
# WHEN: process_command called with ERR4
pjlink.process_command(cmd='CLSS', data=PJLINK_ERRORS[E_PROJECTOR])
# THEN: Appropriate log entries should have been made and methods called
mock_log.error.assert_has_calls(log_error_text)
mock_log.debug.assert_has_calls(log_debug_text)
mock_process_clss.assert_called_once_with(data=PJLINK_ERRORS[E_PROJECTOR])
@skip('Needs update to new setup')
def test_process_command_future(self):
"""
Test command valid but no method to process yet
"""
# 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:
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))]
# WHEN: Processing a possible future command
pjlink.process_command(cmd='CLSS', data="Huh?")
# 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
mock_log.debug.assert_has_calls(log_debug_calls)
mock_send_command.assert_called_once_with(cmd='CLSS')
mock_process_clss.assert_not_called()