forked from openlp/openlp
PJLink2 Update V03
bzr-revno: 2861
This commit is contained in:
Tomas Groth
commit
433f95d1c7
@ -48,6 +48,7 @@ Website: http://pjlink.jbmia.or.jp/english/dl_class2.html
|
||||
"""
|
||||
import logging
|
||||
from codecs import decode
|
||||
from copy import copy
|
||||
|
||||
from PyQt5 import QtCore, QtNetwork
|
||||
|
||||
@ -276,6 +277,7 @@ class PJLink(QtNetwork.QTcpSocket):
|
||||
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 @@ class PJLink(QtNetwork.QTcpSocket):
|
||||
: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 @@ class PJLink(QtNetwork.QTcpSocket):
|
||||
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 @@ class PJLink(QtNetwork.QTcpSocket):
|
||||
: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 @@ class PJLink(QtNetwork.QTcpSocket):
|
||||
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 @@ class PJLink(QtNetwork.QTcpSocket):
|
||||
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
|
||||
|
||||
@ -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 @@ class TestPJLinkBase(TestCase):
|
||||
"""
|
||||
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, '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):
|
||||
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
|
||||
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", '
|
||||
'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={})')]
|
||||
mock_state.return_value = S_NOT_CONNECTED
|
||||
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
|
||||
pjlink.send_queue = []
|
||||
pjlink.priority_queue = []
|
||||
# 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 no data and queue's empty
|
||||
pjlink.send_command(cmd='DONTCARE')
|
||||
# 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)
|
||||
assert mock_reset.called is True
|
||||
assert mock_reset.called is True
|
||||
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'
|
||||
|
||||
@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_local_send_command_no_data(self, mock_log):
|
||||
def test_send_command_duplicate_priority_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 priority queue
|
||||
"""
|
||||
# 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')]
|
||||
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]
|
||||
|
||||
pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
|
||||
pjlink.send_queue = []
|
||||
pjlink.priority_queue = []
|
||||
# WHEN: send_command called with same command in priority queue
|
||||
self.pjlink.send_command(cmd='CLSS', priority=True)
|
||||
|
||||
# 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)
|
||||
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'
|
||||
|
||||
# THEN:
|
||||
mock_log.debug.assert_has_calls(log_debug_calls)
|
||||
assert mock_timer.called is False
|
||||
@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'
|
||||
|
||||
@ -59,15 +59,16 @@ class TestPJLinkRouting(TestCase):
|
||||
# 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))
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user