openlp
/
openlp
mirror of https://gitlab.com/openlp/openlp.git
1
0
Fork 1
Code Issues Projects Releases Wiki Activity

Merge branch 'test-commands-2' into 'master' 

Update tests for pjlinkcommands 2022-02-03

See merge request openlp/openlp!408
This commit is contained in:
Tim Bentley 2022-02-05 17:13:52 +00:00
parent d31c5979a2 be21ef15e6
commit db64a7d5bf
4 changed files with 291 additions and 137 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -159,31 +159,30 @@ def process_clss(projector, data):
# : Received: '%1CLSS=Class 1' (Optoma)
# : Received: '%1CLSS=Version1' (BenQ)
if len(data) > 1:
log.warning('({ip}) Non-standard CLSS reply: "{data}"'.format(ip=projector.entry.name, data=data))
log.warning(f'({projector.entry.name}) Non-standard CLSS reply: "{data}"')
# Due to stupid projectors not following standards (Optoma, BenQ comes to mind),
# AND the different responses that can be received, the semi-permanent way to
# fix the class reply is to just remove all non-digit characters.
chk = re.findall(r'\d', data)
if len(chk) < 1:
log.warning('({ip}) No numbers found in class version reply "{data}" - '
'defaulting to class "1"'.format(ip=projector.entry.name, data=data))
log.warning(f'({projector.entry.name}) No numbers found in class version reply '
f'"{data}" - defaulting to class "1"')
clss = '1'
else:
clss = chk[0] # Should only be the first match
elif not data.isdigit():
log.warning('({ip}) NAN CLSS version reply "{data}" - '
'defaulting to class "1"'.format(ip=projector.entry.name, data=data))
log.warning(f'({projector.entry.name}) NAN CLSS version reply '
f'"{data}" - defaulting to class "1"')
clss = '1'
else:
clss = data
projector.pjlink_class = clss
log.debug('({ip}) Setting pjlink_class for this projector to "{data}"'.format(ip=projector.entry.name,
data=projector.pjlink_class))
log.debug(f'({projector.entry.name}) Setting pjlink_class for this projector to "{projector.pjlink_class}"')
if projector.no_poll:
return
# Since we call this one on first connect, setup polling from here
log.debug('({ip}) process_pjlink(): Starting timer'.format(ip=projector.entry.name))
log.debug(f'({projector.entry.name}) process_pjlink(): Starting timer')
projector.poll_timer.setInterval(1000) # Set 1 second for initial information
projector.poll_timer.start()
return
@ -514,7 +513,10 @@ def process_srch(projector=None, data=None):
:param projector: Projector instance (actually ignored for this command)
:param data: Data in packet
"""
log.warning("({ip}) SRCH packet detected - ignoring".format(ip=projector.entry.ip))
if projector is None:
log.warning('SRCH packet detected - ignoring')
else:
log.warning(f'({projector.entry.name}) SRCH packet detected - ignoring')
return

216
tests/openlp_core/projectors/messages/test_clss.py Normal file
View File

@ -0,0 +1,216 @@
# -*- 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/>. #
##########################################################################
"""
Tests for PJLink CLSS command
"""
import logging
import openlp.core.projectors.pjlinkcommands
from openlp.core.projectors.pjlinkcommands import process_clss
from unittest.mock import patch
test_module = openlp.core.projectors.pjlinkcommands.__name__
def test_reply_long_no_number(pjlink, caplog):
"""
Tests reply data length too long
"""
# GIVEN: Test setup
caplog.set_level(logging.DEBUG)
t_data = 'Class A'
logs = [(f'{test_module}', logging.WARNING,
f'({pjlink.entry.name}) Non-standard CLSS reply: "{t_data}"'),
(f'{test_module}', logging.WARNING,
f'({pjlink.entry.name}) No numbers found in class version reply '
f'"{t_data}" - defaulting to class "1"'),
(f'{test_module}', logging.DEBUG,
f'({pjlink.entry.name}) Setting pjlink_class for this projector to "1"')
]
with patch.object(pjlink, 'poll_timer') as mock_timer:
# WHEN: process_clss called
caplog.clear()
t_chk = process_clss(projector=pjlink, data=t_data)
# THEN: Log entries and settings apply
assert t_chk is None, f'Invalid return code {t_chk}'
assert caplog.record_tuples == logs, 'Invalid log entries'
assert pjlink.pjlink_class == '1', 'Should have set pjlink_class = "1"'
mock_timer.setInterval.assert_not_called()
mock_timer.start.assert_not_called()
def test_reply_long_optoma(pjlink, caplog):
"""
Tests invalid reply from Optoma projectors
"""
# GIVEN: Test setup
caplog.set_level(logging.DEBUG)
t_data = 'Class 1'
logs = [(f'{test_module}', logging.WARNING,
f'({pjlink.entry.name}) Non-standard CLSS reply: "{t_data}"'),
(f'{test_module}', logging.DEBUG,
f'({pjlink.entry.name}) Setting pjlink_class for this projector to "1"')
]
with patch.object(pjlink, 'poll_timer') as mock_timer:
# WHEN: process_clss called
caplog.clear()
t_chk = process_clss(projector=pjlink, data=t_data)
# THEN: Log entries and settings apply
assert t_chk is None, f'Invalid return code {t_chk}'
assert caplog.record_tuples == logs, 'Invalid log entries'
assert pjlink.pjlink_class == '1', 'Should have set pjlink_class = "1"'
mock_timer.setInterval.assert_not_called()
mock_timer.start.assert_not_called()
def test_reply_long_benq(pjlink, caplog):
"""
Tests invalid reply from BenQ projectors
"""
# GIVEN: Test setup
caplog.set_level(logging.DEBUG)
t_data = 'Version1'
logs = [(f'{test_module}', logging.WARNING,
f'({pjlink.entry.name}) Non-standard CLSS reply: "{t_data}"'),
(f'{test_module}', logging.DEBUG,
f'({pjlink.entry.name}) Setting pjlink_class for this projector to "1"')
]
with patch.object(pjlink, 'poll_timer') as mock_timer:
# WHEN: process_clss called
caplog.clear()
t_chk = process_clss(projector=pjlink, data=t_data)
# THEN: Log entries and settings apply
assert t_chk is None, f'Invalid return code {t_chk}'
assert caplog.record_tuples == logs, 'Invalid log entries'
assert pjlink.pjlink_class == '1', 'Should have set pjlink_class = "1"'
mock_timer.setInterval.assert_not_called()
mock_timer.start.assert_not_called()
def test_reply_nan(pjlink, caplog):
"""
Tests invalid reply (Not A Number)
"""
# GIVEN: Test setup
caplog.set_level(logging.DEBUG)
t_data = 'A'
logs = [(f'{test_module}', logging.WARNING,
f'({pjlink.entry.name}) NAN CLSS version reply '
f'"{t_data}" - defaulting to class "1"'),
(f'{test_module}', logging.DEBUG,
f'({pjlink.entry.name}) Setting pjlink_class for this projector to "1"')
]
with patch.object(pjlink, 'poll_timer') as mock_timer:
# WHEN: process_clss called
caplog.clear()
t_chk = process_clss(projector=pjlink, data=t_data)
# THEN: Log entries and settings apply
assert t_chk is None, f'Invalid return code {t_chk}'
assert caplog.record_tuples == logs, 'Invalid log entries'
assert pjlink.pjlink_class == '1', 'Should have set pjlink_class = "1"'
mock_timer.setInterval.assert_not_called()
mock_timer.start.assert_not_called()
def test_class_1(pjlink, caplog):
"""
Tests valid Class 1 reply
"""
# GIVEN: Test setup
caplog.set_level(logging.DEBUG)
t_data = '1'
logs = [(f'{test_module}', logging.DEBUG,
f'({pjlink.entry.name}) Setting pjlink_class for this projector to "{t_data}"')
]
with patch.object(pjlink, 'poll_timer') as mock_timer:
# WHEN: process_clss called
caplog.clear()
t_chk = process_clss(projector=pjlink, data=t_data)
# THEN: Log entries and settings apply
assert t_chk is None, f'Invalid return code {t_chk}'
assert caplog.record_tuples == logs, 'Invalid log entries'
assert pjlink.pjlink_class == '1', 'Should have set pjlink_class = "1"'
mock_timer.setInterval.assert_not_called()
mock_timer.start.assert_not_called()
def test_class_2(pjlink, caplog):
"""
Tests valid Class 1 reply
"""
# GIVEN: Test setup
caplog.set_level(logging.DEBUG)
t_data = '2'
pjlink.pjlink_class = '1'
logs = [(f'{test_module}', logging.DEBUG,
f'({pjlink.entry.name}) Setting pjlink_class for this projector to "{t_data}"')
]
with patch.object(pjlink, 'poll_timer') as mock_timer:
# WHEN: process_clss called
caplog.clear()
t_chk = process_clss(projector=pjlink, data=t_data)
# THEN: Log entries and settings apply
assert t_chk is None, f'Invalid return code {t_chk}'
assert caplog.record_tuples == logs, 'Invalid log entries'
assert pjlink.pjlink_class == t_data, f'Should have set pjlink_class = "{t_data}"'
mock_timer.setInterval.assert_not_called()
mock_timer.start.assert_not_called()
def test_start_poll(pjlink, caplog):
"""
Tests poll_loop starts
"""
# GIVEN: Test setup
caplog.set_level(logging.DEBUG)
t_data = '2'
pjlink.no_poll = False
logs = [(f'{test_module}', logging.DEBUG,
f'({pjlink.entry.name}) Setting pjlink_class for this projector to "{t_data}"'),
(f'{test_module}', logging.DEBUG,
f'({pjlink.entry.name}) process_pjlink(): Starting timer')
]
with patch.object(pjlink, 'poll_timer') as mock_timer:
# WHEN: process_clss called
caplog.clear()
t_chk = process_clss(projector=pjlink, data=t_data)
# THEN: Log entries and settings apply
assert t_chk is None, f'Invalid return code {t_chk}'
assert caplog.record_tuples == logs, 'Invalid log entries'
assert pjlink.pjlink_class == t_data, f'Should have set pjlink_class = "{t_data}"'
mock_timer.setInterval.assert_called_with(1000)
mock_timer.start.assert_called_once()

64
tests/openlp_core/projectors/messages/test_misc.py Normal file
View File

@ -0,0 +1,64 @@
# -*- 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/>. #
##########################################################################
"""
Tests for commands that do not need much testing
"""
import logging
import openlp.core.projectors.pjlinkcommands
from openlp.core.projectors.pjlinkcommands import process_srch
test_module = openlp.core.projectors.pjlinkcommands.__name__
def test_srch_no_projector(caplog):
"""
Test SRCH command with no projector instance
"""
# GIVEN: Test setup
caplog.set_level(logging.DEBUG)
logs = [(f'{test_module}', logging.WARNING, 'SRCH packet detected - ignoring')]
# WHEN: Called
t_chk = process_srch()
# THEN: Appropriate return code and log entries
assert t_chk is None, 'Invalid return code'
assert caplog.record_tuples == logs, 'Invalid log entries'
def test_srch_with_projector(pjlink, caplog):
"""
Test SRCH command with projector
"""
# GIVEN: Test setup
caplog.set_level(logging.DEBUG)
logs = [(f'{test_module}', logging.WARNING,
f'({pjlink.entry.name}) SRCH packet detected - ignoring')]
# WHEN: Called
t_chk = process_srch(projector=pjlink)
# THEN: Appropriate return code and log entries
assert t_chk is None, 'Invalid return code'
assert caplog.record_tuples == logs, 'Invalid log entries'

128
tests/openlp_core/projectors/test_projector_commands_01.py
View File

@ -199,134 +199,6 @@ def test_projector_avmt_status_timer_check_delete(mock_log, mock_UpdateIcons, pj
mock_log.debug.assert_has_calls(log_debug_text)
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_clss_1(mock_log, pjlink):
"""
Test CLSS request returns non-standard reply 1
"""
# GIVEN: Test object
log_error_calls = []
log_warning_calls = []
log_debug_calls = [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: Process non-standard reply
process_command(projector=pjlink, cmd='CLSS', data='1')
# THEN: Projector class should be set with proper value
assert '1' == pjlink.pjlink_class, 'Should have set class=1'
mock_log.error.assert_has_calls(log_error_calls)
mock_log.warning.assert_has_calls(log_warning_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_clss_2(mock_log, pjlink):
"""
Test CLSS request returns non-standard reply 1
"""
# GIVEN: Test object
log_error_calls = []
log_warning_calls = []
log_debug_calls = [call('({ip}) Processing command "CLSS" with data "2"'.format(ip=pjlink.name)),
call('({ip}) Calling function for CLSS'.format(ip=pjlink.name)),
call('({ip}) Setting pjlink_class for this projector to "2"'.format(ip=pjlink.name))]
# WHEN: Process non-standard reply
process_command(projector=pjlink, cmd='CLSS', data='2')
# THEN: Projector class should be set with proper value
assert '2' == pjlink.pjlink_class, 'Should have set class=2'
mock_log.error.assert_has_calls(log_error_calls)
mock_log.warning.assert_has_calls(log_warning_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_clss_invalid_nan(mock_log, pjlink):
"""
Test CLSS reply has no class number
"""
# GIVEN: Test setup
log_warning_calls = [call('({ip}) NAN CLSS version reply "Z" - '
'defaulting to class "1"'.format(ip=pjlink.name))]
log_debug_calls = [call('({ip}) Processing command "CLSS" with data "Z"'.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: Process invalid reply
process_command(projector=pjlink, cmd='CLSS', data='Z')
# THEN: Projector class should be set with default value
assert pjlink.pjlink_class == '1', 'Invalid NaN class reply should have set class=1'
mock_log.warning.assert_has_calls(log_warning_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_clss_invalid_no_version(mock_log, pjlink):
"""
Test CLSS reply has no class number
"""
# GIVEN: Test object
log_warning_calls = [call('({ip}) No numbers found in class version reply "Invalid" '
'- defaulting to class "1"'.format(ip=pjlink.name))]
log_debug_calls = [call('({ip}) Processing command "CLSS" with data "Invalid"'.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: Process invalid reply
process_command(projector=pjlink, cmd='CLSS', data='Invalid')
# THEN: Projector class should be set with default value
assert pjlink.pjlink_class == '1', 'Invalid class reply should have set class=1'
mock_log.warning.assert_has_calls(log_warning_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_clss_nonstandard_reply_1(mock_log, pjlink):
"""
Test CLSS request returns non-standard reply 1
"""
# GIVEN: Test object
log_error_calls = []
log_warning_calls = [call('({ip}) Non-standard CLSS reply: "Class 1"'.format(ip=pjlink.name))]
log_debug_calls = [call('({ip}) Processing command "CLSS" with data "Class 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: Process non-standard reply
process_command(projector=pjlink, cmd='CLSS', data='Class 1')
# THEN: Projector class should be set with proper value
assert '1' == pjlink.pjlink_class, 'Non-standard class reply should have set class=1'
mock_log.error.assert_has_calls(log_error_calls)
mock_log.warning.assert_has_calls(log_warning_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_clss_nonstandard_reply_2(mock_log, pjlink):
"""
Test CLSS request returns non-standard reply 1
"""
# GIVEN: Test object
log_warning_calls = [call('({ip}) Non-standard CLSS reply: "Version2"'.format(ip=pjlink.name))]
log_debug_calls = [call('({ip}) Processing command "CLSS" with data "Version2"'.format(ip=pjlink.name)),
call('({ip}) Calling function for CLSS'.format(ip=pjlink.name)),
call('({ip}) Setting pjlink_class for this projector to "2"'.format(ip=pjlink.name))]
# WHEN: Process non-standard reply
process_command(projector=pjlink, cmd='CLSS', data='Version2')
# THEN: Projector class should be set with proper value
assert '2' == pjlink.pjlink_class, 'Non-standard class reply should have set class=1'
mock_log.warning.assert_has_calls(log_warning_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_erst_all_error(mock_log, pjlink):
"""