Merge branch 'commands-tests' into 'master'

commands-tests updates/refactoring 2022-03-02

See merge request openlp/openlp!437
This commit is contained in:
Raoul Snyman 2022-03-16 14:06:39 +00:00
commit 38073ea8be
12 changed files with 587 additions and 385 deletions

View File

@ -37,6 +37,7 @@ PJLINK_MAX_PACKET = 136
PJLINK_PREFIX = '%'
PJLINK_PORT = 4352
PJLINK_SUFFIX = CR
PJLINK_SVER_MAX_LEN = 32
PJLINK_TIMEOUT = 30.0
PJLINK_TOKEN_SIZE = 8 # PJLINK 1 <token> : where <token> is 8 characters

View File

@ -497,7 +497,7 @@ class ProjectorManager(QtWidgets.QWidget, RegistryBase, UiProjectorManager, LogM
if ans == msg.Cancel:
return
try:
projector.link.changeStatus.disconnect(self.update_status)
projector.link.projectorChangeStatus.disconnect(self.update_status)
except (AttributeError, TypeError):
pass
try:
@ -717,7 +717,7 @@ class ProjectorManager(QtWidgets.QWidget, RegistryBase, UiProjectorManager, LogM
widget.setData(QtCore.Qt.UserRole, item)
item.pjlink.db_item = item.db_item
item.widget = widget
item.pjlink.changeStatus.connect(self.update_status)
item.pjlink.projectorChangeStatus.connect(self.update_status)
item.pjlink.projectorAuthentication.connect(self.authentication_error)
item.pjlink.projectorNoAuthentication.connect(self.no_authentication_error)
item.pjlink.projectorUpdateIcons.connect(self.update_icons)

View File

@ -186,12 +186,14 @@ class PJLink(QtNetwork.QTcpSocket):
Socket services for PJLink TCP packets.
"""
# Signals sent by this module
changeStatus = QtCore.pyqtSignal(str, int, str)
projectorChangeStatus = QtCore.pyqtSignal(str, int, str)
projectorStatus = QtCore.pyqtSignal(int) # Status update
projectorAuthentication = QtCore.pyqtSignal(str) # Authentication error
projectorNoAuthentication = QtCore.pyqtSignal(str) # PIN set and no authentication needed
projectorReceivedData = QtCore.pyqtSignal() # Notify when received data finished processing
projectorUpdateIcons = QtCore.pyqtSignal() # Update the status icons on toolbar
# Deprecated
changeStatus = projectorChangeStatus # Use projectorChangeStatus
def __init__(self, projector, *args, **kwargs):
"""
@ -207,7 +209,8 @@ class PJLink(QtNetwork.QTcpSocket):
log.debug(f'PJLink(projector="{projector}", args="{args}" kwargs="{kwargs}")')
super().__init__()
self.settings_section = 'projector'
self.entry = projector
self.db = projector
self.entry = self.db # Deprecated use self.db
self.ip = self.entry.ip
self.qhost = QtNetwork.QHostAddress(self.ip)
self.location = self.entry.location

View File

@ -49,14 +49,19 @@ import string
from openlp.core.common.registry import Registry
from openlp.core.projectors.constants import E_AUTHENTICATION, PJLINK_DEFAULT_CODES, PJLINK_ERRORS, \
PJLINK_ERST_DATA, PJLINK_ERST_LIST, PJLINK_ERST_STATUS, PJLINK_POWR_STATUS, PJLINK_TOKEN_SIZE, \
E_NO_AUTHENTICATION, S_AUTHENTICATE, S_CONNECT, S_DATA_OK, S_OFF, S_OK, S_ON, S_STANDBY, STATUS_MSG
PJLINK_ERST_DATA, PJLINK_ERST_LIST, PJLINK_ERST_STATUS, PJLINK_POWR_STATUS, PJLINK_SVER_MAX_LEN, \
PJLINK_TOKEN_SIZE, E_NO_AUTHENTICATION, S_AUTHENTICATE, S_CONNECT, S_DATA_OK, S_OFF, S_OK, S_ON, \
S_STANDBY, STATUS_MSG
log = logging.getLogger(__name__)
log.debug('Loading pjlinkcommands')
__all__ = ['process_command']
_pjlink_functions = {}
# Helper until I update the rest of the tests
pjlink_functions = _pjlink_functions
# This should be the only function that's imported.
def process_command(projector, cmd, data):
@ -103,16 +108,34 @@ def process_ackn(projector, data):
pass
def process_avmt(projector, data):
_pjlink_functions['ACKN'] = process_ackn
def _process_avmt_mute(projector, data):
"""
Helper to set projector.mute
"""
projector.mute = data
def _process_avmt_shutter(projector, data):
"""
Helper to set projector.shutter
"""
projector.shutter = data
def _process_avmt(projector, data):
"""
Process shutter and speaker status. See PJLink specification for format.
Update projector.mute (audio) and projector.shutter (video shutter).
Update projector.mute (audio mute) and projector.shutter (video mute).
10 = Shutter open, audio unchanged
11 = Shutter closed, audio unchanged
20 = Shutter unchanged, Audio normal
21 = Shutter unchanged, Audio muted
30 = Shutter open, audio muted
31 = Shutter closed, audio normal
20 = Shutter unchanged, audio normal
21 = Shutter unchanged, audio mute
30 = Shutter open, audio normal
31 = Shutter closed, audio mute
:param projector: Projector instance
:param data: Shutter and audio status
@ -125,25 +148,26 @@ def process_avmt(projector, data):
'31': {'shutter': True, 'mute': True}
}
if data not in settings:
log.warning('({ip}) Invalid av mute response: {data}'.format(ip=projector.entry.name, data=data))
log.warning(f'({projector.entry.name}) Invalid av mute response: {data}')
return
shutter = settings[data]['shutter']
mute = settings[data]['mute']
# Check if we need to update the icons
update_icons = (shutter != projector.shutter) or (mute != projector.mute)
update_icons = False
if projector.shutter != shutter:
_process_avmt_shutter(projector=projector, data=shutter)
update_icons = True
log.debug(f'({projector.entry.name}) Setting shutter to {"closed" if shutter else "open"}')
if projector.mute != mute:
_process_avmt_mute(projector=projector, data=mute)
projector.mute = mute
update_icons = True
log.debug(f'({projector.entry.name}) Setting speaker to {"muted" if mute else "normal"}')
if update_icons:
if projector.shutter != shutter:
projector.shutter = shutter
log.debug('({ip}) Setting shutter to {chk}'.format(ip=projector.entry.name,
chk='closed' if shutter else 'open'))
if projector.mute != mute:
projector.mute = mute
log.debug('({ip}) Setting speaker to {chk}'.format(ip=projector.entry.name,
chk='muted' if shutter else 'normal'))
if 'AVMT' in projector.status_timer_checks:
projector.status_timer_delete('AVMT')
projector.projectorUpdateIcons.emit()
return
projector.status_timer_delete('AVMT')
_pjlink_functions['AVMT'] = _process_avmt
def process_clss(projector, data):
@ -178,14 +202,14 @@ def process_clss(projector, data):
clss = data
projector.pjlink_class = clss
log.debug(f'({projector.entry.name}) Setting pjlink_class for this projector to "{projector.pjlink_class}"')
if projector.no_poll:
return
if not projector.no_poll:
# Since we call this one on first connect, setup polling from here
log.debug(f'({projector.entry.name}) process_pjlink(): Starting timer')
projector.poll_timer.setInterval(1000) # Set 1 second for initial information
projector.poll_timer.start()
# Since we call this one on first connect, setup polling from here
log.debug(f'({projector.entry.name}) process_pjlink(): Starting timer')
projector.poll_timer.setInterval(1000) # Set 1 second for initial information
projector.poll_timer.start()
return
_pjlink_functions['CLSS'] = process_clss
def process_erst(projector, data):
@ -198,13 +222,11 @@ def process_erst(projector, data):
"""
if len(data) != PJLINK_ERST_DATA['DATA_LENGTH']:
count = PJLINK_ERST_DATA['DATA_LENGTH']
log.warning('({ip}) Invalid error status response "{data}": length != {count}'.format(ip=projector.entry.name,
data=data,
count=count))
log.warning(f'({projector.entry.name}) Invalid error status response "{data}": length != {count}')
return
if not data.isnumeric():
# Bad data - ignore
log.warning('({ip}) Invalid error status response "{data}"'.format(ip=projector.entry.name, data=data))
log.warning(f'({projector.entry.name}) Invalid error status response "{data}"')
return
if int(data) == 0:
projector.projector_errors = None
@ -233,6 +255,9 @@ def process_erst(projector, data):
return
_pjlink_functions['ERST'] = process_erst
def process_inf1(projector, data):
"""
Manufacturer name set in projector.
@ -242,11 +267,13 @@ def process_inf1(projector, data):
:param data: Projector manufacturer
"""
projector.manufacturer = data
log.debug('({ip}) Setting projector manufacturer data to "{data}"'.format(ip=projector.entry.name,
data=projector.manufacturer))
log.debug(f'({projector.entry.name}) Setting projector manufacturer data to "{projector.manufacturer}"')
return
_pjlink_functions['INF1'] = process_inf1
def process_inf2(projector, data):
"""
Projector Model set in projector.
@ -256,10 +283,13 @@ def process_inf2(projector, data):
:param data: Model name
"""
projector.model = data
log.debug('({ip}) Setting projector model to "{data}"'.format(ip=projector.entry.name, data=projector.model))
log.debug(f'({projector.entry.name}) Setting projector model to "{projector.model}"')
return
_pjlink_functions['INF2'] = process_inf2
def process_info(projector, data):
"""
Any extra info set in projector.
@ -269,11 +299,13 @@ def process_info(projector, data):
:param data: Projector other info
"""
projector.other_info = data
log.debug('({ip}) Setting projector other_info to "{data}"'.format(ip=projector.entry.name,
data=projector.other_info))
log.debug(f'({projector.entry.name}) Setting projector other_info to "{projector.other_info}"')
return
_pjlink_functions['INFO'] = process_info
def process_inpt(projector, data):
"""
Current source input selected. See PJLink specification for format.
@ -286,19 +318,20 @@ def process_inpt(projector, data):
if projector.source_available is not None:
# We have available inputs, so verify it's in the list
if data not in projector.source_available:
log.warning('({ip}) Input source not listed in available sources - '
'ignoring'.format(ip=projector.entry.name))
log.warning(f'({projector.entry.name}) Input source not listed in available sources - ignoring')
return
elif data not in PJLINK_DEFAULT_CODES:
# Hmm - no sources available yet, so check with PJLink defaults
log.warning('({ip}) Input source not listed as a PJLink valid source '
'- ignoring'.format(ip=projector.entry.name))
log.warning(f'({projector.entry.name}) Input source not listed as a PJLink valid source - ignoring')
return
projector.source = data
log.debug('({ip}) Setting current source to "{data}"'.format(ip=projector.entry.name, data=projector.source))
log.debug(f'({projector.entry.name}) Setting current source to "{projector.source}"')
return
_pjlink_functions['INPT'] = process_inpt
def process_inst(projector, data):
"""
Available source inputs. See PJLink specification for format.
@ -313,12 +346,14 @@ def process_inst(projector, data):
sources.append(source)
sources.sort()
projector.source_available = sources
log.debug('({ip}) Setting projector source_available to "{data}"'.format(ip=projector.entry.name,
data=projector.source_available))
log.debug(f'({projector.entry.name}) Setting projector source_available to "{projector.source_available}"')
projector.projectorUpdateIcons.emit()
return
_pjlink_functions['INST'] = process_inst
def process_lamp(projector, data):
"""
Lamp(s) status. See PJLink Specifications for format.
@ -332,14 +367,13 @@ def process_lamp(projector, data):
lamp_list = data.split()
if len(lamp_list) < 2:
# Invalid data - not enough information
log.warning('({ip}) process_lamp(): Invalid data "{data}" - '
'Missing data'.format(ip=projector.entry.name, data=data))
log.warning(f'({projector.entry.name}) process_lamp(): Invalid data "{data}" - Missing data')
return
else:
while lamp_list:
if not lamp_list[0].isnumeric() or not lamp_list[1].isnumeric():
# Invalid data - we'll ignore the rest for now
log.warning('({ip}) process_lamp(): Invalid data "{data}"'.format(ip=projector.entry.name, data=data))
log.warning(f'({projector.entry.name}) process_lamp(): Invalid data "{data}"')
return
fill = {'Hours': int(lamp_list[0]), 'On': False if lamp_list[1] == '0' else True}
lamps.append(fill)
@ -349,18 +383,24 @@ def process_lamp(projector, data):
return
def process_lkup(projector, data):
_pjlink_functions['LAMP'] = process_lamp
def _process_lkup(projector, data):
"""
Process UDP request indicating remote is available for connection
:param projector: Projector instance
:param data: Data packet from remote
"""
log.debug('({ip}) Processing LKUP command'.format(ip=projector.entry.name))
log.debug(f'({projector.entry.name}) Processing LKUP command')
if Registry().get('settings').value('projector/connect when LKUP received'):
projector.connect_to_host()
_pjlink_functions['LKUP'] = _process_lkup
def process_name(projector, data):
"""
Projector name set in projector.
@ -370,11 +410,13 @@ def process_name(projector, data):
:param data: Projector name
"""
projector.pjlink_name = data
log.debug('({ip}) Setting projector PJLink name to "{data}"'.format(ip=projector.entry.name,
data=projector.pjlink_name))
log.debug(f'({projector.entry.name}) Setting projector PJLink name to "{projector.pjlink_name}"')
return
_pjlink_functions['NAME'] = process_name
def process_pjlink(projector, data):
"""
Process initial socket connection to terminal.
@ -419,6 +461,9 @@ def process_pjlink(projector, data):
return S_AUTHENTICATE
_pjlink_functions['PJLINK'] = process_pjlink
def process_powr(projector, data):
"""
Power status. See PJLink specification for format.
@ -427,15 +472,14 @@ def process_powr(projector, data):
:param projector: Projector instance
:param data: Power status
"""
log.debug('({ip}) Processing POWR command'.format(ip=projector.entry.name))
log.debug(f'({projector.entry.name}) Processing POWR command')
if data not in PJLINK_POWR_STATUS:
# Log unknown status response
log.warning('({ip}) Unknown power response: "{data}"'.format(ip=projector.entry.name, data=data))
log.warning(f'({projector.entry.name}) Unknown power response: "{data}"')
return
power = PJLINK_POWR_STATUS[data]
update_icons = projector.power != power
if update_icons:
if projector.power != power:
projector.power = power
projector.change_status(PJLINK_POWR_STATUS[data])
projector.projectorUpdateIcons.emit()
@ -443,11 +487,14 @@ def process_powr(projector, data):
# Input sources list should only be available after power on, so update here
projector.send_command('INST')
if projector.power in [S_ON, S_STANDBY, S_OFF] and 'POWR' in projector.status_timer_checks:
if projector.power in [S_ON, S_STANDBY, S_OFF]:
projector.status_timer_delete(cmd='POWR')
return
_pjlink_functions['POWR'] = process_powr
def process_rfil(projector, data):
"""
Process replacement filter type
@ -458,9 +505,12 @@ def process_rfil(projector, data):
if projector.model_filter is None:
projector.model_filter = data
else:
log.warning('({ip}) Filter model already set'.format(ip=projector.entry.name))
log.warning('({ip}) Saved model: "{old}"'.format(ip=projector.entry.name, old=projector.model_filter))
log.warning('({ip}) New model: "{new}"'.format(ip=projector.entry.name, new=data))
log.warning(f'({projector.entry.name}) Filter model already set')
log.warning(f'({projector.entry.name}) Saved model: "{projector.model_filter}"')
log.warning(f'({projector.entry.name}) New model: "{data}"')
_pjlink_functions['RFIL'] = process_rfil
def process_rlmp(projector, data):
@ -473,9 +523,12 @@ def process_rlmp(projector, data):
if projector.model_lamp is None:
projector.model_lamp = data
else:
log.warning('({ip}) Lamp model already set'.format(ip=projector.entry.name))
log.warning('({ip}) Saved lamp: "{old}"'.format(ip=projector.entry.name, old=projector.model_lamp))
log.warning('({ip}) New lamp: "{new}"'.format(ip=projector.entry.name, new=data))
log.warning(f'({projector.entry.name}) Lamp model already set')
log.warning(f'({projector.entry.name}) Saved lamp: "{projector.model_lamp}"')
log.warning(f'({projector.entry.name}) New lamp: "{data}"')
_pjlink_functions['RLMP'] = process_rlmp
def process_snum(projector, data):
@ -486,22 +539,24 @@ def process_snum(projector, data):
:param data: Serial number from projector.
"""
if projector.serial_no is None:
log.debug('({ip}) Setting projector serial number to "{data}"'.format(ip=projector.entry.name, data=data))
log.debug(f'({projector.entry.name}) Setting projector serial number to "{data}"')
projector.serial_no = data
projector.db_update = False
return
# Compare serial numbers and see if we got the same projector
if projector.serial_no != data:
log.warning('({ip}) Projector serial number does not match saved serial '
'number'.format(ip=projector.entry.name))
log.warning('({ip}) Saved: "{old}"'.format(ip=projector.entry.name, old=projector.serial_no))
log.warning('({ip}) Received: "{new}"'.format(ip=projector.entry.name, new=data))
log.warning('({ip}) NOT saving serial number'.format(ip=projector.entry.name))
log.warning(f'({projector.entry.name}) Projector serial number does not match saved serial number')
log.warning(f'({projector.entry.name}) Saved: "{projector.serial_no}"')
log.warning(f'({projector.entry.name}) Received: "{data}"')
log.warning(f'({projector.entry.name}) NOT saving serial number')
projector.serial_no_received = data
def process_srch(projector=None, data=None):
_pjlink_functions['SNUM'] = process_snum
def _process_srch(projector=None, data=None):
"""
Process the SRCH command.
@ -512,58 +567,38 @@ def process_srch(projector=None, data=None):
:param projector: Projector instance (actually ignored for this command)
:param data: Data in packet
"""
if projector is None:
log.warning('SRCH packet detected - ignoring')
else:
log.warning(f'({projector.entry.name}) SRCH packet detected - ignoring')
return
msg = 'SRCH packet detected - ignoring'
name = ''
if projector is not None:
name = f'({projector.entry.name}) '
log.warning(f'{name}{msg}')
def process_sver(projector, data):
_pjlink_functions['SRCH'] = _process_srch
def _process_sver(projector, data):
"""
Software version of projector
:param projector: Projector instance
:param data: Software version of projector
"""
if len(data) > 32:
# Defined in specs max version is 32 characters
log.warning('Invalid software version - too long')
if len(data) > PJLINK_SVER_MAX_LEN:
# Defined in specs 0-32 characters max
log.warning(f'({projector.name}) Invalid software version - too long')
return
if projector.sw_version is not None:
if projector.sw_version == data:
log.debug('({ip}) Software version same as saved version - returning'.format(ip=projector.entry.name))
return
log.warning('({ip}) Projector software version does not match saved '
'software version'.format(ip=projector.entry.name))
log.warning('({ip}) Saved: "{old}"'.format(ip=projector.entry.name, old=projector.sw_version))
log.warning('({ip}) Received: "{new}"'.format(ip=projector.entry.name, new=data))
log.warning('({ip}) Updating software version'.format(ip=projector.entry.name))
elif projector.sw_version == data:
log.debug(f'({projector.name}) Software version unchanged - returning')
return
elif projector.sw_version is not None:
log.debug(f'({projector.name}) Old software version "{projector.sw_version}"')
log.debug(f'({projector.name}) New software version "{data}"')
log.debug('({ip}) Setting projector software version to "{data}"'.format(ip=projector.entry.name, data=data))
# Software version changed - save
log.debug(f'({projector.entry.name}) Setting projector software version to "{data}"')
projector.sw_version = data
projector.db_update = True
# Map command to function.
pjlink_functions = {
'ACKN': process_ackn, # Class 2 (command is SRCH)
'AVMT': process_avmt,
'CLSS': process_clss,
'ERST': process_erst,
'INFO': process_info,
'INF1': process_inf1,
'INF2': process_inf2,
'INPT': process_inpt,
'INST': process_inst,
'LAMP': process_lamp,
'LKUP': process_lkup, # Class 2 (terminal request only - no cmd)
'NAME': process_name,
'PJLINK': process_pjlink,
'POWR': process_powr,
'SNUM': process_snum,
'SRCH': process_srch, # Class 2 (reply is ACKN)
'SVER': process_sver,
'RFIL': process_rfil,
'RLMP': process_rlmp
}
_pjlink_functions['SVER'] = _process_sver

View File

@ -25,7 +25,7 @@ Help classes/functions for PJLink Projector tests
from unittest.mock import MagicMock
from PyQt5 import QtNetwork
from openlp.core.projectors.constants import S_OK, S_NOT_CONNECTED
from openlp.core.projectors.constants import S_NOT_CONNECTED, S_OFF, S_OK
from openlp.core.projectors.db import Projector
from tests.resources.projector.data import TEST1_DATA
@ -57,8 +57,10 @@ class FakePJLink(object):
"""
Helper class with signals and methods mocked
"""
def __init__(self, projector=None, *args, **kwargs):
def __init__(self, projector=Projector(**TEST1_DATA), *args, **kwargs):
# Signal mocks
self.changeStatus = MagicMock() # Deprecated use projectorChangeStatus
self.projectorChangeStatus = MagicMock()
self.projectorStatus = MagicMock()
self.projectorAuthentication = MagicMock()
self.projectorNoAuthentication = MagicMock()
@ -66,7 +68,7 @@ class FakePJLink(object):
self.projectorUpdateIcons = MagicMock()
# Method mocks
self.changeStatus = MagicMock()
self.change_status = MagicMock()
self.connect_to_host = MagicMock()
self.disconnect_from_host = MagicMock()
self.poll_timer = MagicMock()
@ -75,8 +77,10 @@ class FakePJLink(object):
self.set_shutter_closed = MagicMock()
self.set_shutter_open = MagicMock()
self.socket_timer = MagicMock()
self.status_timer = MagicMock()
self.state = MagicMock()
self.status_timer = MagicMock()
self.status_timer_add = MagicMock()
self.status_timer_delete = MagicMock()
# Some tests that may include what it thinks are ProjectorItem()
# If ProjectorItem() is called, will probably overwrite these - OK
@ -84,19 +88,20 @@ class FakePJLink(object):
self.pjlink = self
# Normal entries from PJLink
self.entry = Projector(**TEST1_DATA) if projector is None else projector
self.ip = self.entry.ip
self.db = projector
self.entry = self.db # Deprecated use self.db
self.ip = self.db.ip
self.qhost = QtNetwork.QHostAddress(self.ip)
self.location = self.entry.location
self.mac_adx = self.entry.mac_adx
self.name = self.entry.name
self.notes = self.entry.notes
self.pin = self.entry.pin
self.port = int(self.entry.port)
self.pjlink_class = "1" if self.entry.pjlink_class is None else self.entry.pjlink_class
self.poll_time = 20000 if 'poll_time' not in kwargs else kwargs['poll_time'] * 1000
self.socket_timeout = 5000 if 'socket_timeout' not in kwargs else kwargs['socket_timeout'] * 1000
self.no_poll = 'no_poll' in kwargs
self.location = self.db.location
self.mac_adx = self.db.mac_adx
self.name = self.db.name
self.notes = self.db.notes
self.pin = self.db.pin
self.port = int(self.db.port)
self.pjlink_class = "1" if self.db.pjlink_class is None else self.db.pjlink_class
self.poll_time = 20000
self.socket_timeout = 5000
self.no_poll = True
self.status_connect = S_NOT_CONNECTED
self.last_command = ''
self.projector_status = S_NOT_CONNECTED
@ -105,4 +110,25 @@ class FakePJLink(object):
self.priority_queue = []
self.send_busy = False
self.status_timer_checks = {} # Keep track of events for the status timer
# Default mock return values
# reset_information attributes
self.fan = None # ERST
self.filter_time = None # FILT
self.lamp = None # LAMP
self.mac_adx_received = None # ACKN
self.manufacturer = None # INF1
self.model = None # INF2
self.model_filter = None # RFIL
self.model_lamp = None # RLMP
self.mute = None # AVMT
self.other_info = None # INFO
self.pjlink_name = None # NAME
self.power = S_OFF # POWR
self.projector_errors = {} # Full ERST errors
self.serial_no = None # SNUM
self.serial_no_received = None
self.sw_version = None # SVER
self.sw_version_received = None
self.shutter = None # AVMT
self.source_available = None # INST
self.source = None # INPT

View File

@ -0,0 +1,222 @@
# -*- coding: utf-8 -*-
##########################################################################
# OpenLP - Open Source Lyrics Projection #
# ---------------------------------------------------------------------- #
# Copyright (c) 2008-2022 OpenLP Developers #
# ---------------------------------------------------------------------- #
# This program is free software: you can redistribute it and/or modify #
# it under the terms of the GNU General Public License as published by #
# the Free Software Foundation, either version 3 of the License, or #
# (at your option) any later version. #
# #
# This program is distributed in the hope that it will be useful, #
# but WITHOUT ANY WARRANTY; without even the implied warranty of #
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the #
# GNU General Public License for more details. #
# #
# You should have received a copy of the GNU General Public License #
# along with this program. If not, see <https://www.gnu.org/licenses/>. #
##########################################################################
"""
Test _process_avmt
"""
import logging
import openlp.core.projectors.pjlinkcommands
from unittest.mock import patch
test_module = openlp.core.projectors.pjlinkcommands.__name__
_process_avmt = openlp.core.projectors.pjlinkcommands._process_avmt
def test_avmt_mute(fake_pjlink):
"""
Test _proces_avmt_mute helper
"""
# GIVEN: Test setup
fake_pjlink.mute = False
# WHEN: Called
openlp.core.projectors.pjlinkcommands._process_avmt_mute(projector=fake_pjlink, data=True)
# THEN: mute should have been changed
assert fake_pjlink.mute is True, 'Mute did not change'
def test_avmt_shutter(fake_pjlink):
"""
Test _proces_avmt_shutter helper
"""
# GIVEN: Test setup
fake_pjlink.shutter = False
# WHEN: Called
openlp.core.projectors.pjlinkcommands._process_avmt_shutter(projector=fake_pjlink, data=True)
# THEN: mute should have been changed
assert fake_pjlink.shutter is True, 'Mute did not change'
@patch.object(openlp.core.projectors.pjlinkcommands, '_process_avmt_mute')
@patch.object(openlp.core.projectors.pjlinkcommands, '_process_avmt_shutter')
def test_avmt_bad_data(mock_shutter, mock_mute, fake_pjlink, caplog):
"""
Test avmt bad data fail
"""
# GIVEN: Test setup
t_data = '36'
caplog.set_level(logging.DEBUG)
logs = [(test_module, logging.WARNING, f'({fake_pjlink.name}) Invalid av mute response: {t_data}')]
# WHEN: Called with an invalid setting
caplog.clear()
_process_avmt(projector=fake_pjlink, data=t_data)
# THEN: No other calls made
assert caplog.record_tuples == logs, 'Invalid log entries'
mock_mute.assert_not_called()
mock_shutter.assert_not_called()
fake_pjlink.projectorUpdateIcons.assert_not_called()
fake_pjlink.status_timer_delete.assert_not_called()
@patch.object(openlp.core.projectors.pjlinkcommands, '_process_avmt_mute')
@patch.object(openlp.core.projectors.pjlinkcommands, '_process_avmt_shutter')
def test_avmt_10(mock_shutter, mock_mute, fake_pjlink, caplog):
"""
Test 10 = Shutter open, audio unchanged
"""
# GIVEN: Test setup
t_data = '10'
caplog.set_level(logging.DEBUG)
logs = [(test_module, logging.DEBUG, f'({fake_pjlink.name}) Setting shutter to open')]
# WHEN: Called
_process_avmt(projector=fake_pjlink, data=t_data)
# THEN: Shutter and mute should be set correctly
assert caplog.record_tuples == logs, 'Invalid log entries'
mock_mute.assert_not_called()
mock_shutter.assert_called_with(projector=fake_pjlink, data=False)
fake_pjlink.projectorUpdateIcons.emit.assert_called_once()
fake_pjlink.status_timer_delete.assert_called_once_with('AVMT')
@patch.object(openlp.core.projectors.pjlinkcommands, '_process_avmt_mute')
@patch.object(openlp.core.projectors.pjlinkcommands, '_process_avmt_shutter')
def test_avmt_11(mock_shutter, mock_mute, fake_pjlink, caplog):
"""
Test 11 = Shutter closed, audio unchanged
"""
# GIVEN: Test setup
t_data = '11'
caplog.set_level(logging.DEBUG)
logs = [(test_module, logging.DEBUG, f'({fake_pjlink.name}) Setting shutter to closed')]
# WHEN: Called with setting shutter closed and mute on
_process_avmt(projector=fake_pjlink, data=t_data)
# THEN: Shutter and mute should be set correctly
assert caplog.record_tuples == logs, 'Invalid log entries'
mock_mute.assert_not_called()
mock_shutter.assert_called_once_with(projector=fake_pjlink, data=True)
fake_pjlink.projectorUpdateIcons.emit.assert_called_once()
fake_pjlink.status_timer_delete.assert_called_once_with('AVMT')
@patch.object(openlp.core.projectors.pjlinkcommands, '_process_avmt_mute')
@patch.object(openlp.core.projectors.pjlinkcommands, '_process_avmt_shutter')
def test_avmt_20(mock_shutter, mock_mute, fake_pjlink, caplog):
"""
Test 20 = Shutter unchanged, audio normal
"""
# GIVEN: Test setup
t_data = '20'
caplog.set_level(logging.DEBUG)
logs = [(test_module, logging.DEBUG, f'({fake_pjlink.name}) Setting speaker to normal')]
# WHEN: Called with setting shutter closed and mute on
_process_avmt(projector=fake_pjlink, data=t_data)
# THEN: Shutter and mute should be set correctly
assert caplog.record_tuples == logs, 'Invalid log entries'
mock_mute.assert_called_once_with(projector=fake_pjlink, data=False)
mock_shutter.assert_not_called()
fake_pjlink.projectorUpdateIcons.emit.assert_called_once()
fake_pjlink.status_timer_delete.assert_called_once_with('AVMT')
@patch.object(openlp.core.projectors.pjlinkcommands, '_process_avmt_mute')
@patch.object(openlp.core.projectors.pjlinkcommands, '_process_avmt_shutter')
def test_avmt_21(mock_shutter, mock_mute, fake_pjlink, caplog):
"""
Test 21 = Shutter unchanged, audio mute
"""
# GIVEN: Test setup
t_data = '21'
caplog.set_level(logging.DEBUG)
logs = [(test_module, logging.DEBUG, f'({fake_pjlink.name}) Setting speaker to muted')]
# WHEN: Called with setting shutter closed and mute on
_process_avmt(projector=fake_pjlink, data=t_data)
# THEN: Shutter and mute should be set correctly
assert caplog.record_tuples == logs, 'Invalid log entries'
mock_mute.assert_called_once_with(projector=fake_pjlink, data=True)
mock_shutter.assert_not_called()
fake_pjlink.projectorUpdateIcons.emit.assert_called_once()
fake_pjlink.status_timer_delete.assert_called_once_with('AVMT')
@patch.object(openlp.core.projectors.pjlinkcommands, '_process_avmt_mute')
@patch.object(openlp.core.projectors.pjlinkcommands, '_process_avmt_shutter')
def test_avmt_30(mock_shutter, mock_mute, fake_pjlink, caplog):
"""
Test 30 = Shutter open, audio normal
"""
# GIVEN: Test setup
t_data = '30'
caplog.set_level(logging.DEBUG)
logs = [(test_module, logging.DEBUG, f'({fake_pjlink.name}) Setting shutter to open'),
(test_module, logging.DEBUG, f'({fake_pjlink.name}) Setting speaker to normal')
]
# WHEN: Called
_process_avmt(projector=fake_pjlink, data=t_data)
# THEN: Shutter and mute should be set correctly
assert caplog.record_tuples == logs, 'Invalid log entries'
mock_mute.assert_called_once_with(projector=fake_pjlink, data=False)
mock_shutter.assert_called_once_with(projector=fake_pjlink, data=False)
fake_pjlink.projectorUpdateIcons.emit.assert_called_once()
fake_pjlink.status_timer_delete.assert_called_once_with('AVMT')
@patch.object(openlp.core.projectors.pjlinkcommands, '_process_avmt_mute')
@patch.object(openlp.core.projectors.pjlinkcommands, '_process_avmt_shutter')
def test_avmt_31(mock_shutter, mock_mute, fake_pjlink, caplog):
"""
Test 31 = Shutter closed, audio mute
"""
# GIVEN: Test object
t_data = '31'
caplog.set_level(logging.DEBUG)
logs = [(test_module, logging.DEBUG, f'({fake_pjlink.name}) Setting shutter to closed'),
(test_module, logging.DEBUG, f'({fake_pjlink.name}) Setting speaker to muted')
]
# WHEN: Called
_process_avmt(projector=fake_pjlink, data=t_data)
# THEN: Shutter and mute should be set correctly
assert caplog.record_tuples == logs, 'Invalid log entries'
mock_mute.assert_called_once_with(projector=fake_pjlink, data=True)
mock_shutter.assert_called_once_with(projector=fake_pjlink, data=True)
fake_pjlink.projectorUpdateIcons.emit.assert_called_once()
fake_pjlink.status_timer_delete.assert_called_once_with('AVMT')

View File

@ -26,9 +26,43 @@ Tests for commands that do not need much testing
import logging
import openlp.core.projectors.pjlinkcommands
from openlp.core.projectors.pjlinkcommands import process_srch
test_module = openlp.core.projectors.pjlinkcommands.__name__
_process_lkup = openlp.core.projectors.pjlinkcommands._process_lkup
_process_srch = openlp.core.projectors.pjlinkcommands._process_srch
def test_lkup_connect(fake_pjlink, settings, caplog):
"""
Test LKUP when settings indicate connect
"""
# GIVEN: Test setup
caplog.set_level(logging.DEBUG)
logs = [(test_module, logging.DEBUG, f'({fake_pjlink.name}) Processing LKUP command')]
settings.setValue('projector/connect when LKUP received', True)
# WHEN: Called
_process_lkup(projector=fake_pjlink, data=None)
# THEN: Only log entry made
assert caplog.record_tuples == logs, 'Invalid log entries'
fake_pjlink.connect_to_host.assert_called_once()
def test_lkup_no_connect(fake_pjlink, settings, caplog):
"""
Test LKUP when settings indicate no connect
"""
# GIVEN: Test setup
caplog.set_level(logging.DEBUG)
logs = [(test_module, logging.DEBUG, f'({fake_pjlink.name}) Processing LKUP command')]
settings.setValue('projector/connect when LKUP received', False)
# WHEN: Called
_process_lkup(projector=fake_pjlink, data=None)
# THEN: Only log entry made
assert caplog.record_tuples == logs, 'Invalid log entries'
fake_pjlink.connect_to_host.assert_not_called()
def test_srch_no_projector(caplog):
@ -40,7 +74,7 @@ def test_srch_no_projector(caplog):
logs = [(f'{test_module}', logging.WARNING, 'SRCH packet detected - ignoring')]
# WHEN: Called
t_chk = process_srch()
t_chk = _process_srch()
# THEN: Appropriate return code and log entries
assert t_chk is None, 'Invalid return code'
@ -57,7 +91,7 @@ def test_srch_with_projector(pjlink, caplog):
f'({pjlink.entry.name}) SRCH packet detected - ignoring')]
# WHEN: Called
t_chk = process_srch(projector=pjlink)
t_chk = _process_srch(projector=pjlink)
# THEN: Appropriate return code and log entries
assert t_chk is None, 'Invalid return code'

View File

@ -24,28 +24,16 @@ Test process_pjlink method
import logging
import openlp.core.projectors.pjlinkcommands
import pytest
from openlp.core.projectors.pjlinkcommands import process_pjlink
from openlp.core.projectors.constants import E_AUTHENTICATION, E_NO_AUTHENTICATION, \
S_AUTHENTICATE, S_CONNECT
from tests.helpers.projector import FakeProjector
from tests.resources.projector.data import TEST_PIN, TEST_SALT
test_module = openlp.core.projectors.pjlinkcommands.__name__
@pytest.fixture
def fake_pjlink():
"""
Helper since we don't need a full-blown PJLink() instance
"""
dumb_projector = FakeProjector()
yield dumb_projector
del(dumb_projector)
def test_normal_no_authentication_type(fake_pjlink, caplog):
"""
Test login prompt with not enough parameters
@ -147,6 +135,7 @@ def test_normal_login(fake_pjlink, caplog):
(f'{test_module}', logging.DEBUG,
f'({fake_pjlink.entry.name}) PJLINK: Returning S_CONNECT')
]
fake_pjlink.pin = None
# WHEN: Calling function
caplog.clear()
@ -230,7 +219,6 @@ def test_authenticate_invalid_salt(fake_pjlink, caplog):
# GIVEN: Test setup
caplog.set_level(logging.DEBUG)
t_data = '1 1a2b3c4g'
print(t_data)
logs = [(f'{test_module}', logging.DEBUG,
f'({fake_pjlink.entry.name}) Processing PJLINK command'),
(f'{test_module}', logging.ERROR,
@ -253,12 +241,12 @@ def test_authenticate_no_pin(fake_pjlink, caplog):
# GIVEN: Test setup
caplog.set_level(logging.DEBUG)
t_data = f'1 {TEST_SALT}'
print(t_data)
logs = [(f'{test_module}', logging.DEBUG,
f'({fake_pjlink.entry.name}) Processing PJLINK command'),
(f'{test_module}', logging.ERROR,
f'({fake_pjlink.entry.name}) Authenticate connection but no PIN - aborting')
]
fake_pjlink.pin = None
# WHEN: Calling function
caplog.clear()
@ -276,7 +264,6 @@ def test_authenticate_login(fake_pjlink, caplog):
# GIVEN: Test setup
caplog.set_level(logging.DEBUG)
t_data = f'1 {TEST_SALT}'
print(t_data)
logs = [(f'{test_module}', logging.DEBUG,
f'({fake_pjlink.entry.name}) Processing PJLINK command'),
(f'{test_module}', logging.DEBUG,

View File

@ -0,0 +1,120 @@
# -*- coding: utf-8 -*-
##########################################################################
# OpenLP - Open Source Lyrics Projection #
# ---------------------------------------------------------------------- #
# Copyright (c) 2008-2022 OpenLP Developers #
# ---------------------------------------------------------------------- #
# This program is free software: you can redistribute it and/or modify #
# it under the terms of the GNU General Public License as published by #
# the Free Software Foundation, either version 3 of the License, or #
# (at your option) any later version. #
# #
# This program is distributed in the hope that it will be useful, #
# but WITHOUT ANY WARRANTY; without even the implied warranty of #
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the #
# GNU General Public License for more details. #
# #
# You should have received a copy of the GNU General Public License #
# along with this program. If not, see <https://www.gnu.org/licenses/>. #
##########################################################################
"""
Tests for commands that do not need much testing
"""
import logging
import openlp.core.projectors.pjlinkcommands
from copy import deepcopy
from openlp.core.projectors.constants import PJLINK_SVER_MAX_LEN
from tests.resources.projector.data import TEST1_DATA, TEST2_DATA
test_module = openlp.core.projectors.pjlinkcommands.__name__
_process_sver = openlp.core.projectors.pjlinkcommands._process_sver
def test_sver_none(fake_pjlink, caplog):
"""
Test SVER update when saved version is None
"""
# GIVEN: Test setup
fake_pjlink.sw_version = None
fake_pjlink.db_update = False
t_data = deepcopy(TEST1_DATA['sw_version'])
caplog.set_level(logging.DEBUG)
logs = [(test_module, logging.DEBUG, f'({fake_pjlink.name}) Setting projector software version to "{t_data}"')]
# WHEN: Called
_process_sver(projector=fake_pjlink, data=t_data)
# THEN: Only log entry made
assert caplog.record_tuples == logs, 'Invalid log entries'
assert fake_pjlink.sw_version == t_data, 'sw_version should have updated'
assert fake_pjlink.db_update is True, 'db_update should be set'
def test_sver_same(fake_pjlink, caplog):
"""
Test SVER same as saved version
"""
# GIVEN: Test setup
fake_pjlink.sw_version = deepcopy(TEST1_DATA['sw_version'])
fake_pjlink.db_update = False
t_data = fake_pjlink.sw_version
caplog.set_level(logging.DEBUG)
logs = [(test_module, logging.DEBUG, f'({fake_pjlink.name}) Software version unchanged - returning')]
# WHEN: Called
_process_sver(projector=fake_pjlink, data=t_data)
# THEN: Only log entry made
assert caplog.record_tuples == logs, 'Invalid log entries'
assert fake_pjlink.sw_version == t_data, 'sw_version should not have changed'
assert fake_pjlink.db_update is False, 'db_update should not have changed'
def test_sver_too_long(fake_pjlink, caplog):
"""
Test SVER too long
"""
# GIVEN: Test setup
fake_pjlink.sw_version = None
fake_pjlink.db_update = False
t_data = '1' * PJLINK_SVER_MAX_LEN
t_data += 'z' # One longer than max length
caplog.set_level(logging.DEBUG)
logs = [(test_module, logging.WARNING, f'({fake_pjlink.name}) Invalid software version - too long')]
# WHEN: Called
caplog.clear()
_process_sver(projector=fake_pjlink, data=t_data)
# THEN: Only log entry made
assert caplog.record_tuples == logs, 'Invalid log entries'
assert fake_pjlink.sw_version is None, 'sw_version should not have changed'
assert fake_pjlink.db_update is False, 'db_update should not have changed'
def test_sver_update(fake_pjlink, caplog):
"""
Test SVER update when saved version is different than received version
"""
# GIVEN: Test setup
fake_pjlink.sw_version = deepcopy(TEST1_DATA['sw_version'])
fake_pjlink.db_update = False
t_data = deepcopy(TEST2_DATA['sw_version'])
caplog.set_level(logging.DEBUG)
logs = [(test_module, logging.DEBUG, f'({fake_pjlink.name}) Old software version "{fake_pjlink.sw_version}"'),
(test_module, logging.DEBUG, f'({fake_pjlink.name}) New software version "{t_data}"'),
(test_module, logging.DEBUG, f'({fake_pjlink.name}) Setting projector software version to "{t_data}"')]
# WHEN: Called
_process_sver(projector=fake_pjlink, data=t_data)
# THEN: Only log entry made
assert caplog.record_tuples == logs, 'Invalid log entries'
assert fake_pjlink.sw_version == t_data, 'sw_version should have updated'
assert fake_pjlink.db_update is True, 'db_update should be set'

View File

@ -28,6 +28,7 @@ from unittest.mock import patch
from openlp.core.projectors.db import Projector, ProjectorDB
from openlp.core.projectors.manager import ProjectorManager
from openlp.core.projectors.pjlink import PJLink
from tests.helpers.projector import FakePJLink
from tests.resources.projector.data import TEST_DB, TEST1_DATA
'''
@ -78,6 +79,13 @@ def projector_manager_mtdb(settings):
del t_manager
@pytest.fixture
def fake_pjlink():
faker = FakePJLink()
yield faker
del(faker)
@pytest.fixture()
def pjlink():
pj_link = PJLink(Projector(**TEST1_DATA), no_poll=True)

View File

@ -44,161 +44,6 @@ def test_projector_ackn(mock_log, pjlink):
mock_log.debug.assert_has_calls(log_debug_text)
@patch.object(openlp.core.projectors.pjlink.PJLink, 'projectorUpdateIcons')
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_avmt_audio_muted(mock_log, mock_UpdateIcons, pjlink):
"""
Test avmt status shutter unchanged and mute on
"""
# GIVEN: Test setup
log_warning_text = []
log_debug_text = [call('({ip}) Processing command "AVMT" with data "21"'.format(ip=pjlink.name)),
call('({ip}) Calling function for AVMT'.format(ip=pjlink.name)),
call('({ip}) Setting speaker to muted'.format(ip=pjlink.name))]
pjlink.shutter = True
pjlink.mute = False
# WHEN: Called with setting shutter closed and mute on
process_command(projector=pjlink, cmd='AVMT', data='21')
# THEN: Shutter should be closed and mute should be True
assert pjlink.shutter, 'Shutter should not have changed'
assert pjlink.mute, 'Audio should be off'
assert mock_UpdateIcons.emit.called, 'Update icons should have been called'
mock_log.warning.assert_has_calls(log_warning_text)
mock_log.debug.assert_has_calls(log_debug_text)
@patch.object(openlp.core.projectors.pjlink.PJLink, 'projectorUpdateIcons')
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_avmt_bad_data(mock_log, mock_UpdateIcons, pjlink):
"""
Test avmt bad data fail
"""
# GIVEN: Test object
log_warning_text = [call('({ip}) Invalid av mute response: 36'.format(ip=pjlink.name))]
log_debug_text = [call('({ip}) Processing command "AVMT" with data "36"'.format(ip=pjlink.name)),
call('({ip}) Calling function for AVMT'.format(ip=pjlink.name))]
pjlink.shutter = True
pjlink.mute = True
# WHEN: Called with an invalid setting
process_command(projector=pjlink, cmd='AVMT', data='36')
# THEN: Shutter should be closed and mute should be True
assert pjlink.shutter, 'Shutter should changed'
assert pjlink.mute, 'Audio should not have changed'
assert not mock_UpdateIcons.emit.called, 'Update icons should NOT have been called'
mock_log.warning.assert_has_calls(log_warning_text)
mock_log.debug.assert_has_calls(log_debug_text)
@patch.object(openlp.core.projectors.pjlink.PJLink, 'projectorUpdateIcons')
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_avmt_closed_muted(mock_log, mock_UpdateIcons, pjlink):
"""
Test avmt status shutter closed and mute off
"""
# GIVEN: Test object
log_warning_text = []
log_debug_text = [call('({ip}) Processing command "AVMT" with data "31"'.format(ip=pjlink.name)),
call('({ip}) Calling function for AVMT'.format(ip=pjlink.name)),
call('({ip}) Setting shutter to closed'.format(ip=pjlink.name)),
call('({ip}) Setting speaker to muted'.format(ip=pjlink.name))]
pjlink.shutter = False
pjlink.mute = False
# WHEN: Called with setting shutter to closed and mute on
process_command(projector=pjlink, cmd='AVMT', data='31')
# THEN: Shutter should be closed and mute should be True
assert pjlink.shutter, 'Shutter should have been set to closed'
assert pjlink.mute, 'Audio should be muted'
assert mock_UpdateIcons.emit.called, 'Update icons should have been called'
mock_log.warning.assert_has_calls(log_warning_text)
mock_log.debug.assert_has_calls(log_debug_text)
@patch.object(openlp.core.projectors.pjlink.PJLink, 'projectorUpdateIcons')
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_avmt_open_unmuted(mock_log, mock_UpdateIcons, pjlink):
"""
Test avmt status shutter open and mute off
"""
# GIVEN: Test object
log_warning_text = []
log_debug_text = [call('({ip}) Processing command "AVMT" with data "30"'.format(ip=pjlink.name)),
call('({ip}) Calling function for AVMT'.format(ip=pjlink.name)),
call('({ip}) Setting shutter to open'.format(ip=pjlink.name)),
call('({ip}) Setting speaker to normal'.format(ip=pjlink.name))]
pjlink.shutter = True
pjlink.mute = True
# WHEN: Called with setting shutter to closed and mute on
process_command(projector=pjlink, cmd='AVMT', data='30')
# THEN: Shutter should be closed and mute should be True
assert not pjlink.shutter, 'Shutter should have been set to off'
assert not pjlink.mute, 'Audio should be on'
assert mock_UpdateIcons.emit.called, 'Update icons should have been called'
mock_log.warning.assert_has_calls(log_warning_text)
mock_log.debug.assert_has_calls(log_debug_text)
@patch.object(openlp.core.projectors.pjlink.PJLink, 'projectorUpdateIcons')
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_avmt_shutter_closed(mock_log, mock_UpdateIcons, pjlink):
"""
Test avmt status shutter closed and audio unchanged
"""
# GIVEN: Test object
log_warning_text = []
log_debug_text = [call('({ip}) Processing command "AVMT" with data "11"'.format(ip=pjlink.name)),
call('({ip}) Calling function for AVMT'.format(ip=pjlink.name)),
call('({ip}) Setting shutter to closed'.format(ip=pjlink.name))]
pjlink.shutter = False
pjlink.mute = True
# WHEN: Called with setting shutter closed and mute off
process_command(projector=pjlink, cmd='AVMT', data='11')
# THEN: Shutter should be True and mute should be False
assert pjlink.shutter, 'Shutter should have been set to closed'
assert pjlink.mute, 'Audio should not have changed'
assert mock_UpdateIcons.emit.called, 'Update icons should have been called'
mock_log.warning.assert_has_calls(log_warning_text)
mock_log.debug.assert_has_calls(log_debug_text)
@patch.object(openlp.core.projectors.pjlink.PJLink, 'projectorUpdateIcons')
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_avmt_status_timer_check_delete(mock_log, mock_UpdateIcons, pjlink):
"""
Test avmt deletes callback in projector.status_timer_check
"""
# GIVEN: Test object
log_warning_text = []
log_debug_text = [call('({ip}) Processing command "AVMT" with data "11"'.format(ip=pjlink.name)),
call('({ip}) Calling function for AVMT'.format(ip=pjlink.name)),
call('({ip}) Setting shutter to closed'.format(ip=pjlink.name))]
pjlink.shutter = False
pjlink.mute = True
pjlink.status_timer_checks = {'AVMT': pjlink.get_av_mute_status}
# WHEN: Called with setting shutter closed and mute off
with patch.object(pjlink, 'status_timer') as mock_status_timer:
process_command(projector=pjlink, cmd='AVMT', data='11')
# THEN: Shutter should be True and mute should be False
assert pjlink.shutter, 'Shutter should have been set to closed'
assert pjlink.mute, 'Audio should not have changed'
assert mock_UpdateIcons.emit.called, 'Update icons should have been called'
assert 'AVMT' not in pjlink.status_timer_checks, 'Status timer list should not have AVMT callback'
assert mock_status_timer.stop.called, 'Projector status_timer.stop() should have been called'
mock_log.warning.assert_has_calls(log_warning_text)
mock_log.debug.assert_has_calls(log_debug_text)
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_erst_all_error(mock_log, pjlink):
"""

View File

@ -468,82 +468,3 @@ def test_projector_snum_set(mock_log, pjlink):
assert pjlink.serial_no == new_data, 'Projector serial number should have been set'
mock_log.warning.assert_has_calls(log_warning_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_sver_changed(mock_log, pjlink):
"""
Test invalid software version information - Received different than saved
"""
# GIVEN: Test object
old_data = 'Test 1 Subtest 1'
new_data = 'Test 1 Subtest 2'
log_warning_calls = [call('({ip}) Projector software version does not match '
'saved software version'.format(ip=pjlink.name)),
call('({ip}) Saved: "{data}"'.format(ip=pjlink.name, data=old_data)),
call('({ip}) Received: "{data}"'.format(ip=pjlink.name, data=new_data)),
call('({ip}) Updating software version'.format(ip=pjlink.name))]
log_debug_calls = [call('({ip}) Processing command "SVER" with data '
'"{data}"'.format(ip=pjlink.name, data=new_data)),
call('({ip}) Calling function for SVER'.format(ip=pjlink.name)),
call('({ip}) Setting projector software version to '
'"{data}"'.format(ip=pjlink.name, data=new_data))]
pjlink.sw_version = old_data
# WHEN: process_sver called with invalid data
process_command(pjlink, cmd='SVER', data=new_data)
# THEN: Version information should change
assert pjlink.sw_version == new_data, 'Software version should have changed'
mock_log.warning.assert_has_calls(log_warning_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_sver_invalid(mock_log, pjlink):
"""
Test invalid software version information - too long
"""
# GIVEN: Test object
new_data = 'This is a test software version line that is too long based on PJLink version 2 specs'
log_warning_calls = [call('Invalid software version - too long')]
log_debug_calls = [call('({ip}) Processing command "SVER" with data "{data}"'.format(ip=pjlink.name,
data=new_data)),
call('({ip}) Calling function for SVER'.format(ip=pjlink.name))]
pjlink.sw_version = None
pjlink.sw_version_received = None
# WHEN: process_sver called with invalid data
process_command(projector=pjlink, cmd='SVER', data=new_data)
# THEN: Version information should not change
assert not pjlink.sw_version, 'Software version should not have changed'
assert not pjlink.sw_version_received, 'Received software version should not have changed'
mock_log.warning.assert_has_calls(log_warning_calls)
mock_log.debug.assert_has_calls(log_debug_calls)
@patch.object(openlp.core.projectors.pjlinkcommands, 'log')
def test_projector_sver_save(mock_log, pjlink):
"""
Test invalid software version information - too long
"""
# GIVEN: Test object
new_data = 'Test 1 Subtest 1'
log_warning_calls = []
log_debug_calls = [call('({ip}) Processing command "SVER" with data '
'"{data}"'.format(ip=pjlink.name, data=new_data)),
call('({ip}) Calling function for SVER'.format(ip=pjlink.name)),
call('({ip}) Setting projector software version to '
'"{data}"'.format(ip=pjlink.name, data=new_data))]
pjlink.sw_version = None
pjlink.sw_version_received = None
# WHEN: process_sver called with invalid data
process_command(projector=pjlink, cmd='SVER', data=new_data)
# THEN: Version information should not change
assert pjlink.sw_version == new_data, 'Software version should have been updated'
assert not pjlink.sw_version_received, 'Received version field should not have changed'
mock_log.warning.assert_has_calls(log_warning_calls)
mock_log.debug.assert_has_calls(log_debug_calls)