Update pjlinkcommands tests

This commit is contained in:
Ken Roberts 2022-02-01 16:43:09 +00:00 committed by Tim Bentley
parent ff0605cb21
commit e02b359b7e
6 changed files with 287 additions and 292 deletions

View File

@ -73,7 +73,7 @@ SocketSTate = QtNetwork.QAbstractSocket.SocketState
# Add prefix here, but defer linkclass expansion until later when we have the actual # Add prefix here, but defer linkclass expansion until later when we have the actual
# PJLink class for the command # PJLink class for the command
PJLINK_HEADER = '{prefix}{{linkclass}}'.format(prefix=PJLINK_PREFIX) PJLINK_HEADER = f'{PJLINK_PREFIX}{{linkclass}}'
class PJLinkUDP(QtNetwork.QUdpSocket): class PJLinkUDP(QtNetwork.QUdpSocket):

View File

@ -3,7 +3,7 @@
########################################################################## ##########################################################################
# OpenLP - Open Source Lyrics Projection # # OpenLP - Open Source Lyrics Projection #
# ---------------------------------------------------------------------- # # ---------------------------------------------------------------------- #
# Copyright (c) 2008-2021 OpenLP Developers # # Copyright (c) 2008-2022 OpenLP Developers #
# ---------------------------------------------------------------------- # # ---------------------------------------------------------------------- #
# This program is free software: you can redistribute it and/or modify # # 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 # # it under the terms of the GNU General Public License as published by #
@ -24,6 +24,22 @@ processing projector replies.
NOTE: PJLink Class (version) checks are handled in the respective PJLink/PJLinkUDP classes. NOTE: PJLink Class (version) checks are handled in the respective PJLink/PJLinkUDP classes.
process_clss is the only exception. process_clss is the only exception.
NOTE: Some commands are both commannd replies as well as UDP terminal-initiated status
messages.
Ex: POWR
CLSS1 (TCP): controller sends "POWR x", projector replies "POWR=xxxxx"
CLSS2 (UDP): projector sends "POWER=xxxx"
Inn both instances, the messagege is processed the same.
For CLSS1, we initiate communication, so we know which projecttor instance
the message is routed to.
For CLSS2, the terminal initiates communication, so as part of the UDP process
we must find the projector that initiated the message.
""" """
import logging import logging
@ -51,31 +67,26 @@ def process_command(projector, cmd, data):
:param cmd: Command to process :param cmd: Command to process
:param data: Data being processed :param data: Data being processed
""" """
log.debug('({ip}) Processing command "{cmd}" with data "{data}"'.format(ip=projector.entry.name, log.debug(f'({projector.entry.name}) Processing command "{cmd}" with data "{data}"')
cmd=cmd,
data=data))
# cmd should already be in uppercase, but data may be in mixed-case. # cmd should already be in uppercase, but data may be in mixed-case.
# Due to some replies should stay as mixed-case, validate using separate uppercase check # Due to some replies should stay as mixed-case, validate using separate uppercase check
_data = data.upper() _data = data.upper()
# Check if we have a future command not available yet # Check if we have a future command not available yet
if cmd not in pjlink_functions: if cmd not in pjlink_functions:
log.warning('({ip}) Unable to process command="{cmd}" (Future option?)'.format(ip=projector.entry.name, log.warning(f'({projector.entry.name}) Unable to process command="{cmd}" (Future option?)')
cmd=cmd))
return return
elif _data == 'OK': elif _data == 'OK':
log.debug('({ip}) Command "{cmd}" returned OK'.format(ip=projector.entry.name, cmd=cmd)) log.debug(f'({projector.entry.name}) Command "{cmd}" returned OK')
# A command returned successfully, so do a query on command to verify status # A command returned successfully, so do a query on command to verify status
return S_DATA_OK return S_DATA_OK
elif _data in PJLINK_ERRORS: elif _data in PJLINK_ERRORS:
# Oops - projector error # Oops - projector error
log.error('({ip}) {cmd}: {err}'.format(ip=projector.entry.name, log.error(f'({projector.entry.name}) {cmd}: {STATUS_MSG[PJLINK_ERRORS[_data]]}')
cmd=cmd,
err=STATUS_MSG[PJLINK_ERRORS[_data]]))
return PJLINK_ERRORS[_data] return PJLINK_ERRORS[_data]
# Command checks already passed # Command checks already passed
log.debug('({ip}) Calling function for {cmd}'.format(ip=projector.entry.name, cmd=cmd)) log.debug(f'({projector.entry.name}) Calling function for {cmd}')
return pjlink_functions[cmd](projector=projector, data=data) return pjlink_functions[cmd](projector=projector, data=data)
@ -83,6 +94,8 @@ def process_ackn(projector, data):
""" """
Process the ACKN command. Process the ACKN command.
UDP reply to SRCH command
:param projector: Projector instance :param projector: Projector instance
:param data: Data in packet :param data: Data in packet
""" """
@ -339,7 +352,7 @@ def process_lamp(projector, data):
def process_lkup(projector, data): def process_lkup(projector, data):
""" """
Process reply indicating remote is available for connection Process UDP request indicating remote is available for connection
:param projector: Projector instance :param projector: Projector instance
:param data: Data packet from remote :param data: Data packet from remote
@ -496,6 +509,8 @@ def process_srch(projector=None, data=None):
SRCH is processed by terminals so we ignore any packet. SRCH is processed by terminals so we ignore any packet.
UDP command to find active CLSS 2 projectors. Reply is ACKN.
:param projector: Projector instance (actually ignored for this command) :param projector: Projector instance (actually ignored for this command)
:param data: Data in packet :param data: Data in packet
""" """
@ -541,7 +556,7 @@ pjlink_functions = {
'INPT': process_inpt, 'INPT': process_inpt,
'INST': process_inst, 'INST': process_inst,
'LAMP': process_lamp, 'LAMP': process_lamp,
'LKUP': process_lkup, # Class 2 (reply only - no cmd) 'LKUP': process_lkup, # Class 2 (terminal request only - no cmd)
'NAME': process_name, 'NAME': process_name,
'PJLINK': process_pjlink, 'PJLINK': process_pjlink,
'POWR': process_powr, 'POWR': process_powr,

View File

@ -0,0 +1,23 @@
# -*- coding: utf-8 -*-
##########################################################################
# OpenLP - Open Source Lyrics Projection #
# ---------------------------------------------------------------------- #
# Copyright (c) 2008-2022 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/>. #
##########################################################################
"""
:mod tests/openlp_core/projectors/commands: Tests for PJLink commands
"""

View File

@ -0,0 +1,235 @@
# -*- coding: utf-8 -*-
##########################################################################
# OpenLP - Open Source Lyrics Projection #
# ---------------------------------------------------------------------- #
# Copyright (c) 2008-2021 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/>. #
##########################################################################
"""
Tests for PJLink command routing
"""
import logging
import openlp.core.projectors.pjlinkcommands
from unittest.mock import MagicMock, patch
from openlp.core.projectors.pjlinkcommands import process_command
from openlp.core.projectors.constants import PJLINK_ERRORS, STATUS_CODE, \
E_AUTHENTICATION, E_UNDEFINED, E_PARAMETER, E_UNAVAILABLE, E_PROJECTOR, \
S_DATA_OK, STATUS_MSG
test_module = openlp.core.projectors.pjlinkcommands.__name__
def test_valid_command(pjlink, caplog):
"""
Test valid command routing
"""
# GIVEN: Test setup
caplog.set_level(logging.DEBUG)
test_cmd = "CLSS"
test_data = "1"
test_return = None
logs = [(f'{test_module}', logging.DEBUG,
f'({pjlink.name}) Processing command "{test_cmd}" with data "{test_data}"'),
(f'{test_module}', logging.DEBUG,
f'({pjlink.name}) Calling function for {test_cmd}')
]
caplog.clear()
with patch.dict(openlp.core.projectors.pjlinkcommands.pjlink_functions,
{test_cmd: MagicMock(return_value=None)}) as mock_clss:
# WHEN: called with CLSS command
chk = process_command(projector=pjlink, cmd=test_cmd, data=test_data)
# THEN: Appropriate log entries should have been made and methods called/not called
assert chk is test_return, f'"{test_data}" reply should have returned {STATUS_CODE[test_return]}'
mock_clss[test_cmd].assert_called_with(projector=pjlink, data=test_data)
assert caplog.record_tuples == logs, 'Log entries mismatch'
def test_invalid_command(pjlink, caplog):
"""
Test invalid command
"""
# GIVEN: Test setup
caplog.set_level(logging.DEBUG)
test_cmd = "DONTCARE"
test_data = "1"
test_return = None
logs = [(f'{test_module}', logging.DEBUG,
f'({pjlink.name}) Processing command "{test_cmd}" with data "{test_data}"'),
(f'{test_module}', logging.WARNING,
f'({pjlink.name}) Unable to process command="DONTCARE" (Future option?)'),
]
# WHEN: called with invalid command and data
caplog.clear()
chk = process_command(projector=pjlink, cmd=test_cmd, data=test_data)
# THEN: Appropriate log entries should have been made and methods called/not called
assert chk is test_return, 'Invalid command should have returned None'
assert caplog.record_tuples == logs, 'Log entries mismatch'
def test_data_ok(pjlink, caplog):
"""
Test data == 'OK'
"""
# GIVEN: Test setup
caplog.set_level(logging.DEBUG)
test_cmd = "CLSS"
test_data = "OK"
test_reply = "OK"
test_return = S_DATA_OK
logs = [(f'{test_module}', logging.DEBUG,
f'({pjlink.name}) Processing command "{test_cmd}" with data "{test_data}"'),
(f'{test_module}', logging.DEBUG,
f'({pjlink.name}) Command "{test_cmd}" returned {test_reply}')
]
# WHEN: called with OK
caplog.clear()
chk = process_command(projector=pjlink, cmd=test_cmd, data=test_data)
# THEN: Appropriate log entries should have been made and methods called/not called
assert chk is test_return, f'"{test_data}" reply should have returned {STATUS_CODE[test_return]}'
assert caplog.record_tuples == logs, 'Log entries mismatch'
def test_e_authentication(pjlink, caplog):
"""
Test data == ERRA
"""
# GIVEN: Test setup
caplog.set_level(logging.DEBUG)
test_error = E_AUTHENTICATION
test_cmd = 'CLSS'
test_data = PJLINK_ERRORS[test_error]
test_reply = STATUS_MSG[test_error]
test_return = test_error
logs = [(f'{test_module}', logging.DEBUG,
f'({pjlink.name}) Processing command "{test_cmd}" with data "{test_data}"'),
(f'{test_module}', logging.ERROR,
f'({pjlink.name}) {test_cmd}: {test_reply}')
]
# WHEN: called with error
caplog.clear()
chk = process_command(projector=pjlink, cmd=test_cmd, data=test_data)
# THEN: Appropriate log entries should have been made and methods called/not called
assert chk is test_return, f'"{PJLINK_ERRORS[test_error]}" reply should have returned {STATUS_CODE[test_error]}'
assert caplog.record_tuples == logs, 'Log entries mismatch'
def test_e_undefined(pjlink, caplog):
"""
Test data == ERR1
"""
# GIVEN: Test setup
caplog.set_level(logging.DEBUG)
test_error = E_UNDEFINED
test_cmd = 'CLSS'
test_data = PJLINK_ERRORS[test_error]
test_reply = STATUS_MSG[test_error]
test_return = test_error
logs = [(f'{test_module}', logging.DEBUG,
f'({pjlink.name}) Processing command "{test_cmd}" with data "{test_data}"'),
(f'{test_module}', logging.ERROR,
f'({pjlink.name}) {test_cmd}: {test_reply}')
]
# WHEN: called with error
caplog.clear()
chk = process_command(projector=pjlink, cmd=test_cmd, data=test_data)
# THEN: Appropriate log entries should have been made and methods called/not called
assert chk is test_return, f'"{PJLINK_ERRORS[test_error]}" reply should have returned {STATUS_CODE[test_error]}'
assert caplog.record_tuples == logs, 'Log entries mismatch'
def test_e_parameter(pjlink, caplog):
"""
Test data == ERR2
"""
# GIVEN: Test setup
caplog.set_level(logging.DEBUG)
test_error = E_PARAMETER
test_cmd = 'CLSS'
test_data = PJLINK_ERRORS[test_error]
test_reply = STATUS_MSG[test_error]
test_return = test_error
logs = [(f'{test_module}', logging.DEBUG,
f'({pjlink.name}) Processing command "{test_cmd}" with data "{test_data}"'),
(f'{test_module}', logging.ERROR,
f'({pjlink.name}) {test_cmd}: {test_reply}')
]
# WHEN: called with error
caplog.clear()
chk = process_command(projector=pjlink, cmd=test_cmd, data=test_data)
# THEN: Appropriate log entries should have been made and methods called/not called
assert chk is test_return, f'"{PJLINK_ERRORS[test_error]}" reply should have returned {STATUS_CODE[test_error]}'
assert caplog.record_tuples == logs, 'Log entries mismatch'
def test_e_unavailable(pjlink, caplog):
"""
Test data == ERR3
"""
# GIVEN: Test setup
caplog.set_level(logging.DEBUG)
test_error = E_UNAVAILABLE
test_cmd = 'CLSS'
test_data = PJLINK_ERRORS[test_error]
test_reply = STATUS_MSG[test_error]
test_return = test_error
logs = [(f'{test_module}', logging.DEBUG,
f'({pjlink.name}) Processing command "{test_cmd}" with data "{test_data}"'),
(f'{test_module}', logging.ERROR,
f'({pjlink.name}) {test_cmd}: {test_reply}')
]
# WHEN: called with error
caplog.clear()
chk = process_command(projector=pjlink, cmd=test_cmd, data=test_data)
# THEN: Appropriate log entries should have been made and methods called/not called
assert chk is test_return, f'"{PJLINK_ERRORS[test_error]}" reply should have returned {STATUS_CODE[test_error]}'
assert caplog.record_tuples == logs, 'Log entries mismatch'
def test_e_projector(pjlink, caplog):
"""
Test data == ERR4
"""
# GIVEN: Test setup
caplog.set_level(logging.DEBUG)
test_error = E_PROJECTOR
test_cmd = 'CLSS'
test_data = PJLINK_ERRORS[test_error]
test_reply = STATUS_MSG[test_error]
test_return = test_error
logs = [(f'{test_module}', logging.DEBUG,
f'({pjlink.name}) Processing command "{test_cmd}" with data "{test_data}"'),
(f'{test_module}', logging.ERROR,
f'({pjlink.name}) {test_cmd}: {test_reply}')
]
# WHEN: called with error
caplog.clear()
chk = process_command(projector=pjlink, cmd=test_cmd, data=test_data)
# THEN: Appropriate log entries should have been made and methods called/not called
assert chk is test_return, f'"{PJLINK_ERRORS[test_error]}" reply should have returned {STATUS_CODE[test_error]}'
assert caplog.record_tuples == logs, 'Log entries mismatch'

View File

@ -1,68 +0,0 @@
# -*- coding: utf-8 -*-
##########################################################################
# OpenLP - Open Source Lyrics Projection #
# ---------------------------------------------------------------------- #
# Copyright (c) 2008-2021 OpenLP Developers #
# ---------------------------------------------------------------------- #
# This program is free software: you can redistribute it and/or modify #
# it under the terms of the GNU General Public License as published by #
# the Free Software Foundation, either version 3 of the License, or #
# (at your option) any later version. #
# #
# This program is distributed in the hope that it will be useful, #
# but WITHOUT ANY WARRANTY; without even the implied warranty of #
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the #
# GNU General Public License for more details. #
# #
# You should have received a copy of the GNU General Public License #
# along with this program. If not, see <https://www.gnu.org/licenses/>. #
##########################################################################
"""
Package to test the openlp.core.projectors.pjlink base package.
"""
def test_bug_1550891_process_clss_nonstandard_reply_1():
"""
Bugfix 1550891: CLSS request returns non-standard reply with Optoma/Viewsonic projector
"""
# Test now part of test_projector_pjlink_commands_01
# Keeping here for bug reference
pass
def test_bug_1550891_process_clss_nonstandard_reply_2():
"""
Bugfix 1550891: CLSS request returns non-standard reply with BenQ projector
"""
# Test now part of test_projector_pjlink_commands_01
# Keeping here for bug reference
pass
def test_bug_1593882_no_pin_authenticated_connection():
"""
Test bug 1593882 no pin and authenticated request exception
"""
# Test now part of test_projector_pjlink_commands_02
# Keeping here for bug reference
pass
def test_bug_1593883_pjlink_authentication():
"""
Test bugfix 1593883 pjlink authentication and ticket 92187
"""
# Test now part of test_projector_pjlink_commands_02
# Keeping here for bug reference
pass
def test_bug_1734275_process_lamp_nonstandard_reply():
"""
Test bugfix 17342785 non-standard LAMP response with one lamp hours only
"""
# Test now part of test_projector_pjlink_commands_01
# Keeping here for bug reference
pass

View File

@ -1,210 +0,0 @@
# -*- coding: utf-8 -*-
##########################################################################
# OpenLP - Open Source Lyrics Projection #
# ---------------------------------------------------------------------- #
# Copyright (c) 2008-2021 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.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, \
S_DATA_OK, PJLINK_ERRORS, STATUS_MSG
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_routing_command(mock_log, pjlink):
"""
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=pjlink.name)),
call('({ip}) Calling function for CLSS'.format(ip=pjlink.name)),
call('({ip}) Setting pjlink_class for this projector to "1"'.format(ip=pjlink.name))]
# WHEN: called with valid command and data
chk = process_command(projector=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(mock_log, mock_functions, mock_clss, pjlink):
"""
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=pjlink.name))]
log_debug_text = [call('({ip}) Processing command "CLSS" with data "?"'.format(ip=pjlink.name))]
# Fake CLSS command is not in list
mock_functions.__contains__.return_value = False
# WHEN: called with unknown command
process_command(projector=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(mock_log, mock_clss, mock_send, pjlink):
"""
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=pjlink.name)),
call('({ip}) Command "CLSS" returned OK'.format(ip=pjlink.name))
]
# WHEN: Command called with OK
chk = process_command(projector=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_err1(mock_log, pjlink):
"""
Test rouing when PJLink ERR1 received
"""
# GIVEN: Test setup
err_code = E_UNDEFINED
err_msg = STATUS_MSG[err_code]
err_str = PJLINK_ERRORS[err_code]
log_error_text = [call('({ip}) CLSS: {err}'.format(ip=pjlink.name, err=err_msg))]
log_debug_text = [call('({ip}) Processing command "CLSS" with data "{err}"'.format(ip=pjlink.name,
err=err_str))]
# WHEN: routing called
chk = process_command(projector=pjlink, cmd='CLSS', data=err_str)
# 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 {err}'.format(err=PJLINK_ERRORS[err_code])
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_routing_pjink_err2(mock_log, pjlink):
"""
Test rouing when PJLink ERR2 received
"""
# GIVEN: Test setup
err_code = E_PARAMETER
err_msg = STATUS_MSG[err_code]
err_str = PJLINK_ERRORS[err_code]
log_error_text = [call('({ip}) CLSS: {err}'.format(ip=pjlink.name, err=err_msg))]
log_debug_text = [call('({ip}) Processing command "CLSS" with data "{err}"'.format(ip=pjlink.name,
err=err_str))]
# WHEN: routing called
chk = process_command(projector=pjlink, cmd='CLSS', data=err_str)
# 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 {err}'.format(err=PJLINK_ERRORS[err_code])
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_routing_pjink_err3(mock_log, pjlink):
"""
Test rouing when PJLink ERR3 received
"""
# GIVEN: Test setup
err_code = E_UNAVAILABLE
err_msg = STATUS_MSG[err_code]
err_str = PJLINK_ERRORS[err_code]
log_error_text = [call('({ip}) CLSS: {err}'.format(ip=pjlink.name, err=err_msg))]
log_debug_text = [call('({ip}) Processing command "CLSS" with data "{err}"'.format(ip=pjlink.name,
err=err_str))]
# WHEN: routing called
chk = process_command(projector=pjlink, cmd='CLSS', data=err_str)
# 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 {err}'.format(err=PJLINK_ERRORS[err_code])
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_routing_pjink_err4(mock_log, pjlink):
"""
Test rouing when PJLink ERR4 received
"""
# GIVEN: Test setup
err_code = E_PROJECTOR
err_msg = STATUS_MSG[err_code]
err_str = PJLINK_ERRORS[err_code]
log_error_text = [call('({ip}) CLSS: {err}'.format(ip=pjlink.name, err=err_msg))]
log_debug_text = [call('({ip}) Processing command "CLSS" with data "{err}"'.format(ip=pjlink.name,
err=err_str))]
# WHEN: routing called
chk = process_command(projector=pjlink, cmd='CLSS', data=err_str)
# 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 {err}'.format(err=PJLINK_ERRORS[err_code])
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_routing_pjink_errA(mock_log, pjlink):
"""
Test rouing when PJLink ERRA received
"""
# GIVEN: Test setup
err_code = E_AUTHENTICATION
err_msg = STATUS_MSG[err_code]
err_str = PJLINK_ERRORS[err_code]
log_error_text = [call('({ip}) CLSS: {err}'.format(ip=pjlink.name, err=err_msg))]
log_debug_text = [call('({ip}) Processing command "CLSS" with data "{err}"'.format(ip=pjlink.name,
err=err_str))]
# WHEN: routing called
chk = process_command(projector=pjlink, cmd='CLSS', data=err_str)
# 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 {err}'.format(err=PJLINK_ERRORS[err_code])