forked from openlp/openlp
PJLink2 update V04
bzr-revno: 2865
This commit is contained in:
Tim Bentley
commit
3b179c74a0
@ -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'),
|
||||
|
||||
@ -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,
|
||||
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'))
|
||||
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'])
|
||||
|
||||
@ -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
132
tests/openlp_core/projectors/test_projector_command_routing.py
Normal 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'
|
||||
@ -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)
|
||||
|
||||
@ -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]
|
||||
|
||||
|
||||
@ -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)),
|
||||
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()
|
||||
|
||||
Loading…
Reference in New Issue
Block a user