openlp-core team mailing list archive
-
openlp-core team
-
Mailing list archive
-
Message #33827
[Merge] lp:~alisonken1/openlp/pjlink2-v03 into lp:openlp
Ken Roberts has proposed merging lp:~alisonken1/openlp/pjlink2-v03 into lp:openlp with lp:~alisonken1/openlp/pjlink2-v02 as a prerequisite.
Commit message:
PJLink2 Update V03
Requested reviews:
OpenLP Core (openlp-core)
For more details, see:
https://code.launchpad.net/~alisonken1/openlp/pjlink2-v03/+merge/366612
NOTE: Part 3 of a multi-part merge.
v[1..n] merges are to fix tests
WARNING: Requires merge of lp:~alisonken1/openlp/pjlink2-v02
- Update tests for test_projector_pjlink_base_02.py
- Added setUp() and tearDown() to have a common pjlink instance
- Rename test_send_command_no_data() to test_send_command_not_connected() to reflect actual test
- Add tests for PJLink.send_command and PJLink._underscore_send_command
- Fix test_projector_pjlink_cmd_routing for logging change to PJLink._underscore_send_command()
- In openlp.core.projectors.pjlink.PJLink class:
- Added copy import for PJLink reset information so PJLINK_CLASS default is not changed
- Fix send_command() socket state check
- Change some log entries from debug to warning
- Add call to pjlink._underscore_send_command() for invalid command check if
send_queue or priority_queue has data to send
- Fix _underscore_send_command no data check to include setting send_busy to False
--------------------------------------------------------------------------------
lp:~alisonken1/openlp/pjlink2-v03 (revision 2861)
https://ci.openlp.io/job/Branch-01-Pull/2722/ [SUCCESS]
https://ci.openlp.io/job/Branch-02a-Linux-Tests/2616/ [SUCCESS]
https://ci.openlp.io/job/Branch-02b-macOS-Tests/391/ [SUCCESS]
https://ci.openlp.io/job/Branch-03a-Build-Source/214/ [SUCCESS]
https://ci.openlp.io/job/Branch-03b-Build-macOS/193/ [SUCCESS]
https://ci.openlp.io/job/Branch-04a-Code-Lint/1676/ [SUCCESS]
https://ci.openlp.io/job/Branch-04b-Test-Coverage/1489/ [SUCCESS]
https://ci.openlp.io/job/Branch-05-AppVeyor-Tests/371/ [SUCCESS]
All builds passed
--
Your team OpenLP Core is requested to review the proposed merge of lp:~alisonken1/openlp/pjlink2-v03 into lp:openlp.
=== modified file 'openlp/core/projectors/pjlink.py'
--- openlp/core/projectors/pjlink.py 2019-04-28 07:10:59 +0000
+++ openlp/core/projectors/pjlink.py 2019-04-28 07:10:59 +0000
@@ -48,6 +48,7 @@
"""
import logging
from codecs import decode
+from copy import copy
from PyQt5 import QtCore, QtNetwork
@@ -276,6 +277,7 @@
self.model_lamp = None # RLMP
self.mute = None # AVMT
self.other_info = None # INFO
+ self.pjlink_class = copy(PJLINK_CLASS)
self.pjlink_name = None # NAME
self.power = S_OFF # POWR
self.serial_no = None # SNUM
@@ -629,11 +631,14 @@
:param salt: Optional salt for md5 hash initial authentication
:param priority: Option to send packet now rather than queue it up
"""
- if QSOCKET_STATE[self.state()] != S_CONNECTED:
+ if QSOCKET_STATE[self.state()] != QSOCKET_STATE[S_CONNECTED]:
log.warning('({ip}) send_command(): Not connected - returning'.format(ip=self.entry.name))
return self.reset_information()
if cmd not in PJLINK_VALID_CMD:
log.error('({ip}) send_command(): Invalid command requested - ignoring.'.format(ip=self.entry.name))
+ if self.priority_queue or self.send_queue:
+ # Just in case there's already something to send
+ return self._send_command()
return
log.debug('({ip}) send_command(): Building cmd="{command}" opts="{data}"{salt}'.format(ip=self.entry.name,
command=cmd,
@@ -649,9 +654,9 @@
options=opts,
suffix=PJLINK_SUFFIX)
if out in self.priority_queue:
- log.debug('({ip}) send_command(): Already in priority queue - skipping'.format(ip=self.entry.name))
+ log.warning('({ip}) send_command(): Already in priority queue - skipping'.format(ip=self.entry.name))
elif out in self.send_queue:
- log.debug('({ip}) send_command(): Already in normal queue - skipping'.format(ip=self.entry.name))
+ log.warning('({ip}) send_command(): Already in normal queue - skipping'.format(ip=self.entry.name))
else:
if priority:
log.debug('({ip}) send_command(): Adding to priority queue'.format(ip=self.entry.name))
@@ -672,7 +677,8 @@
:param utf8: Send as UTF-8 string otherwise send as ASCII string
"""
if not data and not self.priority_queue and not self.send_queue:
- log.debug('({ip}) _send_command(): Nothing to send - returning'.format(ip=self.entry.name))
+ log.warning('({ip}) _send_command(): Nothing to send - returning'.format(ip=self.entry.name))
+ self.send_busy = False
return
log.debug('({ip}) _send_command(data="{data}")'.format(ip=self.entry.name,
data=data.strip() if data else data))
@@ -684,7 +690,7 @@
log.debug('({ip}) _send_command(): Connection status: {data}'.format(ip=self.entry.name,
data=conn_state))
if QSOCKET_STATE[self.state()] != S_CONNECTED:
- log.debug('({ip}) _send_command() Not connected - abort'.format(ip=self.entry.name))
+ log.warning('({ip}) _send_command() Not connected - abort'.format(ip=self.entry.name))
self.send_busy = False
return self.disconnect_from_host()
if data and data not in self.priority_queue:
@@ -707,7 +713,7 @@
log.debug('({ip}) _send_command(): Getting normal queued packet'.format(ip=self.entry.name))
else:
# No data to send
- log.debug('({ip}) _send_command(): No data to send'.format(ip=self.entry.name))
+ log.warning('({ip}) _send_command(): No data to send'.format(ip=self.entry.name))
self.send_busy = False
return
self.send_busy = True
=== modified file 'tests/openlp_core/projectors/test_projector_pjlink_base_02.py'
--- tests/openlp_core/projectors/test_projector_pjlink_base_02.py 2019-04-28 07:10:59 +0000
+++ tests/openlp_core/projectors/test_projector_pjlink_base_02.py 2019-04-28 07:10:59 +0000
@@ -22,11 +22,12 @@
"""
Package to test the openlp.core.projectors.pjlink base package.
"""
-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 S_NOT_CONNECTED
+from openlp.core.projectors.constants import E_NETWORK, PJLINK_PREFIX, PJLINK_SUFFIX, QSOCKET_STATE, \
+ S_CONNECTED, S_NOT_CONNECTED
from openlp.core.projectors.db import Projector
from openlp.core.projectors.pjlink import PJLink
from tests.resources.projector.data import TEST1_DATA
@@ -36,67 +37,658 @@
"""
Tests for the PJLink module
"""
- @skip('Needs update to new setup')
- @patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
- @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
- @patch.object(openlp.core.projectors.pjlink.PJLink, '_send_command')
- @patch.object(openlp.core.projectors.pjlink, 'log')
- def test_send_command_no_data(self, mock_log, mock_send_command, mock_reset, mock_state):
- """
- Test _send_command with no data to send
- """
- # GIVEN: Test object
- log_warning_calls = [call('({ip}) send_command(): Not connected - returning'.format(ip=TEST1_DATA['name']))]
-
- log_debug_calls = [call('PJlink(projector="< Projector(id="None", ip="111.111.111.111", port="1111", '
- 'mac_adx="11:11:11:11:11:11", pin="1111", name="___TEST_ONE___", '
- 'location="location one", notes="notes one", pjlink_name="None", '
- 'pjlink_class="None", manufacturer="None", model="None", '
- 'serial_no="Serial Number 1", other="None", sources="None", source_list="[]", '
- 'model_filter="Filter type 1", model_lamp="Lamp type 1", '
- 'sw_version="Version 1") >", args="()" kwargs="{\'no_poll\': True}")'),
- call('PJlinkCommands(args=() kwargs={})')]
+ def setUp(self):
+ """
+ Initialize test state(s)
+ """
+ # Default PJLink instance for tests
+ self.pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
+
+ def tearDown(self):
+ """
+ Cleanup test state(s)
+ """
+ del(self.pjlink)
+
+ # ------------ Test PJLink._underscore_send_command ----------
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'change_status')
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'write')
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'disconnect_from_host')
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
+ @patch.object(openlp.core.projectors.pjlink, 'log')
+ def test_local_send_command_network_error(self, mock_log, mock_reset, mock_state, mock_disconnect, mock_write,
+ mock_change_status):
+ """
+ Test _underscore_send_command when possible network error occured
+ """
+ # GIVEN: Test object
+ test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX,
+ clss=self.pjlink.pjlink_class,
+ suff=PJLINK_SUFFIX)
+ log_error_calls = []
+ log_warning_calls = [call('({ip}) _send_command(): -1 received - '
+ 'disconnecting from host'.format(ip=self.pjlink.name))]
+ log_debug_calls = [call('({ip}) _send_command(data="None")'.format(ip=self.pjlink.name)),
+ call('({ip}) _send_command(): priority_queue: []'.format(ip=self.pjlink.name)),
+ call("({ip}) _send_command(): send_queue: ['{data}\\r']".format(ip=self.pjlink.name,
+ data=test_command.strip())),
+ call('({ip}) _send_command(): Connection status: S_CONNECTED'.format(ip=self.pjlink.name)),
+ call('({ip}) _send_command(): Getting normal queued packet'.format(ip=self.pjlink.name)),
+ call('({ip}) _send_command(): Sending "{data}"'.format(ip=self.pjlink.name,
+ data=test_command.strip()))
+ ]
+
+ mock_state.return_value = QSOCKET_STATE[S_CONNECTED]
+ mock_write.return_value = -1
+ self.pjlink.send_queue = [test_command]
+ self.pjlink.priority_queue = []
+
+ # WHEN: _send_command called with no data and queue's emtpy
+ # Patch some attributes here since they are not available until after instantiation
+ with patch.object(self.pjlink, 'socket_timer') as mock_timer, \
+ patch.object(self.pjlink, 'waitForBytesWritten') as mock_waitBytes:
+ mock_waitBytes.return_value = True
+ self.pjlink._send_command()
+
+ # THEN:
+ 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)
+ mock_change_status.called_with(E_NETWORK, 'Error while sending data to projector')
+ assert (not self.pjlink.send_queue), 'Send queue should be empty'
+ assert (not self.pjlink.priority_queue), 'Priority queue should be empty'
+ assert mock_timer.start.called, 'Timer should have been called'
+ assert (not mock_reset.called), 'reset_information() should not should have been called'
+ assert mock_disconnect.called, 'disconnect_from_host() should have been called'
+ assert self.pjlink.send_busy, 'send_busy should be True'
+
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
+ @patch.object(openlp.core.projectors.pjlink, 'log')
+ def test_local_send_command_no_data(self, mock_log, mock_reset, mock_state):
+ """
+ Test _underscore_send_command with no data to send
+ """
+ # GIVEN: Test object
+ log_error_calls = []
+ log_warning_calls = [call('({ip}) _send_command(): Nothing to send - returning'.format(ip=self.pjlink.name))]
+ log_debug_calls = []
+ mock_state.return_value = S_CONNECTED
+ self.pjlink.send_queue = []
+ self.pjlink.priority_queue = []
+
+ # WHEN: _send_command called with no data and queue's emtpy
+ # Patch some attributes here since they are not available until after instantiation
+ with patch.object(self.pjlink, 'socket_timer') as mock_timer:
+ self.pjlink._send_command()
+
+ # THEN:
+ 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)
+ assert (not self.pjlink.send_queue), 'Send queue should be empty'
+ assert (not self.pjlink.priority_queue), 'Priority queue should be empty'
+ assert (not mock_timer.called), 'Timer should not have been called'
+ assert (not mock_reset.called), 'reset_information() should not have been called'
+
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
+ @patch.object(openlp.core.projectors.pjlink, 'log')
+ def test_local_send_command_no_data_queue_check(self, mock_log, mock_reset, mock_state):
+ """
+ Test _underscore_send_command last queue length check
+ """
+ # GIVEN: Test object
+ log_error_calls = []
+ log_warning_calls = [call('({ip}) _send_command(): No data to send'.format(ip=self.pjlink.name))]
+ log_debug_calls = []
+ mock_state.return_value = QSOCKET_STATE[S_CONNECTED]
+ self.pjlink.priority_queue = []
+
+ # WHEN: _send_command called with no data and queue's emtpy
+ # Patch some attributes here since they are not available until after instantiation
+ with patch.object(self.pjlink, 'socket_timer') as mock_timer, \
+ patch.object(self.pjlink, 'send_queue') as mock_queue:
+ # Unlikely case of send_queue not really empty, but len(send_queue) returns 0
+ mock_queue.return_value = ['test']
+ mock_queue.__len__.return_value = 0
+ self.pjlink._send_command(data=None)
+
+ # THEN:
+ 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)
+ assert (not self.pjlink.priority_queue), 'Priority queue should be empty'
+ assert (not mock_timer.called), 'Timer should not have been called'
+ assert (not mock_reset.called), 'reset_information() should not have been called'
+
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'write')
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'disconnect_from_host')
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
+ @patch.object(openlp.core.projectors.pjlink, 'log')
+ def test_local_send_command_normal_send(self, mock_log, mock_reset, mock_state, mock_disconnect, mock_write):
+ """
+ Test _underscore_send_command using normal queue
+ """
+ # GIVEN: Test object
+ test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX,
+ clss=self.pjlink.pjlink_class,
+ suff=PJLINK_SUFFIX)
+ log_error_calls = []
+ log_warning_calls = []
+ log_debug_calls = [call('({ip}) _send_command(data="None")'.format(ip=self.pjlink.name)),
+ call('({ip}) _send_command(): priority_queue: []'.format(ip=self.pjlink.name)),
+ call("({ip}) _send_command(): send_queue: ['{data}\\r']".format(ip=self.pjlink.name,
+ data=test_command.strip())),
+ call('({ip}) _send_command(): Connection status: S_CONNECTED'.format(ip=self.pjlink.name)),
+ call('({ip}) _send_command(): Getting normal queued packet'.format(ip=self.pjlink.name)),
+ call('({ip}) _send_command(): Sending "{data}"'.format(ip=self.pjlink.name,
+ data=test_command.strip()))
+ ]
+
+ mock_state.return_value = QSOCKET_STATE[S_CONNECTED]
+ mock_write.return_value = len(test_command)
+ self.pjlink.send_queue = [test_command]
+ self.pjlink.priority_queue = []
+
+ # WHEN: _send_command called with no data and queue's emtpy
+ # Patch some attributes here since they are not available until after instantiation
+ with patch.object(self.pjlink, 'socket_timer') as mock_timer, \
+ patch.object(self.pjlink, 'waitForBytesWritten') as mock_waitBytes:
+ mock_waitBytes.return_value = True
+ self.pjlink._send_command()
+
+ # THEN:
+ 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)
+ assert (not self.pjlink.send_queue), 'Send queue should be empty'
+ assert (not self.pjlink.priority_queue), 'Priority queue should be empty'
+ assert mock_timer.start.called, 'Timer should have been called'
+ assert (not mock_reset.called), 'reset_information() should not have been called'
+ assert (not mock_disconnect.called), 'disconnect_from_host() should not have been called'
+ assert self.pjlink.send_busy, 'send_busy flag should be True'
+
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'disconnect_from_host')
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
+ @patch.object(openlp.core.projectors.pjlink, 'log')
+ def test_local_send_command_not_connected(self, mock_log, mock_reset, mock_state, mock_disconnect):
+ """
+ Test _underscore_send_command when not connected
+ """
+ # GIVEN: Test object
+ test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX,
+ clss=self.pjlink.pjlink_class,
+ suff=PJLINK_SUFFIX)
+ log_error_calls = []
+ log_warning_calls = [call('({ip}) _send_command() Not connected - abort'.format(ip=self.pjlink.name))]
+ log_debug_calls = [call('({ip}) _send_command(data="None")'.format(ip=self.pjlink.name)),
+ call('({ip}) _send_command(): priority_queue: []'.format(ip=self.pjlink.name)),
+ call("({ip}) _send_command(): send_queue: ['%1CLSS ?\\r']".format(ip=self.pjlink.name)),
+ call('({ip}) _send_command(): Connection status: S_OK'.format(ip=self.pjlink.name))]
mock_state.return_value = S_NOT_CONNECTED
- pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
- pjlink.send_queue = []
- pjlink.priority_queue = []
-
- # WHEN: _send_command called with no data and queue's empty
- pjlink.send_command(cmd='DONTCARE')
-
- # THEN:
- mock_log.debug.assert_has_calls(log_debug_calls)
- mock_log.warning.assert_has_calls(log_warning_calls)
- assert mock_reset.called is True
- assert mock_reset.called is True
-
- @skip('Needs update to new setup')
- @patch.object(openlp.core.projectors.pjlink, 'log')
- def test_local_send_command_no_data(self, mock_log):
- """
- Test _send_command with no data to send
- """
- # GIVEN: Test object
- log_debug_calls = [call('PJlink(projector="< Projector(id="None", ip="111.111.111.111", port="1111", '
- 'mac_adx="11:11:11:11:11:11", pin="1111", name="___TEST_ONE___", '
- 'location="location one", notes="notes one", pjlink_name="None", '
- 'pjlink_class="None", manufacturer="None", model="None", '
- 'serial_no="Serial Number 1", other="None", sources="None", source_list="[]", '
- 'model_filter="Filter type 1", model_lamp="Lamp type 1", '
- 'sw_version="Version 1") >", args="()" kwargs="{\'no_poll\': True}")'),
- call('PJlinkCommands(args=() kwargs={})'),
- call('(___TEST_ONE___) reset_information() connect status is S_NOT_CONNECTED'),
- call('(___TEST_ONE___) _send_command(): Nothing to send - returning')]
-
- pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
- pjlink.send_queue = []
- pjlink.priority_queue = []
+ self.pjlink.send_queue = [test_command]
+ self.pjlink.priority_queue = []
# WHEN: _send_command called with no data and queue's emtpy
# Patch here since pjlink does not have socket_timer until after instantiation
- with patch.object(pjlink, 'socket_timer') as mock_timer:
- pjlink._send_command(data=None, utf8=False)
-
- # THEN:
- mock_log.debug.assert_has_calls(log_debug_calls)
- assert mock_timer.called is False
+ with patch.object(self.pjlink, 'socket_timer') as mock_timer:
+ self.pjlink._send_command()
+
+ # THEN:
+ 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)
+ assert (self.pjlink.send_queue == [test_command]), 'Send queue should have one entry'
+ assert (not self.pjlink.priority_queue), 'Priority queue should be empty'
+ assert (not mock_timer.called), 'Timer should not have been called'
+ assert (not mock_reset.called), 'reset_information() should not have been called'
+ assert mock_disconnect.called, 'disconnect_from_host() should have been called'
+ assert (not self.pjlink.send_busy), 'send_busy flag should be False'
+
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'write')
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'disconnect_from_host')
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
+ @patch.object(openlp.core.projectors.pjlink, 'log')
+ def test_local_send_command_priority_send(self, mock_log, mock_reset, mock_state, mock_disconnect, mock_write):
+ """
+ Test _underscore_send_command with priority queue
+ """
+ # GIVEN: Test object
+ test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX,
+ clss=self.pjlink.pjlink_class,
+ suff=PJLINK_SUFFIX)
+ log_error_calls = []
+ log_warning_calls = []
+ log_debug_calls = [call('({ip}) _send_command(data="{data}")'.format(ip=self.pjlink.name,
+ data=test_command.strip())),
+ call('({ip}) _send_command(): priority_queue: []'.format(ip=self.pjlink.name)),
+ call('({ip}) _send_command(): send_queue: []'.format(ip=self.pjlink.name)),
+ call('({ip}) _send_command(): Connection status: S_CONNECTED'.format(ip=self.pjlink.name)),
+ call('({ip}) _send_command(): Priority packet - '
+ 'adding to priority queue'.format(ip=self.pjlink.name)),
+ call('({ip}) _send_command(): Getting priority queued packet'.format(ip=self.pjlink.name)),
+ call('({ip}) _send_command(): Sending "{data}"'.format(ip=self.pjlink.name,
+ data=test_command.strip()))
+ ]
+
+ mock_state.return_value = QSOCKET_STATE[S_CONNECTED]
+ mock_write.return_value = len(test_command)
+ self.pjlink.send_queue = []
+ self.pjlink.priority_queue = []
+
+ # WHEN: _send_command called with no data and queue's emtpy
+ # Patch some attributes here since they are not available until after instantiation
+ with patch.object(self.pjlink, 'socket_timer') as mock_timer, \
+ patch.object(self.pjlink, 'waitForBytesWritten') as mock_waitBytes:
+ mock_waitBytes.return_value = True
+ self.pjlink._send_command(data=test_command)
+
+ # THEN:
+ 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)
+ assert (not self.pjlink.send_queue), 'Send queue should be empty'
+ assert (not self.pjlink.priority_queue), 'Priority queue should be empty'
+ assert mock_timer.start.called, 'Timer should have been called'
+ assert (not mock_reset.called), 'reset_information() should not have been called'
+ assert (not mock_disconnect.called), 'disconnect_from_host() should not have been called'
+ assert self.pjlink.send_busy, 'send_busy flag should be True'
+
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'write')
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'disconnect_from_host')
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
+ @patch.object(openlp.core.projectors.pjlink, 'log')
+ def test_local_send_command_priority_send_with_normal_queue(self, mock_log, mock_reset, mock_state,
+ mock_disconnect, mock_write):
+ """
+ Test _underscore_send_command with priority queue when normal queue active
+ """
+ # GIVEN: Test object
+ test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX,
+ clss=self.pjlink.pjlink_class,
+ suff=PJLINK_SUFFIX)
+ log_error_calls = []
+ log_warning_calls = []
+ log_debug_calls = [call('({ip}) _send_command(data="{data}")'.format(ip=self.pjlink.name,
+ data=test_command.strip())),
+ call('({ip}) _send_command(): priority_queue: []'.format(ip=self.pjlink.name)),
+ call("({ip}) _send_command(): send_queue: ['{data}\\r']".format(ip=self.pjlink.name,
+ data=test_command.strip())),
+ call('({ip}) _send_command(): Connection status: S_CONNECTED'.format(ip=self.pjlink.name)),
+ call('({ip}) _send_command(): Priority packet - '
+ 'adding to priority queue'.format(ip=self.pjlink.name)),
+ call('({ip}) _send_command(): Getting priority queued packet'.format(ip=self.pjlink.name)),
+ call('({ip}) _send_command(): Sending "{data}"'.format(ip=self.pjlink.name,
+ data=test_command.strip()))
+ ]
+
+ mock_state.return_value = QSOCKET_STATE[S_CONNECTED]
+ mock_write.return_value = len(test_command)
+ self.pjlink.send_queue = [test_command]
+ self.pjlink.priority_queue = []
+
+ # WHEN: _send_command called with no data and queue's emtpy
+ # Patch some attributes here since they are not available until after instantiation
+ with patch.object(self.pjlink, 'socket_timer') as mock_timer, \
+ patch.object(self.pjlink, 'waitForBytesWritten') as mock_waitBytes:
+ mock_waitBytes.return_value = True
+ self.pjlink._send_command(data=test_command)
+
+ # THEN:
+ 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)
+ assert self.pjlink.send_queue, 'Send queue should have one entry'
+ assert (not self.pjlink.priority_queue), 'Priority queue should be empty'
+ assert mock_timer.start.called, 'Timer should have been called'
+ assert (not mock_reset.called), 'reset_information() should not have been called'
+ assert (not mock_disconnect.called), 'disconnect_from_host() should not have been called'
+ assert self.pjlink.send_busy, 'send_busy flag should be True'
+
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
+ @patch.object(openlp.core.projectors.pjlink, 'log')
+ def test_local_send_command_send_busy_normal_queue(self, mock_log, mock_reset, mock_state):
+ """
+ Test _underscore_send_command send_busy flag with normal queue
+ """
+ # GIVEN: Test object
+ test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX,
+ clss=self.pjlink.pjlink_class,
+ suff=PJLINK_SUFFIX)
+ log_error_calls = []
+ log_warning_calls = []
+ log_debug_calls = [call('({ip}) _send_command(data="None")'.format(ip=self.pjlink.name)),
+ call('({ip}) _send_command(): priority_queue: []'.format(ip=self.pjlink.name)),
+ call("({ip}) _send_command(): send_queue: ['{data}\\r']".format(ip=self.pjlink.name,
+ data=test_command.strip())),
+ call('({ip}) _send_command(): Connection status: S_CONNECTED'.format(ip=self.pjlink.name)),
+ call('({ip}) _send_command(): Still busy, returning'.format(ip=self.pjlink.name)),
+ call('({ip}) _send_command(): Priority queue = []'.format(ip=self.pjlink.name)),
+ call("({ip}) _send_command(): Normal queue = "
+ "['{data}\\r']".format(ip=self.pjlink.name, data=test_command.strip()))]
+
+ mock_state.return_value = QSOCKET_STATE[S_CONNECTED]
+ self.pjlink.send_busy = True
+ self.pjlink.send_queue = [test_command]
+ self.pjlink.priority_queue = []
+
+ # WHEN: _send_command called with no data and queue's emtpy
+ # Patch some attributes here since they are not available until after instantiation
+ with patch.object(self.pjlink, 'socket_timer') as mock_timer:
+ self.pjlink._send_command()
+
+ # THEN:
+ 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)
+ assert self.pjlink.send_queue, 'Send queue should have one entry'
+ assert (not self.pjlink.priority_queue), 'Priority queue should be empty'
+ assert (not mock_timer.start.called), 'Timer should not have been called'
+ assert (not mock_reset.called), 'reset_information() should not have been called'
+ assert self.pjlink.send_busy, 'send_busy flag should be True'
+
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
+ @patch.object(openlp.core.projectors.pjlink, 'log')
+ def test_local_send_command_send_busy_priority_queue(self, mock_log, mock_reset, mock_state):
+ """
+ Test _underscore_send_command send_busy flag with priority queue
+ """
+ # GIVEN: Test object
+ test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX,
+ clss=self.pjlink.pjlink_class,
+ suff=PJLINK_SUFFIX)
+ log_error_calls = []
+ log_warning_calls = []
+ log_debug_calls = [call('({ip}) _send_command(data="None")'.format(ip=self.pjlink.name)),
+ call("({ip}) _send_command(): priority_queue: "
+ "['{data}\\r']".format(ip=self.pjlink.name,
+ data=test_command.strip())),
+ call('({ip}) _send_command(): send_queue: []'.format(ip=self.pjlink.name)),
+ call('({ip}) _send_command(): Connection status: S_CONNECTED'.format(ip=self.pjlink.name)),
+ call('({ip}) _send_command(): Still busy, returning'.format(ip=self.pjlink.name)),
+ call("({ip}) _send_command(): Priority queue = "
+ "['{data}\\r']".format(ip=self.pjlink.name, data=test_command.strip())),
+ call('({ip}) _send_command(): Normal queue = []'.format(ip=self.pjlink.name))
+ ]
+
+ mock_state.return_value = QSOCKET_STATE[S_CONNECTED]
+ self.pjlink.send_busy = True
+ self.pjlink.send_queue = []
+ self.pjlink.priority_queue = [test_command]
+
+ # WHEN: _send_command called with no data and queue's emtpy
+ # Patch some attributes here since they are not available until after instantiation
+ with patch.object(self.pjlink, 'socket_timer') as mock_timer:
+ self.pjlink._send_command()
+
+ # THEN:
+ 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)
+ assert (not self.pjlink.send_queue), 'Send queue should be empty'
+ assert self.pjlink.priority_queue, 'Priority queue should have one entry'
+ assert (not mock_timer.start.called), 'Timer should not have been called'
+ assert (not mock_reset.called), 'reset_information() should not have been called'
+ assert self.pjlink.send_busy, 'send_busy flag should be True'
+
+ # ------------ Test PJLink.send_command ----------
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
+ @patch.object(openlp.core.projectors.pjlink.PJLink, '_send_command')
+ @patch.object(openlp.core.projectors.pjlink, 'log')
+ def test_send_command_add_normal_command(self, mock_log, mock_send_command, mock_reset, mock_state):
+ """
+ Test send_command adding normal queue item
+ """
+ # GIVEN: Test object
+ test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX,
+ clss=self.pjlink.pjlink_class,
+ suff=PJLINK_SUFFIX)
+ log_error_calls = []
+ log_warning_calls = []
+ log_debug_calls = [call('({ip}) send_command(): Building cmd="CLSS" opts="?"'.format(ip=self.pjlink.name)),
+ call('({ip}) send_command(): Adding to normal queue'.format(ip=self.pjlink.name))]
+ mock_state.return_value = S_CONNECTED
+
+ # Patch here since pjlink does not have priority or send queue's until instantiated
+ with patch.object(self.pjlink, 'send_queue') as mock_send, \
+ patch.object(self.pjlink, 'priority_queue') as mock_priority:
+
+ # WHEN: send_command called with valid normal command
+ self.pjlink.send_command(cmd='CLSS')
+
+ # THEN:
+ mock_send.append.called_with(test_command)
+ mock_priority.append.called is False
+ mock_log.debug.assert_has_calls(log_debug_calls)
+ mock_log.warning.assert_has_calls(log_warning_calls)
+ mock_log.error.assert_has_calls(log_error_calls)
+ assert (not mock_reset.called), 'reset_information() should not have been called'
+ assert mock_send_command.called, '_underscore_send_command() should have been called'
+
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
+ @patch.object(openlp.core.projectors.pjlink.PJLink, '_send_command')
+ @patch.object(openlp.core.projectors.pjlink, 'log')
+ def test_send_command_add_priority_command(self, mock_log, mock_send_command, mock_reset, mock_state):
+ """
+ Test _send_command adding priority queue item
+ """
+ # GIVEN: Test object
+ test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX,
+ clss=self.pjlink.pjlink_class,
+ suff=PJLINK_SUFFIX)
+ log_error_calls = []
+ log_warning_calls = []
+ log_debug_calls = [call('({ip}) send_command(): Building cmd="CLSS" opts="?"'.format(ip=self.pjlink.name)),
+ call('({ip}) send_command(): Adding to priority queue'.format(ip=self.pjlink.name))]
+ mock_state.return_value = S_CONNECTED
+
+ # Patch here since pjlink does not have priority or send queue's until instantiated
+ with patch.object(self.pjlink, 'send_queue') as mock_send, \
+ patch.object(self.pjlink, 'priority_queue') as mock_priority:
+
+ # WHEN: send_command called with valid priority command
+ self.pjlink.send_command(cmd='CLSS', priority=True)
+
+ # THEN:
+ mock_log.debug.assert_has_calls(log_debug_calls)
+ mock_log.warning.assert_has_calls(log_warning_calls)
+ mock_log.error.assert_has_calls(log_error_calls)
+ mock_priority.append.assert_called_with(test_command)
+ assert (not mock_send.append.called), 'send_queue should not have changed'
+ assert (not mock_reset.called), 'reset_information() should not have been called'
+ assert mock_send_command.called, '_underscore_send_command() should have been called'
+
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
+ @patch.object(openlp.core.projectors.pjlink.PJLink, '_send_command')
+ @patch.object(openlp.core.projectors.pjlink, 'log')
+ def test_send_command_duplicate_normal_command(self, mock_log, mock_send_command, mock_reset, mock_state):
+ """
+ Test send_command with duplicate item for normal queue
+ """
+ # GIVEN: Test object
+ test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX,
+ clss=self.pjlink.pjlink_class,
+ suff=PJLINK_SUFFIX)
+ log_error_calls = []
+ log_warning_calls = [call('({ip}) send_command(): Already in normal queue - '
+ 'skipping'.format(ip=self.pjlink.name))]
+ log_debug_calls = [call('({ip}) send_command(): Building cmd="CLSS" opts="?"'.format(ip=self.pjlink.name))]
+ mock_state.return_value = S_CONNECTED
+ self.pjlink.send_queue = [test_command]
+ self.pjlink.priority_queue = []
+
+ # WHEN: send_command called with same command in normal queue
+ self.pjlink.send_command(cmd='CLSS')
+
+ # THEN:
+ mock_log.debug.assert_has_calls(log_debug_calls)
+ mock_log.warning.assert_has_calls(log_warning_calls)
+ mock_log.error.assert_has_calls(log_error_calls)
+ assert (self.pjlink.send_queue == [test_command]), 'Send queue should have one entry'
+ assert (not self.pjlink.priority_queue), 'Priority queue should be empty'
+ assert (not mock_reset.called), 'reset_information() should not have been called'
+ assert mock_send_command.called, '_underscore_send_command() should have been called'
+
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
+ @patch.object(openlp.core.projectors.pjlink.PJLink, '_send_command')
+ @patch.object(openlp.core.projectors.pjlink, 'log')
+ def test_send_command_duplicate_priority_command(self, mock_log, mock_send_command, mock_reset, mock_state):
+ """
+ Test send_command with duplicate item for priority queue
+ """
+ # GIVEN: Test object
+ test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX,
+ clss=self.pjlink.pjlink_class,
+ suff=PJLINK_SUFFIX)
+ log_error_calls = []
+ log_warning_calls = [call('({ip}) send_command(): Already in priority queue - '
+ 'skipping'.format(ip=self.pjlink.name))]
+ log_debug_calls = [call('({ip}) send_command(): Building cmd="CLSS" opts="?"'.format(ip=self.pjlink.name))]
+ mock_state.return_value = S_CONNECTED
+ self.pjlink.send_queue = []
+ self.pjlink.priority_queue = [test_command]
+
+ # WHEN: send_command called with same command in priority queue
+ self.pjlink.send_command(cmd='CLSS', priority=True)
+
+ # THEN:
+ mock_log.debug.assert_has_calls(log_debug_calls)
+ mock_log.warning.assert_has_calls(log_warning_calls)
+ mock_log.error.assert_has_calls(log_error_calls)
+ assert (not self.pjlink.send_queue), 'Send queue should be empty'
+ assert (self.pjlink.priority_queue == [test_command]), 'Priority queue should have one entry'
+ assert (not mock_reset.called), 'reset_information() should not have been called'
+ assert mock_send_command.called, '_underscore_send_command() should have been called'
+
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
+ @patch.object(openlp.core.projectors.pjlink.PJLink, '_send_command')
+ @patch.object(openlp.core.projectors.pjlink, 'log')
+ def test_send_command_invalid_command_empty_queues(self, mock_log, mock_send_command, mock_reset, mock_state):
+ """
+ Test send_command with invalid command
+ """
+ # GIVEN: Test object
+ log_error_calls = [call('({ip}) send_command(): Invalid command requested - '
+ 'ignoring.'.format(ip=self.pjlink.name))]
+ log_warning_calls = []
+ log_debug_calls = []
+ mock_state.return_value = S_CONNECTED
+ self.pjlink.send_queue = []
+ self.pjlink.priority_queue = []
+
+ # WHEN: send_command with invalid command
+ self.pjlink.send_command(cmd='DONTCARE')
+
+ # THEN:
+ mock_log.debug.assert_has_calls(log_debug_calls)
+ mock_log.warning.assert_has_calls(log_warning_calls)
+ mock_log.error.assert_has_calls(log_error_calls)
+ assert (not self.pjlink.send_queue), 'Send queue should be empty'
+ assert (not self.pjlink.priority_queue), 'Priority queue should be empty'
+ assert (not mock_reset.called), 'reset_information() should not have been called'
+ assert (not mock_send_command.called), '_underscore_send_command() should not have been called'
+
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
+ @patch.object(openlp.core.projectors.pjlink.PJLink, '_send_command')
+ @patch.object(openlp.core.projectors.pjlink, 'log')
+ def test_send_command_invalid_command_normal_queue(self, mock_log, mock_send_command, mock_reset, mock_state):
+ """
+ Test _send_command with invalid command for normal queue
+ """
+ # GIVEN: Test object
+ test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX,
+ clss=self.pjlink.pjlink_class,
+ suff=PJLINK_SUFFIX)
+ log_error_calls = [call('({ip}) send_command(): Invalid command requested - '
+ 'ignoring.'.format(ip=self.pjlink.name))]
+ log_warning_calls = []
+ log_debug_calls = []
+ mock_state.return_value = S_CONNECTED
+ self.pjlink.send_queue = [test_command]
+ self.pjlink.priority_queue = []
+
+ # WHEN: send_command with invalid command
+ self.pjlink.send_command(cmd='DONTCARE')
+
+ # THEN:
+ mock_log.debug.assert_has_calls(log_debug_calls)
+ mock_log.warning.assert_has_calls(log_warning_calls)
+ mock_log.error.assert_has_calls(log_error_calls)
+ assert self.pjlink.send_queue, 'Send queue should have one entry'
+ assert (not self.pjlink.priority_queue), 'Priority queue should be empty'
+ assert (not mock_reset.called), 'reset_information() should not have been called'
+ assert mock_send_command.called, '_underscore_send_command() should have been called'
+
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
+ @patch.object(openlp.core.projectors.pjlink.PJLink, '_send_command')
+ @patch.object(openlp.core.projectors.pjlink, 'log')
+ def test_send_command_invalid_command_priority_queue(self, mock_log, mock_send_command, mock_reset, mock_state):
+ """
+ Test _send_command with invalid command for priority queue
+ """
+ # GIVEN: Test object
+ test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX,
+ clss=self.pjlink.pjlink_class,
+ suff=PJLINK_SUFFIX)
+ log_error_calls = [call('({ip}) send_command(): Invalid command requested - '
+ 'ignoring.'.format(ip=self.pjlink.name))]
+ log_warning_calls = []
+ log_debug_calls = []
+ mock_state.return_value = S_CONNECTED
+ self.pjlink.send_queue = []
+ self.pjlink.priority_queue = [test_command]
+
+ # WHEN: send_command with invalid command
+ self.pjlink.send_command(cmd='DONTCARE', priority=True)
+
+ # THEN:
+ mock_log.debug.assert_has_calls(log_debug_calls)
+ mock_log.warning.assert_has_calls(log_warning_calls)
+ mock_log.error.assert_has_calls(log_error_calls)
+ assert (not self.pjlink.send_queue), 'Send queue should be empty'
+ assert self.pjlink.priority_queue, 'Priority queue should have one entry'
+ assert (not mock_reset.called), 'reset_information() should not have been called'
+ assert mock_send_command.called, '_underscore_send_command() should have been called'
+
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
+ @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
+ @patch.object(openlp.core.projectors.pjlink.PJLink, '_send_command')
+ @patch.object(openlp.core.projectors.pjlink, 'log')
+ def test_send_command_not_connected(self, mock_log, mock_send_command, mock_reset, mock_state):
+ """
+ Test send_command when not connected
+ """
+ # GIVEN: Test object
+ log_error_calls = []
+ log_warning_calls = [call('({ip}) send_command(): Not connected - returning'.format(ip=self.pjlink.name))]
+ log_debug_calls = []
+ mock_state.return_value = S_NOT_CONNECTED
+ self.pjlink.send_queue = []
+ self.pjlink.priority_queue = []
+
+ # WHEN: send_command called when not connected
+ self.pjlink.send_command(cmd=None)
+
+ # THEN:
+ mock_log.debug.assert_has_calls(log_debug_calls)
+ mock_log.warning.assert_has_calls(log_warning_calls)
+ mock_log.error.assert_has_calls(log_error_calls)
+ assert (not self.pjlink.send_queue), 'Send queue should be empty'
+ assert (not self.pjlink.priority_queue), 'Priority queue should be empty'
+ assert mock_reset.called, 'reset_information() should have been called'
+ assert (not mock_send_command.called), '_underscore_send_command() should not have been called'
=== modified file 'tests/openlp_core/projectors/test_projector_pjlink_cmd_routing.py'
--- tests/openlp_core/projectors/test_projector_pjlink_cmd_routing.py 2019-04-28 07:10:59 +0000
+++ tests/openlp_core/projectors/test_projector_pjlink_cmd_routing.py 2019-04-28 07:10:59 +0000
@@ -59,15 +59,16 @@
# GIVEN: Test object
self.pjlink.pjlink_functions = MagicMock()
log_warning_text = [call('({ip}) get_data(): Invalid packet - '
- 'unknown command "UNKN"'.format(ip=self.pjlink.name))]
+ 'unknown command "UNKN"'.format(ip=self.pjlink.name)),
+ call('({ip}) _send_command(): Nothing to send - '
+ 'returning'.format(ip=self.pjlink.name))]
log_debug_text = [call('(___TEST_ONE___) get_data(buffer="%1UNKN=Huh?"'),
call('(___TEST_ONE___) get_data(): Checking new data "%1UNKN=Huh?"'),
call('(___TEST_ONE___) get_data() header="%1UNKN" data="Huh?"'),
call('(___TEST_ONE___) get_data() version="1" cmd="UNKN"'),
call('(___TEST_ONE___) Cleaning buffer - msg = "get_data(): '
'Invalid packet - unknown command "UNKN""'),
- call('(___TEST_ONE___) Finished cleaning buffer - 0 bytes dropped'),
- call('(___TEST_ONE___) _send_command(): Nothing to send - returning')]
+ call('(___TEST_ONE___) Finished cleaning buffer - 0 bytes dropped')]
# WHEN: get_data called with an unknown command
self.pjlink.get_data(buff='{prefix}1UNKN=Huh?'.format(prefix=PJLINK_PREFIX))
Follow ups
