raoul
/
openlp
forked from openlp/openlp
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

PJLink2 update V03

This commit is contained in:
Ken Roberts 2019-04-28 00:03:51 -07:00
parent ddd14a5fc2
commit d6cf6cf6bf
3 changed files with 653 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -48,6 +48,7 @@ Website: http://pjlink.jbmia.or.jp/english/dl_class2.html
""" """
import logging import logging
from codecs import decode from codecs import decode
from copy import copy
from PyQt5 import QtCore, QtNetwork from PyQt5 import QtCore, QtNetwork
@ -276,6 +277,7 @@ class PJLink(QtNetwork.QTcpSocket):
self.model_lamp = None # RLMP self.model_lamp = None # RLMP
self.mute = None # AVMT self.mute = None # AVMT
self.other_info = None # INFO self.other_info = None # INFO
self.pjlink_class = copy(PJLINK_CLASS)
self.pjlink_name = None # NAME self.pjlink_name = None # NAME
self.power = S_OFF # POWR self.power = S_OFF # POWR
self.serial_no = None # SNUM self.serial_no = None # SNUM
@ -629,11 +631,14 @@ class PJLink(QtNetwork.QTcpSocket):
:param salt: Optional salt for md5 hash initial authentication :param salt: Optional salt for md5 hash initial authentication
:param priority: Option to send packet now rather than queue it up :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)) log.warning('({ip}) send_command(): Not connected - returning'.format(ip=self.entry.name))
return self.reset_information() return self.reset_information()
if cmd not in PJLINK_VALID_CMD: if cmd not in PJLINK_VALID_CMD:
log.error('({ip}) send_command(): Invalid command requested - ignoring.'.format(ip=self.entry.name)) 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 return
log.debug('({ip}) send_command(): Building cmd="{command}" opts="{data}"{salt}'.format(ip=self.entry.name, log.debug('({ip}) send_command(): Building cmd="{command}" opts="{data}"{salt}'.format(ip=self.entry.name,
command=cmd, command=cmd,
@ -649,9 +654,9 @@ class PJLink(QtNetwork.QTcpSocket):
options=opts, options=opts,
suffix=PJLINK_SUFFIX) suffix=PJLINK_SUFFIX)
if out in self.priority_queue: 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: 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: else:
if priority: if priority:
log.debug('({ip}) send_command(): Adding to priority queue'.format(ip=self.entry.name)) log.debug('({ip}) send_command(): Adding to priority queue'.format(ip=self.entry.name))
@ -672,7 +677,8 @@ class PJLink(QtNetwork.QTcpSocket):
:param utf8: Send as UTF-8 string otherwise send as ASCII string :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: 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 return
log.debug('({ip}) _send_command(data="{data}")'.format(ip=self.entry.name, log.debug('({ip}) _send_command(data="{data}")'.format(ip=self.entry.name,
data=data.strip() if data else data)) data=data.strip() if data else data))
@ -684,7 +690,7 @@ class PJLink(QtNetwork.QTcpSocket):
log.debug('({ip}) _send_command(): Connection status: {data}'.format(ip=self.entry.name, log.debug('({ip}) _send_command(): Connection status: {data}'.format(ip=self.entry.name,
data=conn_state)) data=conn_state))
if QSOCKET_STATE[self.state()] != S_CONNECTED: 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 self.send_busy = False
return self.disconnect_from_host() return self.disconnect_from_host()
if data and data not in self.priority_queue: if data and data not in self.priority_queue:
@ -707,7 +713,7 @@ class PJLink(QtNetwork.QTcpSocket):
log.debug('({ip}) _send_command(): Getting normal queued packet'.format(ip=self.entry.name)) log.debug('({ip}) _send_command(): Getting normal queued packet'.format(ip=self.entry.name))
else: else:
# No data to send # 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 self.send_busy = False
return return
self.send_busy = True self.send_busy = True

680
tests/openlp_core/projectors/test_projector_pjlink_base_02.py
View File

@ -22,11 +22,12 @@
""" """
Package to test the openlp.core.projectors.pjlink base package. 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 from unittest.mock import call, patch
import openlp.core.projectors.pjlink 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.db import Projector
from openlp.core.projectors.pjlink import PJLink from openlp.core.projectors.pjlink import PJLink
from tests.resources.projector.data import TEST1_DATA from tests.resources.projector.data import TEST1_DATA
@ -36,67 +37,658 @@ class TestPJLinkBase(TestCase):
""" """
Tests for the PJLink module Tests for the PJLink module
""" """
@skip('Needs update to new setup') 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
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(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, 'state')
@patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information') @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
@patch.object(openlp.core.projectors.pjlink.PJLink, '_send_command') @patch.object(openlp.core.projectors.pjlink.PJLink, '_send_command')
@patch.object(openlp.core.projectors.pjlink, 'log') @patch.object(openlp.core.projectors.pjlink, 'log')
def test_send_command_no_data(self, mock_log, mock_send_command, mock_reset, mock_state): def test_send_command_add_normal_command(self, mock_log, mock_send_command, mock_reset, mock_state):
""" """
Test _send_command with no data to send Test send_command adding normal queue item
""" """
# GIVEN: Test object # GIVEN: Test object
log_warning_calls = [call('({ip}) send_command(): Not connected - returning'.format(ip=TEST1_DATA['name']))] 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
log_debug_calls = [call('PJlink(projector="< Projector(id="None", ip="111.111.111.111", port="1111", ' # Patch here since pjlink does not have priority or send queue's until instantiated
'mac_adx="11:11:11:11:11:11", pin="1111", name="___TEST_ONE___", ' with patch.object(self.pjlink, 'send_queue') as mock_send, \
'location="location one", notes="notes one", pjlink_name="None", ' patch.object(self.pjlink, 'priority_queue') as mock_priority:
'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={})')]
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 # WHEN: send_command called with valid normal command
pjlink.send_command(cmd='DONTCARE') 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: # THEN:
mock_log.debug.assert_has_calls(log_debug_calls) mock_log.debug.assert_has_calls(log_debug_calls)
mock_log.warning.assert_has_calls(log_warning_calls) mock_log.warning.assert_has_calls(log_warning_calls)
assert mock_reset.called is True mock_log.error.assert_has_calls(log_error_calls)
assert mock_reset.called is True 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'
@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') @patch.object(openlp.core.projectors.pjlink, 'log')
def test_local_send_command_no_data(self, mock_log): def test_send_command_duplicate_normal_command(self, mock_log, mock_send_command, mock_reset, mock_state):
""" """
Test _send_command with no data to send Test send_command with duplicate item for normal queue
""" """
# GIVEN: Test object # GIVEN: Test object
log_debug_calls = [call('PJlink(projector="< Projector(id="None", ip="111.111.111.111", port="1111", ' test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX,
'mac_adx="11:11:11:11:11:11", pin="1111", name="___TEST_ONE___", ' clss=self.pjlink.pjlink_class,
'location="location one", notes="notes one", pjlink_name="None", ' suff=PJLINK_SUFFIX)
'pjlink_class="None", manufacturer="None", model="None", ' log_error_calls = []
'serial_no="Serial Number 1", other="None", sources="None", source_list="[]", ' log_warning_calls = [call('({ip}) send_command(): Already in normal queue - '
'model_filter="Filter type 1", model_lamp="Lamp type 1", ' 'skipping'.format(ip=self.pjlink.name))]
'sw_version="Version 1") >", args="()" kwargs="{\'no_poll\': True}")'), log_debug_calls = [call('({ip}) send_command(): Building cmd="CLSS" opts="?"'.format(ip=self.pjlink.name))]
call('PJlinkCommands(args=() kwargs={})'), mock_state.return_value = S_CONNECTED
call('(___TEST_ONE___) reset_information() connect status is S_NOT_CONNECTED'), self.pjlink.send_queue = [test_command]
call('(___TEST_ONE___) _send_command(): Nothing to send - returning')] self.pjlink.priority_queue = []
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True) # WHEN: send_command called with same command in normal queue
pjlink.send_queue = [] self.pjlink.send_command(cmd='CLSS')
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: # THEN:
mock_log.debug.assert_has_calls(log_debug_calls) mock_log.debug.assert_has_calls(log_debug_calls)
assert mock_timer.called is False 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'

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

@ -59,15 +59,16 @@ class TestPJLinkRouting(TestCase):
# GIVEN: Test object # GIVEN: Test object
self.pjlink.pjlink_functions = MagicMock() self.pjlink.pjlink_functions = MagicMock()
log_warning_text = [call('({ip}) get_data(): Invalid packet - ' 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?"'), 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(): Checking new data "%1UNKN=Huh?"'),
call('(___TEST_ONE___) get_data() header="%1UNKN" data="Huh?"'), call('(___TEST_ONE___) get_data() header="%1UNKN" data="Huh?"'),
call('(___TEST_ONE___) get_data() version="1" cmd="UNKN"'), call('(___TEST_ONE___) get_data() version="1" cmd="UNKN"'),
call('(___TEST_ONE___) Cleaning buffer - msg = "get_data(): ' call('(___TEST_ONE___) Cleaning buffer - msg = "get_data(): '
'Invalid packet - unknown command "UNKN""'), 'Invalid packet - unknown command "UNKN""'),
call('(___TEST_ONE___) Finished cleaning buffer - 0 bytes dropped'), call('(___TEST_ONE___) Finished cleaning buffer - 0 bytes dropped')]
call('(___TEST_ONE___) _send_command(): Nothing to send - returning')]
# WHEN: get_data called with an unknown command # WHEN: get_data called with an unknown command
self.pjlink.get_data(buff='{prefix}1UNKN=Huh?'.format(prefix=PJLINK_PREFIX)) self.pjlink.get_data(buff='{prefix}1UNKN=Huh?'.format(prefix=PJLINK_PREFIX))