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

PJLink2 update v05

This commit is contained in:
Ken Roberts 2019-05-04 03:21:48 -07:00
parent 2a479a15da
commit a503c61a51
6 changed files with 689 additions and 517 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -861,6 +861,24 @@ class PJLink(QtNetwork.QTcpSocket):
log.debug('({ip}) Sending POWR command'.format(ip=self.entry.name))
return self.send_command(cmd='POWR', priority=priority)
def set_audio_mute(self, priority=False):
"""
Send command to set audio to muted
"""
log.debug('({ip}) Setting AVMT to 21 (audio mute)'.format(ip=self.entry.name))
self.send_command(cmd='AVMT', opts='21', priority=True)
self.status_timer_add(cmd='AVMT', callback=self.get_av_mute_status)
self.poll_loop()
def set_audio_normal(self, priority=False):
"""
Send command to set audio to normal
"""
log.debug('({ip}) Setting AVMT to 20 (audio normal)'.format(ip=self.entry.name))
self.send_command(cmd='AVMT', opts='20', priority=True)
self.status_timer_add(cmd='AVMT', callback=self.get_av_mute_status)
self.poll_loop()
def set_input_source(self, src=None):
"""
Verify input source available as listed in 'INST' command,
@ -924,9 +942,9 @@ class PJLink(QtNetwork.QTcpSocket):
log.warning('({ip}) "{cmd}" already in checks - returning'.format(ip=self.entry.name, cmd=cmd))
return
log.debug('({ip}) Adding "{cmd}" callback for status timer'.format(ip=self.entry.name, cmd=cmd))
self.status_timer_checks[cmd] = callback
if not self.status_timer.isActive():
self.status_timer.start()
self.status_timer_checks[cmd] = callback
def status_timer_delete(self, cmd):
"""

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

@ -113,15 +113,21 @@ def process_avmt(projector, data):
'31': {'shutter': True, 'mute': True}
}
if data not in settings:
log.warning('({ip}) Invalid shutter response: {data}'.format(ip=projector.entry.name, data=data))
log.warning('({ip}) Invalid av mute response: {data}'.format(ip=projector.entry.name, data=data))
return
shutter = settings[data]['shutter']
mute = settings[data]['mute']
# Check if we need to update the icons
update_icons = (shutter != projector.shutter) or (mute != projector.mute)
projector.shutter = shutter
projector.mute = mute
if update_icons:
if projector.shutter != shutter:
projector.shutter = shutter
log.debug('({ip}) Setting shutter to {chk}'.format(ip=projector.entry.name,
chk='closed' if shutter else 'open'))
if projector.mute != mute:
projector.mute = mute
log.debug('({ip}) Setting speaker to {chk}'.format(ip=projector.entry.name,
chk='muted' if shutter else 'normal'))
if 'AVMT' in projector.status_timer_checks:
projector.status_timer_delete('AVMT')
projector.projectorUpdateIcons.emit()

107
tests/openlp_core/projectors/test_projector_command_routing.py
View File

@ -28,7 +28,8 @@ 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.constants import E_AUTHENTICATION, E_PARAMETER, E_PROJECTOR, E_UNAVAILABLE, E_UNDEFINED, \
S_DATA_OK, PJLINK_ERRORS, STATUS_MSG
from openlp.core.projectors.db import Projector
from openlp.core.projectors.pjlink import PJLink
from tests.resources.projector.data import TEST1_DATA
@ -114,19 +115,111 @@ class TestPJLinkRouting(TestCase):
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):
def test_routing_pjink_err1(self, mock_log):
"""
Test rouing when PJLink error received (err1, err2, err3, err4, erra)
Test rouing when PJLink ERR1 received
"""
# 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
err_msg = STATUS_MSG[err_code]
err_str = PJLINK_ERRORS[err_code]
log_error_text = [call('({ip}) CLSS: {err}'.format(ip=self.pjlink.name, err=err_msg))]
log_debug_text = [call('({ip}) Processing command "CLSS" with data "{err}"'.format(ip=self.pjlink.name,
err=err_str))]
# WHEN: routing called
chk = process_command(projector=self.pjlink, cmd='CLSS', data='ERR1')
chk = process_command(projector=self.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 E_UNDEFINED'
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(self, mock_log):
"""
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=self.pjlink.name, err=err_msg))]
log_debug_text = [call('({ip}) Processing command "CLSS" with data "{err}"'.format(ip=self.pjlink.name,
err=err_str))]
# WHEN: routing called
chk = process_command(projector=self.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(self, mock_log):
"""
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=self.pjlink.name, err=err_msg))]
log_debug_text = [call('({ip}) Processing command "CLSS" with data "{err}"'.format(ip=self.pjlink.name,
err=err_str))]
# WHEN: routing called
chk = process_command(projector=self.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(self, mock_log):
"""
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=self.pjlink.name, err=err_msg))]
log_debug_text = [call('({ip}) Processing command "CLSS" with data "{err}"'.format(ip=self.pjlink.name,
err=err_str))]
# WHEN: routing called
chk = process_command(projector=self.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(self, mock_log):
"""
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=self.pjlink.name, err=err_msg))]
log_debug_text = [call('({ip}) Processing command "CLSS" with data "{err}"'.format(ip=self.pjlink.name,
err=err_str))]
# WHEN: routing called
chk = process_command(projector=self.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])

549
tests/openlp_core/projectors/test_projector_commands_01.py Normal file
View File

@ -0,0 +1,549 @@
# -*- 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 commands package.
"""
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_ERROR, E_WARN, PJLINK_ERST_DATA, PJLINK_ERST_STATUS, S_OK
from openlp.core.projectors.db import Projector
from openlp.core.projectors.pjlink import PJLink
from tests.resources.projector.data import TEST1_DATA
class TestPJLinkCommands(TestCase):
"""
Tests PJLink get status commands part 1
"""
def setUp(self):
"""
Initial test setup
"""
self.pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
def tearDown(self):
"""
Test reset
"""
del(self.pjlink)
@patch.object(openlp.core.projectors.pjlink.PJLink, 'projectorUpdateIcons')
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_avmt_audio_muted(self, mock_log, mock_UpdateIcons):
"""
Test avmt status shutter unchanged and mute on
"""
# GIVEN: Test setup
log_warning_text = []
log_debug_text = [call('({ip}) Processing command "AVMT" with data "21"'.format(ip=self.pjlink.name)),
call('({ip}) Calling function for AVMT'.format(ip=self.pjlink.name)),
call('({ip}) Setting speaker to muted'.format(ip=self.pjlink.name))]
self.pjlink.shutter = True
self.pjlink.mute = False
# WHEN: Called with setting shutter closed and mute on
process_command(projector=self.pjlink, cmd='AVMT', data='21')
# THEN: Shutter should be closed and mute should be True
assert self.pjlink.shutter, 'Shutter should not have changed'
assert self.pjlink.mute, 'Audio should be off'
assert mock_UpdateIcons.emit.called, 'Update icons should have been called'
mock_log.warning.assert_has_calls(log_warning_text)
mock_log.debug.assert_has_calls(log_debug_text)
@patch.object(openlp.core.projectors.pjlink.PJLink, 'projectorUpdateIcons')
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_avmt_bad_data(self, mock_log, mock_UpdateIcons):
"""
Test avmt bad data fail
"""
# GIVEN: Test object
log_warning_text = [call('({ip}) Invalid av mute response: 36'.format(ip=self.pjlink.name))]
log_debug_text = [call('({ip}) Processing command "AVMT" with data "36"'.format(ip=self.pjlink.name)),
call('({ip}) Calling function for AVMT'.format(ip=self.pjlink.name))]
self.pjlink.shutter = True
self.pjlink.mute = True
# WHEN: Called with an invalid setting
process_command(projector=self.pjlink, cmd='AVMT', data='36')
# THEN: Shutter should be closed and mute should be True
assert self.pjlink.shutter, 'Shutter should changed'
assert self.pjlink.mute, 'Audio should not have changed'
assert (not mock_UpdateIcons.emit.called), 'Update icons should NOT have been called'
mock_log.warning.assert_has_calls(log_warning_text)
mock_log.debug.assert_has_calls(log_debug_text)
@patch.object(openlp.core.projectors.pjlink.PJLink, 'projectorUpdateIcons')
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_avmt_closed_muted(self, mock_log, mock_UpdateIcons):
"""
Test avmt status shutter closed and mute off
"""
# GIVEN: Test object
log_warning_text = []
log_debug_text = [call('({ip}) Processing command "AVMT" with data "31"'.format(ip=self.pjlink.name)),
call('({ip}) Calling function for AVMT'.format(ip=self.pjlink.name)),
call('({ip}) Setting shutter to closed'.format(ip=self.pjlink.name)),
call('({ip}) Setting speaker to muted'.format(ip=self.pjlink.name))]
self.pjlink.shutter = False
self.pjlink.mute = False
# WHEN: Called with setting shutter to closed and mute on
process_command(projector=self.pjlink, cmd='AVMT', data='31')
# THEN: Shutter should be closed and mute should be True
assert self.pjlink.shutter, 'Shutter should have been set to closed'
assert self.pjlink.mute, 'Audio should be muted'
assert mock_UpdateIcons.emit.called, 'Update icons should have been called'
mock_log.warning.assert_has_calls(log_warning_text)
mock_log.debug.assert_has_calls(log_debug_text)
@patch.object(openlp.core.projectors.pjlink.PJLink, 'projectorUpdateIcons')
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_avmt_open_unmuted(self, mock_log, mock_UpdateIcons):
"""
Test avmt status shutter open and mute off
"""
# GIVEN: Test object
log_warning_text = []
log_debug_text = [call('({ip}) Processing command "AVMT" with data "30"'.format(ip=self.pjlink.name)),
call('({ip}) Calling function for AVMT'.format(ip=self.pjlink.name)),
call('({ip}) Setting shutter to open'.format(ip=self.pjlink.name)),
call('({ip}) Setting speaker to normal'.format(ip=self.pjlink.name))]
self.pjlink.shutter = True
self.pjlink.mute = True
# WHEN: Called with setting shutter to closed and mute on
process_command(projector=self.pjlink, cmd='AVMT', data='30')
# THEN: Shutter should be closed and mute should be True
assert (not self.pjlink.shutter), 'Shutter should have been set to off'
assert (not self.pjlink.mute), 'Audio should be on'
assert mock_UpdateIcons.emit.called, 'Update icons should have been called'
mock_log.warning.assert_has_calls(log_warning_text)
mock_log.debug.assert_has_calls(log_debug_text)
@patch.object(openlp.core.projectors.pjlink.PJLink, 'projectorUpdateIcons')
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_avmt_shutter_closed(self, mock_log, mock_UpdateIcons):
"""
Test avmt status shutter closed and audio unchanged
"""
# GIVEN: Test object
log_warning_text = []
log_debug_text = [call('({ip}) Processing command "AVMT" with data "11"'.format(ip=self.pjlink.name)),
call('({ip}) Calling function for AVMT'.format(ip=self.pjlink.name)),
call('({ip}) Setting shutter to closed'.format(ip=self.pjlink.name))]
self.pjlink.shutter = False
self.pjlink.mute = True
# WHEN: Called with setting shutter closed and mute off
process_command(projector=self.pjlink, cmd='AVMT', data='11')
# THEN: Shutter should be True and mute should be False
assert self.pjlink.shutter, 'Shutter should have been set to closed'
assert self.pjlink.mute, 'Audio should not have changed'
assert mock_UpdateIcons.emit.called, 'Update icons should have been called'
mock_log.warning.assert_has_calls(log_warning_text)
mock_log.debug.assert_has_calls(log_debug_text)
@patch.object(openlp.core.projectors.pjlink.PJLink, 'projectorUpdateIcons')
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_avmt_status_timer_check_delete(self, mock_log, mock_UpdateIcons):
"""
Test avmt deletes callback in projector.status_timer_check
"""
# GIVEN: Test object
log_warning_text = []
log_debug_text = [call('({ip}) Processing command "AVMT" with data "11"'.format(ip=self.pjlink.name)),
call('({ip}) Calling function for AVMT'.format(ip=self.pjlink.name)),
call('({ip}) Setting shutter to closed'.format(ip=self.pjlink.name))]
self.pjlink.shutter = False
self.pjlink.mute = True
self.pjlink.status_timer_checks = {'AVMT': self.pjlink.get_av_mute_status}
# WHEN: Called with setting shutter closed and mute off
with patch.object(self.pjlink, 'status_timer') as mock_status_timer:
process_command(projector=self.pjlink, cmd='AVMT', data='11')
# THEN: Shutter should be True and mute should be False
assert self.pjlink.shutter, 'Shutter should have been set to closed'
assert self.pjlink.mute, 'Audio should not have changed'
assert mock_UpdateIcons.emit.called, 'Update icons should have been called'
assert ('AVMT' not in self.pjlink.status_timer_checks), 'Status timer list should not have AVMT callback'
assert mock_status_timer.stop.called, 'Projector status_timer.stop() should have been called'
mock_log.warning.assert_has_calls(log_warning_text)
mock_log.debug.assert_has_calls(log_debug_text)
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_clss_1(self, mock_log):
"""
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=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: Process non-standard reply
process_command(projector=self.pjlink, cmd='CLSS', data='1')
# THEN: Projector class should be set with proper value
assert ('1' == self.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(self, mock_log):
"""
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=self.pjlink.name)),
call('({ip}) Calling function for CLSS'.format(ip=self.pjlink.name)),
call('({ip}) Setting pjlink_class for this projector to "2"'.format(ip=self.pjlink.name))]
# WHEN: Process non-standard reply
process_command(projector=self.pjlink, cmd='CLSS', data='2')
# THEN: Projector class should be set with proper value
assert ('2' == self.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(self, mock_log):
"""
Test CLSS reply has no class number
"""
# GIVEN: Test setup
log_error_calls = [call('({ip}) NAN CLSS version reply "Z" - '
'defaulting to class "1"'.format(ip=self.pjlink.name))]
log_debug_calls = [call('({ip}) Processing command "CLSS" with data "Z"'.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: Process invalid reply
process_command(projector=self.pjlink, cmd='CLSS', data='Z')
# THEN: Projector class should be set with default value
assert (self.pjlink.pjlink_class == '1'), 'Invalid NaN class reply should have set class=1'
mock_log.error.assert_has_calls(log_error_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_clss_invalid_no_version(self, mock_log):
"""
Test CLSS reply has no class number
"""
# GIVEN: Test object
log_error_calls = [call('({ip}) No numbers found in class version reply "Invalid" '
'- defaulting to class "1"'.format(ip=self.pjlink.name))]
log_debug_calls = [call('({ip}) Processing command "CLSS" with data "Invalid"'.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: Process invalid reply
process_command(projector=self.pjlink, cmd='CLSS', data='Invalid')
# THEN: Projector class should be set with default value
assert (self.pjlink.pjlink_class == '1'), 'Invalid class reply should have set class=1'
mock_log.error.assert_has_calls(log_error_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_clss_nonstandard_reply_1(self, mock_log):
"""
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=self.pjlink.name))]
log_debug_calls = [call('({ip}) Processing command "CLSS" with data "Class 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: Process non-standard reply
process_command(projector=self.pjlink, cmd='CLSS', data='Class 1')
# THEN: Projector class should be set with proper value
assert ('1' == self.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(self, mock_log):
"""
Test CLSS request returns non-standard reply 1
"""
# GIVEN: Test object
log_warning_calls = [call('({ip}) Non-standard CLSS reply: "Version2"'.format(ip=self.pjlink.name))]
log_debug_calls = [call('({ip}) Processing command "CLSS" with data "Version2"'.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 "2"'.format(ip=self.pjlink.name))]
# WHEN: Process non-standard reply
process_command(projector=self.pjlink, cmd='CLSS', data='Version2')
# THEN: Projector class should be set with proper value
assert ('2' == self.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(self, mock_log):
"""
Test test_projector_process_erst_all_error
"""
# GIVEN: Test object
chk_data = '{fan}{lamp}{temp}{cover}{filt}{other}'.format(fan=PJLINK_ERST_STATUS[E_ERROR],
lamp=PJLINK_ERST_STATUS[E_ERROR],
temp=PJLINK_ERST_STATUS[E_ERROR],
cover=PJLINK_ERST_STATUS[E_ERROR],
filt=PJLINK_ERST_STATUS[E_ERROR],
other=PJLINK_ERST_STATUS[E_ERROR])
chk_test = {'Fan': E_ERROR,
'Lamp': E_ERROR,
'Temperature': E_ERROR,
'Cover': E_ERROR,
'Filter': E_ERROR,
'Other': E_ERROR}
log_warning_calls = []
log_debug_calls = [call('({ip}) Processing command "ERST" with data "{chk}"'.format(ip=self.pjlink.name,
chk=chk_data)),
call('({ip}) Calling function for ERST'.format(ip=self.pjlink.name))]
self.pjlink.projector_errors = None
# WHEN: process_erst with status set to WARN
process_command(projector=self.pjlink, cmd='ERST', data=chk_data)
# THEN: PJLink instance errors should match chk_value
assert (self.pjlink.projector_errors == chk_test), 'Projector errors should be all E_ERROR'
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_ok(self, mock_log):
"""
Test to verify pjlink.projector_errors is set to None when no errors
"""
# GIVEN: Test object
chk_data = '0' * PJLINK_ERST_DATA['DATA_LENGTH']
log_warning_calls = []
log_debug_calls = [call('({ip}) Processing command "ERST" with data "{chk}"'.format(ip=self.pjlink.name,
chk=chk_data)),
call('({ip}) Calling function for ERST'.format(ip=self.pjlink.name))]
# WHEN: process_erst with no errors
process_command(projector=self.pjlink, cmd='ERST', data=chk_data)
# THEN: PJLink instance errors should be None
assert (self.pjlink.projector_errors is None), 'projector_errors should have been set to None'
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_warn(self, mock_log):
"""
Test test_projector_process_erst_all_error
"""
# GIVEN: Test object
chk_data = '{fan}{lamp}{temp}{cover}{filt}{other}'.format(fan=PJLINK_ERST_STATUS[E_WARN],
lamp=PJLINK_ERST_STATUS[E_WARN],
temp=PJLINK_ERST_STATUS[E_WARN],
cover=PJLINK_ERST_STATUS[E_WARN],
filt=PJLINK_ERST_STATUS[E_WARN],
other=PJLINK_ERST_STATUS[E_WARN])
chk_test = {'Fan': E_WARN,
'Lamp': E_WARN,
'Temperature': E_WARN,
'Cover': E_WARN,
'Filter': E_WARN,
'Other': E_WARN}
log_warning_calls = []
log_debug_calls = [call('({ip}) Processing command "ERST" with data "{chk}"'.format(ip=self.pjlink.name,
chk=chk_data)),
call('({ip}) Calling function for ERST'.format(ip=self.pjlink.name))]
self.pjlink.projector_errors = None
# WHEN: process_erst with status set to WARN
process_command(projector=self.pjlink, cmd='ERST', data=chk_data)
# THEN: PJLink instance errors should match chk_value
assert (self.pjlink.projector_errors == chk_test), 'Projector errors should be all E_WARN'
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_data_invalid_length(self, mock_log):
"""
Test test_projector_process_erst_data_invalid_length
"""
# GIVEN: Test object
chk_data = '0' * (PJLINK_ERST_DATA['DATA_LENGTH'] + 1)
log_warn_calls = [call('({ip}) Invalid error status response "{data}": '
'length != {chk}'.format(ip=self.pjlink.name,
data=chk_data, chk=PJLINK_ERST_DATA['DATA_LENGTH']))]
log_debug_calls = [call('({ip}) Processing command "ERST" with data "{data}"'.format(ip=self.pjlink.name,
data=chk_data)),
call('({ip}) Calling function for ERST'.format(ip=self.pjlink.name))]
self.pjlink.projector_errors = None
# WHEN: process_erst called with invalid data (too many values
process_command(self.pjlink, cmd='ERST', data=chk_data)
# THEN: pjlink.projector_errors should be empty and warning logged
assert (not self.pjlink.projector_errors), 'There should be no errors'
mock_log.warning.assert_has_calls(log_warn_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_erst_data_invalid_nan(self, mock_log):
"""
Test ERST called with invalid data
"""
# GIVEN: Test object
chk_data = 'Z' + ('0' * (PJLINK_ERST_DATA['DATA_LENGTH'] - 1))
log_warn_calls = [call('({ip}) Invalid error status response "{data}"'.format(ip=self.pjlink.name,
data=chk_data))]
log_debug_calls = [call('({ip}) Processing command "ERST" with data "{data}"'.format(ip=self.pjlink.name,
data=chk_data)),
call('({ip}) Calling function for ERST'.format(ip=self.pjlink.name))]
self.pjlink.projector_errors = None
# WHEN: process_erst called with invalid data (too many values
process_command(self.pjlink, cmd='ERST', data=chk_data)
# THEN: pjlink.projector_errors should be empty and warning logged
assert (not self.pjlink.projector_errors), 'There should be no errors'
mock_log.warning.assert_has_calls(log_warn_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_erst_warn_cover_only(self, mock_log):
"""
Test test_projector_process_erst_warn_cover_only
"""
# GIVEN: Test object
chk_data = '{fan}{lamp}{temp}{cover}{filt}{other}'.format(fan=PJLINK_ERST_STATUS[S_OK],
lamp=PJLINK_ERST_STATUS[S_OK],
temp=PJLINK_ERST_STATUS[S_OK],
cover=PJLINK_ERST_STATUS[E_WARN],
filt=PJLINK_ERST_STATUS[S_OK],
other=PJLINK_ERST_STATUS[S_OK])
chk_test = {'Cover': E_WARN}
log_warn_calls = []
log_debug_calls = [call('({ip}) Processing command "ERST" with data "{data}"'.format(ip=self.pjlink.name,
data=chk_data)),
call('({ip}) Calling function for ERST'.format(ip=self.pjlink.name))]
self.pjlink.projector_errors = None
# WHEN: process_erst with status set to WARN
process_command(projector=self.pjlink, cmd='ERST', data=chk_data)
# THEN: PJLink instance errors should match only cover warning
assert (1 == len(self.pjlink.projector_errors)), 'There should only be 1 error listed in projector_errors'
assert ('Cover' in self.pjlink.projector_errors), '"Cover" should be the only error listed'
assert (self.pjlink.projector_errors['Cover'] == E_WARN), '"Cover" should have E_WARN listed as error'
assert (chk_test == self.pjlink.projector_errors), 'projector_errors should match test errors'
mock_log.warning.assert_has_calls(log_warn_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_inf1(self, mock_log):
"""
Test saving INF1 data (manufacturer)
"""
# GIVEN: Test object
chk_data = 'TEst INformation MultiCase'
log_warn_calls = []
log_debug_calls = [call('({ip}) Processing command "INF1" with data "{data}"'.format(ip=self.pjlink.name,
data=chk_data)),
call('({ip}) Calling function for INF1'.format(ip=self.pjlink.name)),
call('({ip}) Setting projector manufacturer data to '
'"{data}"'.format(ip=self.pjlink.name, data=chk_data))]
self.pjlink.manufacturer = None
# WHEN: process_inf called with test data
process_command(projector=self.pjlink, cmd='INF1', data=chk_data)
# THEN: Data should be saved
assert (self.pjlink.manufacturer == chk_data), 'Test data should have been saved'
mock_log.warning.assert_has_calls(log_warn_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_inf2(self, mock_log):
"""
Test saving INF2 data (model)
"""
# GIVEN: Test object
chk_data = 'TEst moDEl MultiCase'
log_warn_calls = []
log_debug_calls = [call('({ip}) Processing command "INF2" with data "{data}"'.format(ip=self.pjlink.name,
data=chk_data)),
call('({ip}) Calling function for INF2'.format(ip=self.pjlink.name)),
call('({ip}) Setting projector model to "{data}"'.format(ip=self.pjlink.name,
data=chk_data))]
self.pjlink.model = None
# WHEN: process_inf called with test data
process_command(projector=self.pjlink, cmd='INF2', data=chk_data)
# THEN: Data should be saved
assert (self.pjlink.model == chk_data), 'Test data should have been saved'
mock_log.warning.assert_has_calls(log_warn_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_info(self, mock_log):
"""
Test saving INF2 data (model)
"""
# GIVEN: Test object
chk_data = 'TEst ExtrANEous MultiCase INformatoin that MFGR might Set'
log_warn_calls = []
log_debug_calls = [call('({ip}) Processing command "INFO" with data "{data}"'.format(ip=self.pjlink.name,
data=chk_data)),
call('({ip}) Calling function for INFO'.format(ip=self.pjlink.name)),
call('({ip}) Setting projector other_info to "{data}"'.format(ip=self.pjlink.name,
data=chk_data))]
self.pjlink.other_info = None
# WHEN: process_inf called with test data
process_command(projector=self.pjlink, cmd='INFO', data=chk_data)
# THEN: Data should be saved
assert (self.pjlink.other_info == chk_data), 'Test data should have been saved'
mock_log.warning.assert_has_calls(log_warn_calls)
mock_log.debug.assert_has_calls(log_debug_calls)

397
tests/openlp_core/projectors/test_projector_pjlink_commands_01.py → tests/openlp_core/projectors/test_projector_commands_02.py
View File

@ -26,8 +26,8 @@ from unittest import TestCase, skip
from unittest.mock import call, patch
import openlp.core.projectors.pjlink
from openlp.core.projectors.constants import E_ERROR, E_WARN, PJLINK_ERST_DATA, PJLINK_ERST_STATUS, \
PJLINK_POWR_STATUS, S_NOT_CONNECTED, S_OK, S_ON, S_STANDBY, STATUS_CODE
from openlp.core.projectors.pjlinkcommands import process_command
from openlp.core.projectors.constants import PJLINK_POWR_STATUS, S_ON, S_STANDBY
from openlp.core.projectors.db import Projector
from openlp.core.projectors.pjlink import PJLink
from tests.resources.projector.data import TEST1_DATA
@ -35,396 +35,19 @@ from tests.resources.projector.data import TEST1_DATA
class TestPJLinkCommands(TestCase):
"""
Tests for the PJLinkCommands class part 1
Tests PJLink get status commands part 2
"""
@skip('Needs update to new setup')
def test_projector_process_inf1(self):
def setUp(self):
"""
Test saving INF1 data (manufacturer)
Initial test setup
"""
test_data = 'TEst INformation MultiCase'
self.pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
# GIVEN: Test object
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
pjlink.manufacturer = None
# WHEN: process_inf called with test data
pjlink.process_inf1(data=test_data)
# THEN: Data should be saved
assert pjlink.manufacturer == test_data, 'Test data should have been saved'
@skip('Needs update to new setup')
def test_projector_process_inf2(self):
def tearDown(self):
"""
Test saving INF2 data (model)
Test reset
"""
test_data = 'TEst moDEl MultiCase'
# GIVEN: Test object
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
pjlink.model = None
# WHEN: process_inf called with test data
pjlink.process_inf2(data=test_data)
# THEN: Data should be saved
assert pjlink.model == test_data, 'Test data should have been saved'
@skip('Needs update to new setup')
def test_projector_process_info(self):
"""
Test saving INFO data (other information)
"""
test_data = 'TEst ExtrANEous MultiCase INformatoin that MFGR might Set'
# GIVEN: Test object
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
pjlink.other_info = None
# WHEN: process_inf called with test data
pjlink.process_info(data=test_data)
# THEN: Data should be saved
assert pjlink.other_info == test_data, 'Test data should have been saved'
@skip('Needs update to new setup')
def test_projector_process_avmt_bad_data(self):
"""
Test avmt bad data fail
"""
# GIVEN: Test object
with patch.object(openlp.core.projectors.pjlink.PJLink, 'projectorUpdateIcons') as mock_UpdateIcons:
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
pjlink.shutter = True
pjlink.mute = True
# WHEN: Called with an invalid setting
pjlink.process_avmt('36')
# THEN: Shutter should be closed and mute should be True
assert pjlink.shutter is True, 'Shutter should changed'
assert pjlink.mute is True, 'Audio should not have changed'
assert mock_UpdateIcons.emit.called is False, 'Update icons should NOT have been called'
@skip('Needs update to new setup')
def test_projector_process_avmt_closed_muted(self):
"""
Test avmt status shutter closed and mute off
"""
# GIVEN: Test object
with patch.object(openlp.core.projectors.pjlink.PJLink, 'projectorUpdateIcons') as mock_UpdateIcons:
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
pjlink.shutter = False
pjlink.mute = False
# WHEN: Called with setting shutter to closed and mute on
pjlink.process_avmt('31')
# THEN: Shutter should be closed and mute should be True
assert pjlink.shutter is True, 'Shutter should have been set to closed'
assert pjlink.mute is True, 'Audio should be muted'
assert mock_UpdateIcons.emit.called is True, 'Update icons should have been called'
@skip('Needs update to new setup')
def test_projector_process_avmt_shutter_closed(self):
"""
Test avmt status shutter closed and audio unchanged
"""
# GIVEN: Test object
with patch.object(openlp.core.projectors.pjlink.PJLink, 'projectorUpdateIcons') as mock_UpdateIcons:
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
pjlink.shutter = False
pjlink.mute = True
# WHEN: Called with setting shutter closed and mute off
pjlink.process_avmt('11')
# THEN: Shutter should be True and mute should be False
assert pjlink.shutter is True, 'Shutter should have been set to closed'
assert pjlink.mute is True, 'Audio should not have changed'
assert mock_UpdateIcons.emit.called is True, 'Update icons should have been called'
@skip('Needs update to new setup')
def test_projector_process_avmt_audio_muted(self):
"""
Test avmt status shutter unchanged and mute on
"""
# GIVEN: Test object
with patch.object(openlp.core.projectors.pjlink.PJLink, 'projectorUpdateIcons') as mock_UpdateIcons:
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
pjlink.shutter = True
pjlink.mute = False
# WHEN: Called with setting shutter closed and mute on
pjlink.process_avmt('21')
# THEN: Shutter should be closed and mute should be True
assert pjlink.shutter is True, 'Shutter should not have changed'
assert pjlink.mute is True, 'Audio should be off'
assert mock_UpdateIcons.emit.called is True, 'Update icons should have been called'
@skip('Needs update to new setup')
def test_projector_process_avmt_open_unmuted(self):
"""
Test avmt status shutter open and mute off
"""
# GIVEN: Test object
with patch.object(openlp.core.projectors.pjlink.PJLink, 'projectorUpdateIcons') as mock_UpdateIcons:
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
pjlink.shutter = True
pjlink.mute = True
# WHEN: Called with setting shutter to closed and mute on
pjlink.process_avmt('30')
# THEN: Shutter should be closed and mute should be True
assert pjlink.shutter is False, 'Shutter should have been set to open'
assert pjlink.mute is False, 'Audio should be on'
assert mock_UpdateIcons.emit.called is True, 'Update icons should have been called'
@skip('Needs update to new setup')
def test_projector_process_clss_one(self):
"""
Test class 1 sent from projector
"""
# GIVEN: Test object
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
# WHEN: Process class response
pjlink.process_clss('1')
# THEN: Projector class should be set to 1
assert pjlink.pjlink_class == '1', 'Projector should have set class=1'
@skip('Needs update to new setup')
def test_projector_process_clss_two(self):
"""
Test class 2 sent from projector
"""
# GIVEN: Test object
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
# WHEN: Process class response
pjlink.process_clss('2')
# THEN: Projector class should be set to 1
assert pjlink.pjlink_class == '2', 'Projector should have set class=2'
@skip('Needs update to new setup')
def test_projector_process_clss_invalid_nan(self):
"""
Test CLSS reply has no class number
"""
# GIVEN: Test object
with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log:
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
log_debug_calls = [call('({ip}) reset_information() connect status is '
'{state}'.format(ip=pjlink.name, state=STATUS_CODE[S_NOT_CONNECTED])),
call('({ip}) Setting pjlink_class for this projector to "1"'.format(ip=pjlink.name))]
log_error_calls = [call('({ip}) NAN CLSS version reply "Z" - '
'defaulting to class "1"'.format(ip=pjlink.name))]
# WHEN: Process invalid reply
pjlink.process_clss('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.error.assert_has_calls(log_error_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
@skip('Needs update to new setup')
def test_projector_process_clss_invalid_no_version(self):
"""
Test CLSS reply has no class number
"""
# GIVEN: Test object
with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log:
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
log_debug_calls = [call('({ip}) reset_information() connect status is '
'{state}'.format(ip=pjlink.name, state=STATUS_CODE[S_NOT_CONNECTED])),
call('({ip}) Setting pjlink_class for this projector to "1"'.format(ip=pjlink.name))]
log_error_calls = [call('({ip}) No numbers found in class version reply "Invalid" '
'- defaulting to class "1"'.format(ip=pjlink.name))]
# WHEN: Process invalid reply
pjlink.process_clss('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.error.assert_has_calls(log_error_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
@skip('Needs update to new setup')
def test_projector_process_clss_nonstandard_reply_1(self):
"""
Test CLSS request returns non-standard reply 1
"""
# GIVEN: Test object
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
# WHEN: Process non-standard reply
pjlink.process_clss('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'
@skip('Needs update to new setup')
def test_projector_process_clss_nonstandard_reply_2(self):
"""
Test CLSS request returns non-standard reply 2
"""
# GIVEN: Test object
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
# WHEN: Process non-standard reply
pjlink.process_clss('Version2')
# THEN: Projector class should be set with proper value
assert '2' == pjlink.pjlink_class, 'Non-standard class reply should have set class=2'
@skip('Needs update to new setup')
def test_projector_process_erst_all_ok(self):
"""
Test to verify pjlink.projector_errors is set to None when no errors
"""
chk_data = '0' * PJLINK_ERST_DATA['DATA_LENGTH']
# GIVEN: Test object
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
# WHEN: process_erst with no errors
pjlink.process_erst(chk_data)
# THEN: PJLink instance errors should be None
assert pjlink.projector_errors is None, 'projector_errors should have been set to None'
@skip('Needs update to new setup')
def test_projector_process_erst_data_invalid_length(self):
"""
Test test_projector_process_erst_data_invalid_length
"""
# GIVEN: Test object
with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log:
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
pjlink.projector_errors = None
chk_data = '0' * (PJLINK_ERST_DATA['DATA_LENGTH'] + 1)
log_debug_calls = [call('({ip}) reset_information() connect status is '
'{state}'.format(ip=pjlink.name, state=STATUS_CODE[S_NOT_CONNECTED]))]
log_warn_calls = [call('({ip}) Invalid error status response "0000000": '
'length != {chk}'.format(ip=pjlink.name, chk=PJLINK_ERST_DATA['DATA_LENGTH']))]
# WHEN: process_erst called with invalid data (too many values
pjlink.process_erst(chk_data)
# THEN: pjlink.projector_errors should be empty and warning logged
assert pjlink.projector_errors is None, 'There should be no errors'
mock_log.debug.assert_has_calls(log_debug_calls)
mock_log.warning.assert_has_calls(log_warn_calls)
@skip('Needs update to new setup')
def test_projector_process_erst_data_invalid_nan(self):
"""
Test test_projector_process_erst_data_invalid_nan
"""
# GIVEN: Test object
with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log:
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
pjlink.projector_errors = None
chk_data = 'Z' + ('0' * (PJLINK_ERST_DATA['DATA_LENGTH'] - 1))
log_debug_calls = [call('({ip}) reset_information() connect status is '
'{state}'.format(ip=pjlink.name, state=STATUS_CODE[S_NOT_CONNECTED]))]
log_warn_calls = [call('({ip}) Invalid error status response "Z00000"'.format(ip=pjlink.name))]
# WHEN: process_erst called with invalid data (too many values
pjlink.process_erst(chk_data)
# THEN: pjlink.projector_errors should be empty and warning logged
assert pjlink.projector_errors is None, 'There should be no errors'
mock_log.debug.assert_has_calls(log_debug_calls)
mock_log.warning.assert_has_calls(log_warn_calls)
@skip('Needs update to new setup')
def test_projector_process_erst_all_warn(self):
"""
Test test_projector_process_erst_all_warn
"""
chk_data = '{fan}{lamp}{temp}{cover}{filt}{other}'.format(fan=PJLINK_ERST_STATUS[E_WARN],
lamp=PJLINK_ERST_STATUS[E_WARN],
temp=PJLINK_ERST_STATUS[E_WARN],
cover=PJLINK_ERST_STATUS[E_WARN],
filt=PJLINK_ERST_STATUS[E_WARN],
other=PJLINK_ERST_STATUS[E_WARN])
chk_test = {'Fan': E_WARN,
'Lamp': E_WARN,
'Temperature': E_WARN,
'Cover': E_WARN,
'Filter': E_WARN,
'Other': E_WARN}
# GIVEN: Test object
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
pjlink.projector_errors = None
# WHEN: process_erst with status set to WARN
pjlink.process_erst(chk_data)
# THEN: PJLink instance errors should match chk_value
assert pjlink.projector_errors == chk_test, 'Projector errors should be all E_WARN'
@skip('Needs update to new setup')
def test_projector_process_erst_all_error(self):
"""
Test test_projector_process_erst_all_error
"""
chk_data = '{fan}{lamp}{temp}{cover}{filt}{other}'.format(fan=PJLINK_ERST_STATUS[E_ERROR],
lamp=PJLINK_ERST_STATUS[E_ERROR],
temp=PJLINK_ERST_STATUS[E_ERROR],
cover=PJLINK_ERST_STATUS[E_ERROR],
filt=PJLINK_ERST_STATUS[E_ERROR],
other=PJLINK_ERST_STATUS[E_ERROR])
chk_test = {'Fan': E_ERROR,
'Lamp': E_ERROR,
'Temperature': E_ERROR,
'Cover': E_ERROR,
'Filter': E_ERROR,
'Other': E_ERROR}
# GIVEN: Test object
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
pjlink.projector_errors = None
# WHEN: process_erst with status set to WARN
pjlink.process_erst(chk_data)
# THEN: PJLink instance errors should match chk_value
assert pjlink.projector_errors == chk_test, 'Projector errors should be all E_ERROR'
@skip('Needs update to new setup')
def test_projector_process_erst_warn_cover_only(self):
"""
Test test_projector_process_erst_warn_cover_only
"""
chk_data = '{fan}{lamp}{temp}{cover}{filt}{other}'.format(fan=PJLINK_ERST_STATUS[S_OK],
lamp=PJLINK_ERST_STATUS[S_OK],
temp=PJLINK_ERST_STATUS[S_OK],
cover=PJLINK_ERST_STATUS[E_WARN],
filt=PJLINK_ERST_STATUS[S_OK],
other=PJLINK_ERST_STATUS[S_OK])
chk_test = {'Cover': E_WARN}
# GIVEN: Test object
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
pjlink.projector_errors = None
# WHEN: process_erst with status set to WARN
pjlink.process_erst(chk_data)
# THEN: PJLink instance errors should match only cover warning
assert 1 == len(pjlink.projector_errors), 'There should only be 1 error listed in projector_errors'
assert 'Cover' in pjlink.projector_errors, '"Cover" should be the only error listed'
assert pjlink.projector_errors['Cover'] == E_WARN, '"Cover" should have E_WARN listed as error'
assert chk_test == pjlink.projector_errors, 'projector_errors should match test errors'
del(self.pjlink)
@skip('Needs update to new setup')
def test_projector_process_inpt_valid(self):
@ -441,7 +64,7 @@ class TestPJLinkCommands(TestCase):
pjlink.source_available = chk_source_available
# WHEN: Called with input source
pjlink.process_inpt('21')
process_command(projector=self.pjlink, cmd='INPT', data='21')
# THEN: Input selected should reflect current input
assert pjlink.source == '21', 'Input source should be set to "21"'

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

@ -23,12 +23,11 @@
Package to test the openlp.core.projectors.pjlink command routing.
"""
from unittest import TestCase, skip
from unittest import TestCase
from unittest.mock import call, patch
import openlp.core.projectors.pjlink
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.constants import PJLINK_PREFIX
from openlp.core.projectors.db import Projector
from openlp.core.projectors.pjlink import PJLink
from tests.resources.projector.data import TEST1_DATA
@ -134,119 +133,3 @@ class TestPJLinkRouting(TestCase):
mock_log.warning.assert_has_calls(log_warning_text)
mock_log.debug.assert_has_calls(log_debug_text)
assert (mock_process_cmd.call_count == 0), 'process_command should not have been called'
@skip('Needs update to new setup')
def test_routing_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_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
"""
# GIVEN: Test object
with patch.object(openlp.core.projectors.pjlink, 'log') as mock_log, \
patch.object(openlp.core.projectors.pjlink.PJLink, 'process_pjlink') as mock_process_pjlink, \
patch.object(openlp.core.projectors.pjlink.PJLink, 'change_status') as mock_change_status, \
patch.object(openlp.core.projectors.pjlink.PJLink, 'disconnect_from_host') as mock_disconnect, \
patch.object(openlp.core.projectors.pjlink.PJLink, 'projectorAuthentication') as mock_authentication:
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
log_error_calls = [call('({ip}) PJLINK: {msg}'.format(ip=self.pjlink.name,
msg=STATUS_MSG[E_AUTHENTICATION]))]
log_debug_calls = [call('({ip}) Processing command "PJLINK" with data "ERRA"'.format(ip=self.pjlink.name))]
# WHEN: process_command called with ERRA
pjlink.process_command(cmd='PJLINK', data=PJLINK_ERRORS[E_AUTHENTICATION])
# THEN: Appropriate log entries should have been made and methods called/not called
assert mock_disconnect.called is True, 'disconnect_from_host should have been called'
mock_log.error.assert_has_calls(log_error_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
mock_change_status.assert_called_once_with(status=E_AUTHENTICATION)
mock_authentication.emit.assert_called_once_with(pjlink.name)
mock_process_pjlink.assert_not_called()