From a503c61a51b6b33065559d07eb325529a1b3557d Mon Sep 17 00:00:00 2001 From: Ken Roberts Date: Sat, 4 May 2019 03:21:48 -0700 Subject: [PATCH] PJLink2 update v05 --- openlp/core/projectors/pjlink.py | 20 +- openlp/core/projectors/pjlinkcommands.py | 12 +- .../test_projector_command_routing.py | 107 +++- .../projectors/test_projector_commands_01.py | 549 ++++++++++++++++++ ...ds_01.py => test_projector_commands_02.py} | 397 +------------ .../test_projector_pjlink_cmd_routing.py | 121 +--- 6 files changed, 689 insertions(+), 517 deletions(-) create mode 100644 tests/openlp_core/projectors/test_projector_commands_01.py rename tests/openlp_core/projectors/{test_projector_pjlink_commands_01.py => test_projector_commands_02.py} (54%) diff --git a/openlp/core/projectors/pjlink.py b/openlp/core/projectors/pjlink.py index 1383b27d8..5aa6f2b20 100644 --- a/openlp/core/projectors/pjlink.py +++ b/openlp/core/projectors/pjlink.py @@ -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): """ diff --git a/openlp/core/projectors/pjlinkcommands.py b/openlp/core/projectors/pjlinkcommands.py index 1834acc03..6df155db1 100644 --- a/openlp/core/projectors/pjlinkcommands.py +++ b/openlp/core/projectors/pjlinkcommands.py @@ -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() diff --git a/tests/openlp_core/projectors/test_projector_command_routing.py b/tests/openlp_core/projectors/test_projector_command_routing.py index 38721eb24..866ee3132 100644 --- a/tests/openlp_core/projectors/test_projector_command_routing.py +++ b/tests/openlp_core/projectors/test_projector_command_routing.py @@ -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]) diff --git a/tests/openlp_core/projectors/test_projector_commands_01.py b/tests/openlp_core/projectors/test_projector_commands_01.py new file mode 100644 index 000000000..64fa4c1fc --- /dev/null +++ b/tests/openlp_core/projectors/test_projector_commands_01.py @@ -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 . # +########################################################################## +""" +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) diff --git a/tests/openlp_core/projectors/test_projector_pjlink_commands_01.py b/tests/openlp_core/projectors/test_projector_commands_02.py similarity index 54% rename from tests/openlp_core/projectors/test_projector_pjlink_commands_01.py rename to tests/openlp_core/projectors/test_projector_commands_02.py index 48e479660..f483269a8 100644 --- a/tests/openlp_core/projectors/test_projector_pjlink_commands_01.py +++ b/tests/openlp_core/projectors/test_projector_commands_02.py @@ -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"' diff --git a/tests/openlp_core/projectors/test_projector_pjlink_cmd_routing.py b/tests/openlp_core/projectors/test_projector_pjlink_cmd_routing.py index a5225fd69..669f0a10a 100644 --- a/tests/openlp_core/projectors/test_projector_pjlink_cmd_routing.py +++ b/tests/openlp_core/projectors/test_projector_pjlink_cmd_routing.py @@ -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()