← Back to branch summary

~tomasgroth/openlp/portable-path

« back to all changes in this revision

Viewing changes to tests/openlp_core/projectors/test_projector_pjlink_base_02.py

  • Committer: Tomas Groth
  • Date: 2019-04-30 19:02:42 UTC
  • mfrom: (2829.2.32 openlp)
  • Revision ID: tomasgroth@yahoo.dk-20190430190242-6zwjk8724tyux70m
trunk

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/openlp_core/projectors/test_projector_pjlink_base_02.py
4
4
###############################################################################
5
5
# OpenLP - Open Source Lyrics Projection                                      #
6
6
# --------------------------------------------------------------------------- #
7
 
# Copyright (c) 2008-2015 OpenLP Developers                                   #
 
7
# Copyright (c) 2008-2019 OpenLP Developers                                   #
8
8
# --------------------------------------------------------------------------- #
9
9
# This program is free software; you can redistribute it and/or modify it     #
10
10
# under the terms of the GNU General Public License as published by the Free  #
26
26
from unittest.mock import call, patch
27
27
 
28
28
import openlp.core.projectors.pjlink
29
 
 
30
 
from openlp.core.projectors.constants import S_NOT_CONNECTED
 
29
from openlp.core.projectors.constants import E_NETWORK, PJLINK_PREFIX, PJLINK_SUFFIX, QSOCKET_STATE, \
 
30
    S_CONNECTED, S_NOT_CONNECTED
31
31
from openlp.core.projectors.db import Projector
32
32
from openlp.core.projectors.pjlink import PJLink
33
33
from tests.resources.projector.data import TEST1_DATA
37
37
    """
38
38
    Tests for the PJLink module
39
39
    """
40
 
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
41
 
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
42
 
    @patch.object(openlp.core.projectors.pjlink.PJLink, '_send_command')
43
 
    @patch.object(openlp.core.projectors.pjlink, 'log')
44
 
    def test_send_command_no_data(self, mock_log, mock_send_command, mock_reset, mock_state):
45
 
        """
46
 
        Test _send_command with no data to send
47
 
        """
48
 
        # GIVEN: Test object
49
 
        log_warning_calls = [call('({ip}) send_command(): Not connected - returning'.format(ip=TEST1_DATA['name']))]
50
 
 
51
 
        log_debug_calls = [call('PJlink(projector="< Projector(id="None", ip="111.111.111.111", port="1111", '
52
 
                                'mac_adx="11:11:11:11:11:11", pin="1111", name="___TEST_ONE___", '
53
 
                                'location="location one", notes="notes one", pjlink_name="None", '
54
 
                                'pjlink_class="None", manufacturer="None", model="None", '
55
 
                                'serial_no="Serial Number 1", other="None", sources="None", source_list="[]", '
56
 
                                'model_filter="Filter type 1", model_lamp="Lamp type 1", '
57
 
                                'sw_version="Version 1") >", args="()" kwargs="{\'no_poll\': True}")'),
58
 
                           call('PJlinkCommands(args=() kwargs={})')]
 
40
    def setUp(self):
 
41
        """
 
42
        Initialize test state(s)
 
43
        """
 
44
        # Default PJLink instance for tests
 
45
        self.pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
 
46
 
 
47
    def tearDown(self):
 
48
        """
 
49
        Cleanup test state(s)
 
50
        """
 
51
        del(self.pjlink)
 
52
 
 
53
    # ------------ Test PJLink._underscore_send_command ----------
 
54
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'change_status')
 
55
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'write')
 
56
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'disconnect_from_host')
 
57
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
 
58
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
 
59
    @patch.object(openlp.core.projectors.pjlink, 'log')
 
60
    def test_local_send_command_network_error(self, mock_log, mock_reset, mock_state, mock_disconnect, mock_write,
 
61
                                              mock_change_status):
 
62
        """
 
63
        Test _underscore_send_command when possible network error occured
 
64
        """
 
65
        # GIVEN: Test object
 
66
        test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX,
 
67
                                                           clss=self.pjlink.pjlink_class,
 
68
                                                           suff=PJLINK_SUFFIX)
 
69
        log_error_calls = []
 
70
        log_warning_calls = [call('({ip}) _send_command(): -1 received - '
 
71
                                  'disconnecting from host'.format(ip=self.pjlink.name))]
 
72
        log_debug_calls = [call('({ip}) _send_command(data="None")'.format(ip=self.pjlink.name)),
 
73
                           call('({ip}) _send_command(): priority_queue: []'.format(ip=self.pjlink.name)),
 
74
                           call("({ip}) _send_command(): send_queue: ['{data}\\r']".format(ip=self.pjlink.name,
 
75
                                                                                           data=test_command.strip())),
 
76
                           call('({ip}) _send_command(): Connection status: S_CONNECTED'.format(ip=self.pjlink.name)),
 
77
                           call('({ip}) _send_command(): Getting normal queued packet'.format(ip=self.pjlink.name)),
 
78
                           call('({ip}) _send_command(): Sending "{data}"'.format(ip=self.pjlink.name,
 
79
                                                                                  data=test_command.strip()))
 
80
                           ]
 
81
 
 
82
        mock_state.return_value = QSOCKET_STATE[S_CONNECTED]
 
83
        mock_write.return_value = -1
 
84
        self.pjlink.send_queue = [test_command]
 
85
        self.pjlink.priority_queue = []
 
86
 
 
87
        # WHEN: _send_command called with no data and queue's emtpy
 
88
        # Patch some attributes here since they are not available until after instantiation
 
89
        with patch.object(self.pjlink, 'socket_timer') as mock_timer, \
 
90
                patch.object(self.pjlink, 'waitForBytesWritten') as mock_waitBytes:
 
91
            mock_waitBytes.return_value = True
 
92
            self.pjlink._send_command()
 
93
 
 
94
            # THEN:
 
95
            mock_log.error.assert_has_calls(log_error_calls)
 
96
            mock_log.warning.assert_has_calls(log_warning_calls)
 
97
            mock_log.debug.assert_has_calls(log_debug_calls)
 
98
            mock_change_status.called_with(E_NETWORK, 'Error while sending data to projector')
 
99
            assert (not self.pjlink.send_queue), 'Send queue should be empty'
 
100
            assert (not self.pjlink.priority_queue), 'Priority queue should be empty'
 
101
            assert mock_timer.start.called, 'Timer should have been called'
 
102
            assert (not mock_reset.called), 'reset_information() should not should have been called'
 
103
            assert mock_disconnect.called, 'disconnect_from_host() should have been called'
 
104
            assert self.pjlink.send_busy, 'send_busy should be True'
 
105
 
 
106
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
 
107
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
 
108
    @patch.object(openlp.core.projectors.pjlink, 'log')
 
109
    def test_local_send_command_no_data(self, mock_log, mock_reset, mock_state):
 
110
        """
 
111
        Test _underscore_send_command with no data to send
 
112
        """
 
113
        # GIVEN: Test object
 
114
        log_error_calls = []
 
115
        log_warning_calls = [call('({ip}) _send_command(): Nothing to send - returning'.format(ip=self.pjlink.name))]
 
116
        log_debug_calls = []
 
117
        mock_state.return_value = S_CONNECTED
 
118
        self.pjlink.send_queue = []
 
119
        self.pjlink.priority_queue = []
 
120
 
 
121
        # WHEN: _send_command called with no data and queue's emtpy
 
122
        # Patch some attributes here since they are not available until after instantiation
 
123
        with patch.object(self.pjlink, 'socket_timer') as mock_timer:
 
124
            self.pjlink._send_command()
 
125
 
 
126
            # THEN:
 
127
            mock_log.error.assert_has_calls(log_error_calls)
 
128
            mock_log.warning.assert_has_calls(log_warning_calls)
 
129
            mock_log.debug.assert_has_calls(log_debug_calls)
 
130
            assert (not self.pjlink.send_queue), 'Send queue should be empty'
 
131
            assert (not self.pjlink.priority_queue), 'Priority queue should be empty'
 
132
            assert (not mock_timer.called), 'Timer should not have been called'
 
133
            assert (not mock_reset.called), 'reset_information() should not have been called'
 
134
 
 
135
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
 
136
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
 
137
    @patch.object(openlp.core.projectors.pjlink, 'log')
 
138
    def test_local_send_command_no_data_queue_check(self, mock_log, mock_reset, mock_state):
 
139
        """
 
140
        Test _underscore_send_command last queue length check
 
141
        """
 
142
        # GIVEN: Test object
 
143
        log_error_calls = []
 
144
        log_warning_calls = [call('({ip}) _send_command(): No data to send'.format(ip=self.pjlink.name))]
 
145
        log_debug_calls = []
 
146
        mock_state.return_value = QSOCKET_STATE[S_CONNECTED]
 
147
        self.pjlink.priority_queue = []
 
148
 
 
149
        # WHEN: _send_command called with no data and queue's emtpy
 
150
        # Patch some attributes here since they are not available until after instantiation
 
151
        with patch.object(self.pjlink, 'socket_timer') as mock_timer, \
 
152
                patch.object(self.pjlink, 'send_queue') as mock_queue:
 
153
            # Unlikely case of send_queue not really empty, but len(send_queue) returns 0
 
154
            mock_queue.return_value = ['test']
 
155
            mock_queue.__len__.return_value = 0
 
156
            self.pjlink._send_command(data=None)
 
157
 
 
158
            # THEN:
 
159
            mock_log.error.assert_has_calls(log_error_calls)
 
160
            mock_log.warning.assert_has_calls(log_warning_calls)
 
161
            mock_log.debug.assert_has_calls(log_debug_calls)
 
162
            assert (not self.pjlink.priority_queue), 'Priority queue should be empty'
 
163
            assert (not mock_timer.called), 'Timer should not have been called'
 
164
            assert (not mock_reset.called), 'reset_information() should not have been called'
 
165
 
 
166
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'write')
 
167
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'disconnect_from_host')
 
168
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
 
169
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
 
170
    @patch.object(openlp.core.projectors.pjlink, 'log')
 
171
    def test_local_send_command_normal_send(self, mock_log, mock_reset, mock_state, mock_disconnect, mock_write):
 
172
        """
 
173
        Test _underscore_send_command using normal queue
 
174
        """
 
175
        # GIVEN: Test object
 
176
        test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX,
 
177
                                                           clss=self.pjlink.pjlink_class,
 
178
                                                           suff=PJLINK_SUFFIX)
 
179
        log_error_calls = []
 
180
        log_warning_calls = []
 
181
        log_debug_calls = [call('({ip}) _send_command(data="None")'.format(ip=self.pjlink.name)),
 
182
                           call('({ip}) _send_command(): priority_queue: []'.format(ip=self.pjlink.name)),
 
183
                           call("({ip}) _send_command(): send_queue: ['{data}\\r']".format(ip=self.pjlink.name,
 
184
                                                                                           data=test_command.strip())),
 
185
                           call('({ip}) _send_command(): Connection status: S_CONNECTED'.format(ip=self.pjlink.name)),
 
186
                           call('({ip}) _send_command(): Getting normal queued packet'.format(ip=self.pjlink.name)),
 
187
                           call('({ip}) _send_command(): Sending "{data}"'.format(ip=self.pjlink.name,
 
188
                                                                                  data=test_command.strip()))
 
189
                           ]
 
190
 
 
191
        mock_state.return_value = QSOCKET_STATE[S_CONNECTED]
 
192
        mock_write.return_value = len(test_command)
 
193
        self.pjlink.send_queue = [test_command]
 
194
        self.pjlink.priority_queue = []
 
195
 
 
196
        # WHEN: _send_command called with no data and queue's emtpy
 
197
        # Patch some attributes here since they are not available until after instantiation
 
198
        with patch.object(self.pjlink, 'socket_timer') as mock_timer, \
 
199
                patch.object(self.pjlink, 'waitForBytesWritten') as mock_waitBytes:
 
200
            mock_waitBytes.return_value = True
 
201
            self.pjlink._send_command()
 
202
 
 
203
            # THEN:
 
204
            mock_log.error.assert_has_calls(log_error_calls)
 
205
            mock_log.warning.assert_has_calls(log_warning_calls)
 
206
            mock_log.debug.assert_has_calls(log_debug_calls)
 
207
            assert (not self.pjlink.send_queue), 'Send queue should be empty'
 
208
            assert (not self.pjlink.priority_queue), 'Priority queue should be empty'
 
209
            assert mock_timer.start.called, 'Timer should have been called'
 
210
            assert (not mock_reset.called), 'reset_information() should not have been called'
 
211
            assert (not mock_disconnect.called), 'disconnect_from_host() should not have been called'
 
212
            assert self.pjlink.send_busy, 'send_busy flag should be True'
 
213
 
 
214
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'disconnect_from_host')
 
215
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
 
216
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
 
217
    @patch.object(openlp.core.projectors.pjlink, 'log')
 
218
    def test_local_send_command_not_connected(self, mock_log, mock_reset, mock_state, mock_disconnect):
 
219
        """
 
220
        Test _underscore_send_command when not connected
 
221
        """
 
222
        # GIVEN: Test object
 
223
        test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX,
 
224
                                                           clss=self.pjlink.pjlink_class,
 
225
                                                           suff=PJLINK_SUFFIX)
 
226
        log_error_calls = []
 
227
        log_warning_calls = [call('({ip}) _send_command() Not connected - abort'.format(ip=self.pjlink.name))]
 
228
        log_debug_calls = [call('({ip}) _send_command(data="None")'.format(ip=self.pjlink.name)),
 
229
                           call('({ip}) _send_command(): priority_queue: []'.format(ip=self.pjlink.name)),
 
230
                           call("({ip}) _send_command(): send_queue: ['%1CLSS ?\\r']".format(ip=self.pjlink.name)),
 
231
                           call('({ip}) _send_command(): Connection status: S_OK'.format(ip=self.pjlink.name))]
59
232
        mock_state.return_value = S_NOT_CONNECTED
60
 
        pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
61
 
        pjlink.send_queue = []
62
 
        pjlink.priority_queue = []
63
 
 
64
 
        # WHEN: _send_command called with no data and queue's empty
65
 
        pjlink.send_command(cmd='DONTCARE')
66
 
 
67
 
        # THEN:
68
 
        mock_log.debug.assert_has_calls(log_debug_calls)
69
 
        mock_log.warning.assert_has_calls(log_warning_calls)
70
 
        assert mock_reset.called is True
71
 
        assert mock_reset.called is True
72
 
 
73
 
    @patch.object(openlp.core.projectors.pjlink, 'log')
74
 
    def test_local_send_command_no_data(self, mock_log):
75
 
        """
76
 
        Test _send_command with no data to send
77
 
        """
78
 
        # GIVEN: Test object
79
 
        log_debug_calls = [call('PJlink(projector="< Projector(id="None", ip="111.111.111.111", port="1111", '
80
 
                                'mac_adx="11:11:11:11:11:11", pin="1111", name="___TEST_ONE___", '
81
 
                                'location="location one", notes="notes one", pjlink_name="None", '
82
 
                                'pjlink_class="None", manufacturer="None", model="None", '
83
 
                                'serial_no="Serial Number 1", other="None", sources="None", source_list="[]", '
84
 
                                'model_filter="Filter type 1", model_lamp="Lamp type 1", '
85
 
                                'sw_version="Version 1") >", args="()" kwargs="{\'no_poll\': True}")'),
86
 
                           call('PJlinkCommands(args=() kwargs={})'),
87
 
                           call('(___TEST_ONE___) reset_information() connect status is S_NOT_CONNECTED'),
88
 
                           call('(___TEST_ONE___) _send_command(): Nothing to send - returning')]
89
 
 
90
 
        pjlink = PJLink(Projector(**TEST1_DATA), no_poll=True)
91
 
        pjlink.send_queue = []
92
 
        pjlink.priority_queue = []
 
233
        self.pjlink.send_queue = [test_command]
 
234
        self.pjlink.priority_queue = []
93
235
 
94
236
        # WHEN: _send_command called with no data and queue's emtpy
95
237
        # Patch here since pjlink does not have socket_timer until after instantiation
96
 
        with patch.object(pjlink, 'socket_timer') as mock_timer:
97
 
            pjlink._send_command(data=None, utf8=False)
98
 
 
99
 
            # THEN:
100
 
            mock_log.debug.assert_has_calls(log_debug_calls)
101
 
            assert mock_timer.called is False
 
238
        with patch.object(self.pjlink, 'socket_timer') as mock_timer:
 
239
            self.pjlink._send_command()
 
240
 
 
241
            # THEN:
 
242
            mock_log.error.assert_has_calls(log_error_calls)
 
243
            mock_log.warning.assert_has_calls(log_warning_calls)
 
244
            mock_log.debug.assert_has_calls(log_debug_calls)
 
245
            assert (self.pjlink.send_queue == [test_command]), 'Send queue should have one entry'
 
246
            assert (not self.pjlink.priority_queue), 'Priority queue should be empty'
 
247
            assert (not mock_timer.called), 'Timer should not have been called'
 
248
            assert (not mock_reset.called), 'reset_information() should not have been called'
 
249
            assert mock_disconnect.called, 'disconnect_from_host() should have been called'
 
250
            assert (not self.pjlink.send_busy), 'send_busy flag should be False'
 
251
 
 
252
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'write')
 
253
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'disconnect_from_host')
 
254
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
 
255
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
 
256
    @patch.object(openlp.core.projectors.pjlink, 'log')
 
257
    def test_local_send_command_priority_send(self, mock_log, mock_reset, mock_state, mock_disconnect, mock_write):
 
258
        """
 
259
        Test _underscore_send_command with priority queue
 
260
        """
 
261
        # GIVEN: Test object
 
262
        test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX,
 
263
                                                           clss=self.pjlink.pjlink_class,
 
264
                                                           suff=PJLINK_SUFFIX)
 
265
        log_error_calls = []
 
266
        log_warning_calls = []
 
267
        log_debug_calls = [call('({ip}) _send_command(data="{data}")'.format(ip=self.pjlink.name,
 
268
                                                                             data=test_command.strip())),
 
269
                           call('({ip}) _send_command(): priority_queue: []'.format(ip=self.pjlink.name)),
 
270
                           call('({ip}) _send_command(): send_queue: []'.format(ip=self.pjlink.name)),
 
271
                           call('({ip}) _send_command(): Connection status: S_CONNECTED'.format(ip=self.pjlink.name)),
 
272
                           call('({ip}) _send_command(): Priority packet - '
 
273
                                'adding to priority queue'.format(ip=self.pjlink.name)),
 
274
                           call('({ip}) _send_command(): Getting priority queued packet'.format(ip=self.pjlink.name)),
 
275
                           call('({ip}) _send_command(): Sending "{data}"'.format(ip=self.pjlink.name,
 
276
                                                                                  data=test_command.strip()))
 
277
                           ]
 
278
 
 
279
        mock_state.return_value = QSOCKET_STATE[S_CONNECTED]
 
280
        mock_write.return_value = len(test_command)
 
281
        self.pjlink.send_queue = []
 
282
        self.pjlink.priority_queue = []
 
283
 
 
284
        # WHEN: _send_command called with no data and queue's emtpy
 
285
        # Patch some attributes here since they are not available until after instantiation
 
286
        with patch.object(self.pjlink, 'socket_timer') as mock_timer, \
 
287
                patch.object(self.pjlink, 'waitForBytesWritten') as mock_waitBytes:
 
288
            mock_waitBytes.return_value = True
 
289
            self.pjlink._send_command(data=test_command)
 
290
 
 
291
            # THEN:
 
292
            mock_log.error.assert_has_calls(log_error_calls)
 
293
            mock_log.warning.assert_has_calls(log_warning_calls)
 
294
            mock_log.debug.assert_has_calls(log_debug_calls)
 
295
            assert (not self.pjlink.send_queue), 'Send queue should be empty'
 
296
            assert (not self.pjlink.priority_queue), 'Priority queue should be empty'
 
297
            assert mock_timer.start.called, 'Timer should have been called'
 
298
            assert (not mock_reset.called), 'reset_information() should not have been called'
 
299
            assert (not mock_disconnect.called), 'disconnect_from_host() should not have been called'
 
300
            assert self.pjlink.send_busy, 'send_busy flag should be True'
 
301
 
 
302
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'write')
 
303
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'disconnect_from_host')
 
304
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
 
305
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
 
306
    @patch.object(openlp.core.projectors.pjlink, 'log')
 
307
    def test_local_send_command_priority_send_with_normal_queue(self, mock_log, mock_reset, mock_state,
 
308
                                                                mock_disconnect, mock_write):
 
309
        """
 
310
        Test _underscore_send_command with priority queue when normal queue active
 
311
        """
 
312
        # GIVEN: Test object
 
313
        test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX,
 
314
                                                           clss=self.pjlink.pjlink_class,
 
315
                                                           suff=PJLINK_SUFFIX)
 
316
        log_error_calls = []
 
317
        log_warning_calls = []
 
318
        log_debug_calls = [call('({ip}) _send_command(data="{data}")'.format(ip=self.pjlink.name,
 
319
                                                                             data=test_command.strip())),
 
320
                           call('({ip}) _send_command(): priority_queue: []'.format(ip=self.pjlink.name)),
 
321
                           call("({ip}) _send_command(): send_queue: ['{data}\\r']".format(ip=self.pjlink.name,
 
322
                                                                                           data=test_command.strip())),
 
323
                           call('({ip}) _send_command(): Connection status: S_CONNECTED'.format(ip=self.pjlink.name)),
 
324
                           call('({ip}) _send_command(): Priority packet - '
 
325
                                'adding to priority queue'.format(ip=self.pjlink.name)),
 
326
                           call('({ip}) _send_command(): Getting priority queued packet'.format(ip=self.pjlink.name)),
 
327
                           call('({ip}) _send_command(): Sending "{data}"'.format(ip=self.pjlink.name,
 
328
                                                                                  data=test_command.strip()))
 
329
                           ]
 
330
 
 
331
        mock_state.return_value = QSOCKET_STATE[S_CONNECTED]
 
332
        mock_write.return_value = len(test_command)
 
333
        self.pjlink.send_queue = [test_command]
 
334
        self.pjlink.priority_queue = []
 
335
 
 
336
        # WHEN: _send_command called with no data and queue's emtpy
 
337
        # Patch some attributes here since they are not available until after instantiation
 
338
        with patch.object(self.pjlink, 'socket_timer') as mock_timer, \
 
339
                patch.object(self.pjlink, 'waitForBytesWritten') as mock_waitBytes:
 
340
            mock_waitBytes.return_value = True
 
341
            self.pjlink._send_command(data=test_command)
 
342
 
 
343
            # THEN:
 
344
            mock_log.error.assert_has_calls(log_error_calls)
 
345
            mock_log.warning.assert_has_calls(log_warning_calls)
 
346
            mock_log.debug.assert_has_calls(log_debug_calls)
 
347
            assert self.pjlink.send_queue, 'Send queue should have one entry'
 
348
            assert (not self.pjlink.priority_queue), 'Priority queue should be empty'
 
349
            assert mock_timer.start.called, 'Timer should have been called'
 
350
            assert (not mock_reset.called), 'reset_information() should not have been called'
 
351
            assert (not mock_disconnect.called), 'disconnect_from_host() should not have been called'
 
352
            assert self.pjlink.send_busy, 'send_busy flag should be True'
 
353
 
 
354
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
 
355
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
 
356
    @patch.object(openlp.core.projectors.pjlink, 'log')
 
357
    def test_local_send_command_send_busy_normal_queue(self, mock_log, mock_reset, mock_state):
 
358
        """
 
359
        Test _underscore_send_command send_busy flag with normal queue
 
360
        """
 
361
        # GIVEN: Test object
 
362
        test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX,
 
363
                                                           clss=self.pjlink.pjlink_class,
 
364
                                                           suff=PJLINK_SUFFIX)
 
365
        log_error_calls = []
 
366
        log_warning_calls = []
 
367
        log_debug_calls = [call('({ip}) _send_command(data="None")'.format(ip=self.pjlink.name)),
 
368
                           call('({ip}) _send_command(): priority_queue: []'.format(ip=self.pjlink.name)),
 
369
                           call("({ip}) _send_command(): send_queue: ['{data}\\r']".format(ip=self.pjlink.name,
 
370
                                                                                           data=test_command.strip())),
 
371
                           call('({ip}) _send_command(): Connection status: S_CONNECTED'.format(ip=self.pjlink.name)),
 
372
                           call('({ip}) _send_command(): Still busy, returning'.format(ip=self.pjlink.name)),
 
373
                           call('({ip}) _send_command(): Priority queue = []'.format(ip=self.pjlink.name)),
 
374
                           call("({ip}) _send_command(): Normal queue = "
 
375
                                "['{data}\\r']".format(ip=self.pjlink.name, data=test_command.strip()))]
 
376
 
 
377
        mock_state.return_value = QSOCKET_STATE[S_CONNECTED]
 
378
        self.pjlink.send_busy = True
 
379
        self.pjlink.send_queue = [test_command]
 
380
        self.pjlink.priority_queue = []
 
381
 
 
382
        # WHEN: _send_command called with no data and queue's emtpy
 
383
        # Patch some attributes here since they are not available until after instantiation
 
384
        with patch.object(self.pjlink, 'socket_timer') as mock_timer:
 
385
            self.pjlink._send_command()
 
386
 
 
387
            # THEN:
 
388
            mock_log.error.assert_has_calls(log_error_calls)
 
389
            mock_log.warning.assert_has_calls(log_warning_calls)
 
390
            mock_log.debug.assert_has_calls(log_debug_calls)
 
391
            assert self.pjlink.send_queue, 'Send queue should have one entry'
 
392
            assert (not self.pjlink.priority_queue), 'Priority queue should be empty'
 
393
            assert (not mock_timer.start.called), 'Timer should not have been called'
 
394
            assert (not mock_reset.called), 'reset_information() should not have been called'
 
395
            assert self.pjlink.send_busy, 'send_busy flag should be True'
 
396
 
 
397
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
 
398
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
 
399
    @patch.object(openlp.core.projectors.pjlink, 'log')
 
400
    def test_local_send_command_send_busy_priority_queue(self, mock_log, mock_reset, mock_state):
 
401
        """
 
402
        Test _underscore_send_command send_busy flag with priority queue
 
403
        """
 
404
        # GIVEN: Test object
 
405
        test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX,
 
406
                                                           clss=self.pjlink.pjlink_class,
 
407
                                                           suff=PJLINK_SUFFIX)
 
408
        log_error_calls = []
 
409
        log_warning_calls = []
 
410
        log_debug_calls = [call('({ip}) _send_command(data="None")'.format(ip=self.pjlink.name)),
 
411
                           call("({ip}) _send_command(): priority_queue: "
 
412
                                "['{data}\\r']".format(ip=self.pjlink.name,
 
413
                                                       data=test_command.strip())),
 
414
                           call('({ip}) _send_command(): send_queue: []'.format(ip=self.pjlink.name)),
 
415
                           call('({ip}) _send_command(): Connection status: S_CONNECTED'.format(ip=self.pjlink.name)),
 
416
                           call('({ip}) _send_command(): Still busy, returning'.format(ip=self.pjlink.name)),
 
417
                           call("({ip}) _send_command(): Priority queue = "
 
418
                                "['{data}\\r']".format(ip=self.pjlink.name, data=test_command.strip())),
 
419
                           call('({ip}) _send_command(): Normal queue = []'.format(ip=self.pjlink.name))
 
420
                           ]
 
421
 
 
422
        mock_state.return_value = QSOCKET_STATE[S_CONNECTED]
 
423
        self.pjlink.send_busy = True
 
424
        self.pjlink.send_queue = []
 
425
        self.pjlink.priority_queue = [test_command]
 
426
 
 
427
        # WHEN: _send_command called with no data and queue's emtpy
 
428
        # Patch some attributes here since they are not available until after instantiation
 
429
        with patch.object(self.pjlink, 'socket_timer') as mock_timer:
 
430
            self.pjlink._send_command()
 
431
 
 
432
            # THEN:
 
433
            mock_log.error.assert_has_calls(log_error_calls)
 
434
            mock_log.warning.assert_has_calls(log_warning_calls)
 
435
            mock_log.debug.assert_has_calls(log_debug_calls)
 
436
            assert (not self.pjlink.send_queue), 'Send queue should be empty'
 
437
            assert self.pjlink.priority_queue, 'Priority queue should have one entry'
 
438
            assert (not mock_timer.start.called), 'Timer should not have been called'
 
439
            assert (not mock_reset.called), 'reset_information() should not have been called'
 
440
            assert self.pjlink.send_busy, 'send_busy flag should be True'
 
441
 
 
442
    # ------------ Test PJLink.send_command ----------
 
443
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
 
444
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
 
445
    @patch.object(openlp.core.projectors.pjlink.PJLink, '_send_command')
 
446
    @patch.object(openlp.core.projectors.pjlink, 'log')
 
447
    def test_send_command_add_normal_command(self, mock_log, mock_send_command, mock_reset, mock_state):
 
448
        """
 
449
        Test send_command adding normal queue item
 
450
        """
 
451
        # GIVEN: Test object
 
452
        test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX,
 
453
                                                           clss=self.pjlink.pjlink_class,
 
454
                                                           suff=PJLINK_SUFFIX)
 
455
        log_error_calls = []
 
456
        log_warning_calls = []
 
457
        log_debug_calls = [call('({ip}) send_command(): Building cmd="CLSS" opts="?"'.format(ip=self.pjlink.name)),
 
458
                           call('({ip}) send_command(): Adding to normal queue'.format(ip=self.pjlink.name))]
 
459
        mock_state.return_value = S_CONNECTED
 
460
 
 
461
        # Patch here since pjlink does not have priority or send queue's until instantiated
 
462
        with patch.object(self.pjlink, 'send_queue') as mock_send, \
 
463
                patch.object(self.pjlink, 'priority_queue') as mock_priority:
 
464
 
 
465
            # WHEN: send_command called with valid normal command
 
466
            self.pjlink.send_command(cmd='CLSS')
 
467
 
 
468
            # THEN:
 
469
            mock_send.append.called_with(test_command)
 
470
            mock_priority.append.called is False
 
471
            mock_log.debug.assert_has_calls(log_debug_calls)
 
472
            mock_log.warning.assert_has_calls(log_warning_calls)
 
473
            mock_log.error.assert_has_calls(log_error_calls)
 
474
            assert (not mock_reset.called), 'reset_information() should not have been called'
 
475
            assert mock_send_command.called, '_underscore_send_command() should have been called'
 
476
 
 
477
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
 
478
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
 
479
    @patch.object(openlp.core.projectors.pjlink.PJLink, '_send_command')
 
480
    @patch.object(openlp.core.projectors.pjlink, 'log')
 
481
    def test_send_command_add_priority_command(self, mock_log, mock_send_command, mock_reset, mock_state):
 
482
        """
 
483
        Test _send_command adding priority queue item
 
484
        """
 
485
        # GIVEN: Test object
 
486
        test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX,
 
487
                                                           clss=self.pjlink.pjlink_class,
 
488
                                                           suff=PJLINK_SUFFIX)
 
489
        log_error_calls = []
 
490
        log_warning_calls = []
 
491
        log_debug_calls = [call('({ip}) send_command(): Building cmd="CLSS" opts="?"'.format(ip=self.pjlink.name)),
 
492
                           call('({ip}) send_command(): Adding to priority queue'.format(ip=self.pjlink.name))]
 
493
        mock_state.return_value = S_CONNECTED
 
494
 
 
495
        # Patch here since pjlink does not have priority or send queue's until instantiated
 
496
        with patch.object(self.pjlink, 'send_queue') as mock_send, \
 
497
                patch.object(self.pjlink, 'priority_queue') as mock_priority:
 
498
 
 
499
            # WHEN: send_command called with valid priority command
 
500
            self.pjlink.send_command(cmd='CLSS', priority=True)
 
501
 
 
502
            # THEN:
 
503
            mock_log.debug.assert_has_calls(log_debug_calls)
 
504
            mock_log.warning.assert_has_calls(log_warning_calls)
 
505
            mock_log.error.assert_has_calls(log_error_calls)
 
506
            mock_priority.append.assert_called_with(test_command)
 
507
            assert (not mock_send.append.called), 'send_queue should not have changed'
 
508
            assert (not mock_reset.called), 'reset_information() should not have been called'
 
509
            assert mock_send_command.called, '_underscore_send_command() should have been called'
 
510
 
 
511
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
 
512
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
 
513
    @patch.object(openlp.core.projectors.pjlink.PJLink, '_send_command')
 
514
    @patch.object(openlp.core.projectors.pjlink, 'log')
 
515
    def test_send_command_duplicate_normal_command(self, mock_log, mock_send_command, mock_reset, mock_state):
 
516
        """
 
517
        Test send_command with duplicate item for normal queue
 
518
        """
 
519
        # GIVEN: Test object
 
520
        test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX,
 
521
                                                           clss=self.pjlink.pjlink_class,
 
522
                                                           suff=PJLINK_SUFFIX)
 
523
        log_error_calls = []
 
524
        log_warning_calls = [call('({ip}) send_command(): Already in normal queue - '
 
525
                                  'skipping'.format(ip=self.pjlink.name))]
 
526
        log_debug_calls = [call('({ip}) send_command(): Building cmd="CLSS" opts="?"'.format(ip=self.pjlink.name))]
 
527
        mock_state.return_value = S_CONNECTED
 
528
        self.pjlink.send_queue = [test_command]
 
529
        self.pjlink.priority_queue = []
 
530
 
 
531
        # WHEN: send_command called with same command in normal queue
 
532
        self.pjlink.send_command(cmd='CLSS')
 
533
 
 
534
        # THEN:
 
535
        mock_log.debug.assert_has_calls(log_debug_calls)
 
536
        mock_log.warning.assert_has_calls(log_warning_calls)
 
537
        mock_log.error.assert_has_calls(log_error_calls)
 
538
        assert (self.pjlink.send_queue == [test_command]), 'Send queue should have one entry'
 
539
        assert (not self.pjlink.priority_queue), 'Priority queue should be empty'
 
540
        assert (not mock_reset.called), 'reset_information() should not have been called'
 
541
        assert mock_send_command.called, '_underscore_send_command() should have been called'
 
542
 
 
543
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
 
544
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
 
545
    @patch.object(openlp.core.projectors.pjlink.PJLink, '_send_command')
 
546
    @patch.object(openlp.core.projectors.pjlink, 'log')
 
547
    def test_send_command_duplicate_priority_command(self, mock_log, mock_send_command, mock_reset, mock_state):
 
548
        """
 
549
        Test send_command with duplicate item for priority queue
 
550
        """
 
551
        # GIVEN: Test object
 
552
        test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX,
 
553
                                                           clss=self.pjlink.pjlink_class,
 
554
                                                           suff=PJLINK_SUFFIX)
 
555
        log_error_calls = []
 
556
        log_warning_calls = [call('({ip}) send_command(): Already in priority queue - '
 
557
                                  'skipping'.format(ip=self.pjlink.name))]
 
558
        log_debug_calls = [call('({ip}) send_command(): Building cmd="CLSS" opts="?"'.format(ip=self.pjlink.name))]
 
559
        mock_state.return_value = S_CONNECTED
 
560
        self.pjlink.send_queue = []
 
561
        self.pjlink.priority_queue = [test_command]
 
562
 
 
563
        # WHEN: send_command called with same command in priority queue
 
564
        self.pjlink.send_command(cmd='CLSS', priority=True)
 
565
 
 
566
        # THEN:
 
567
        mock_log.debug.assert_has_calls(log_debug_calls)
 
568
        mock_log.warning.assert_has_calls(log_warning_calls)
 
569
        mock_log.error.assert_has_calls(log_error_calls)
 
570
        assert (not self.pjlink.send_queue), 'Send queue should be empty'
 
571
        assert (self.pjlink.priority_queue == [test_command]), 'Priority queue should have one entry'
 
572
        assert (not mock_reset.called), 'reset_information() should not have been called'
 
573
        assert mock_send_command.called, '_underscore_send_command() should have been called'
 
574
 
 
575
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
 
576
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
 
577
    @patch.object(openlp.core.projectors.pjlink.PJLink, '_send_command')
 
578
    @patch.object(openlp.core.projectors.pjlink, 'log')
 
579
    def test_send_command_invalid_command_empty_queues(self, mock_log, mock_send_command, mock_reset, mock_state):
 
580
        """
 
581
        Test send_command with invalid command
 
582
        """
 
583
        # GIVEN: Test object
 
584
        log_error_calls = [call('({ip}) send_command(): Invalid command requested - '
 
585
                                'ignoring.'.format(ip=self.pjlink.name))]
 
586
        log_warning_calls = []
 
587
        log_debug_calls = []
 
588
        mock_state.return_value = S_CONNECTED
 
589
        self.pjlink.send_queue = []
 
590
        self.pjlink.priority_queue = []
 
591
 
 
592
        # WHEN: send_command with invalid command
 
593
        self.pjlink.send_command(cmd='DONTCARE')
 
594
 
 
595
        # THEN:
 
596
        mock_log.debug.assert_has_calls(log_debug_calls)
 
597
        mock_log.warning.assert_has_calls(log_warning_calls)
 
598
        mock_log.error.assert_has_calls(log_error_calls)
 
599
        assert (not self.pjlink.send_queue), 'Send queue should be empty'
 
600
        assert (not self.pjlink.priority_queue), 'Priority queue should be empty'
 
601
        assert (not mock_reset.called), 'reset_information() should not have been called'
 
602
        assert (not mock_send_command.called), '_underscore_send_command() should not have been called'
 
603
 
 
604
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
 
605
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
 
606
    @patch.object(openlp.core.projectors.pjlink.PJLink, '_send_command')
 
607
    @patch.object(openlp.core.projectors.pjlink, 'log')
 
608
    def test_send_command_invalid_command_normal_queue(self, mock_log, mock_send_command, mock_reset, mock_state):
 
609
        """
 
610
        Test _send_command with invalid command for normal queue
 
611
        """
 
612
        # GIVEN: Test object
 
613
        test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX,
 
614
                                                           clss=self.pjlink.pjlink_class,
 
615
                                                           suff=PJLINK_SUFFIX)
 
616
        log_error_calls = [call('({ip}) send_command(): Invalid command requested - '
 
617
                                'ignoring.'.format(ip=self.pjlink.name))]
 
618
        log_warning_calls = []
 
619
        log_debug_calls = []
 
620
        mock_state.return_value = S_CONNECTED
 
621
        self.pjlink.send_queue = [test_command]
 
622
        self.pjlink.priority_queue = []
 
623
 
 
624
        # WHEN: send_command with invalid command
 
625
        self.pjlink.send_command(cmd='DONTCARE')
 
626
 
 
627
        # THEN:
 
628
        mock_log.debug.assert_has_calls(log_debug_calls)
 
629
        mock_log.warning.assert_has_calls(log_warning_calls)
 
630
        mock_log.error.assert_has_calls(log_error_calls)
 
631
        assert self.pjlink.send_queue, 'Send queue should have one entry'
 
632
        assert (not self.pjlink.priority_queue), 'Priority queue should be empty'
 
633
        assert (not mock_reset.called), 'reset_information() should not have been called'
 
634
        assert mock_send_command.called, '_underscore_send_command() should have been called'
 
635
 
 
636
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
 
637
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
 
638
    @patch.object(openlp.core.projectors.pjlink.PJLink, '_send_command')
 
639
    @patch.object(openlp.core.projectors.pjlink, 'log')
 
640
    def test_send_command_invalid_command_priority_queue(self, mock_log, mock_send_command, mock_reset, mock_state):
 
641
        """
 
642
        Test _send_command with invalid command for priority queue
 
643
        """
 
644
        # GIVEN: Test object
 
645
        test_command = '{prefix}{clss}CLSS ?{suff}'.format(prefix=PJLINK_PREFIX,
 
646
                                                           clss=self.pjlink.pjlink_class,
 
647
                                                           suff=PJLINK_SUFFIX)
 
648
        log_error_calls = [call('({ip}) send_command(): Invalid command requested - '
 
649
                                'ignoring.'.format(ip=self.pjlink.name))]
 
650
        log_warning_calls = []
 
651
        log_debug_calls = []
 
652
        mock_state.return_value = S_CONNECTED
 
653
        self.pjlink.send_queue = []
 
654
        self.pjlink.priority_queue = [test_command]
 
655
 
 
656
        # WHEN: send_command with invalid command
 
657
        self.pjlink.send_command(cmd='DONTCARE', priority=True)
 
658
 
 
659
        # THEN:
 
660
        mock_log.debug.assert_has_calls(log_debug_calls)
 
661
        mock_log.warning.assert_has_calls(log_warning_calls)
 
662
        mock_log.error.assert_has_calls(log_error_calls)
 
663
        assert (not self.pjlink.send_queue), 'Send queue should be empty'
 
664
        assert self.pjlink.priority_queue, 'Priority queue should have one entry'
 
665
        assert (not mock_reset.called), 'reset_information() should not have been called'
 
666
        assert mock_send_command.called, '_underscore_send_command() should have been called'
 
667
 
 
668
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'state')
 
669
    @patch.object(openlp.core.projectors.pjlink.PJLink, 'reset_information')
 
670
    @patch.object(openlp.core.projectors.pjlink.PJLink, '_send_command')
 
671
    @patch.object(openlp.core.projectors.pjlink, 'log')
 
672
    def test_send_command_not_connected(self, mock_log, mock_send_command, mock_reset, mock_state):
 
673
        """
 
674
        Test send_command when not connected
 
675
        """
 
676
        # GIVEN: Test object
 
677
        log_error_calls = []
 
678
        log_warning_calls = [call('({ip}) send_command(): Not connected - returning'.format(ip=self.pjlink.name))]
 
679
        log_debug_calls = []
 
680
        mock_state.return_value = S_NOT_CONNECTED
 
681
        self.pjlink.send_queue = []
 
682
        self.pjlink.priority_queue = []
 
683
 
 
684
        # WHEN: send_command called when not connected
 
685
        self.pjlink.send_command(cmd=None)
 
686
 
 
687
        # THEN:
 
688
        mock_log.debug.assert_has_calls(log_debug_calls)
 
689
        mock_log.warning.assert_has_calls(log_warning_calls)
 
690
        mock_log.error.assert_has_calls(log_error_calls)
 
691
        assert (not self.pjlink.send_queue), 'Send queue should be empty'
 
692
        assert (not self.pjlink.priority_queue), 'Priority queue should be empty'
 
693
        assert mock_reset.called, 'reset_information() should have been called'
 
694
        assert (not mock_send_command.called), '_underscore_send_command() should not have been called'