forked from openlp/openlp
Replace the Memory check with a local server which stops and starts correctly.
One the 2nd instance pass the service file if one is included Stop the 2nd instance starting as it will fail due to port clashes and a monster thread issue(may be connected). Removed redundant code Add a number of new tests. lp:~trb143/openlp/localserver (revision 2850) https://ci.openlp.io/job/Branch-01-Pull/2501/ [SUCCESS] https://ci.openlp.io/job/Branch-02a-Linux-Tests/2402/ ... bzr-revno: 2816
This commit is contained in:
Tim Bentley
commit
fc8cee2a94
|
|
@ -1 +1 @@
|
|||
2.5.0
|
||||
2.9.0
|
||||
|
|
|
|||
|
|
@ -38,7 +38,6 @@ from PyQt5 import QtCore, QtWidgets
|
|||
from openlp.core.common import is_macosx, is_win
|
||||
from openlp.core.common.applocation import AppLocation
|
||||
from openlp.core.common.i18n import LanguageManager, UiStrings, translate
|
||||
from openlp.core.common.mixins import LogMixin
|
||||
from openlp.core.common.path import create_paths, copytree
|
||||
from openlp.core.common.registry import Registry
|
||||
from openlp.core.common.settings import Settings
|
||||
|
|
@ -50,6 +49,7 @@ from openlp.core.ui.firsttimeform import FirstTimeForm
|
|||
from openlp.core.ui.firsttimelanguageform import FirstTimeLanguageForm
|
||||
from openlp.core.ui.mainwindow import MainWindow
|
||||
from openlp.core.ui.style import get_application_stylesheet
|
||||
from openlp.core.server import Server
|
||||
from openlp.core.version import check_for_update, get_version
|
||||
|
||||
__all__ = ['OpenLP', 'main']
|
||||
|
|
@ -58,7 +58,7 @@ __all__ = ['OpenLP', 'main']
|
|||
log = logging.getLogger()
|
||||
|
||||
|
||||
class OpenLP(QtWidgets.QApplication, LogMixin):
|
||||
class OpenLP(QtWidgets.QApplication):
|
||||
"""
|
||||
The core application class. This class inherits from Qt's QApplication
|
||||
class in order to provide the core of the application.
|
||||
|
|
@ -72,7 +72,7 @@ class OpenLP(QtWidgets.QApplication, LogMixin):
|
|||
"""
|
||||
self.is_event_loop_active = True
|
||||
result = QtWidgets.QApplication.exec()
|
||||
self.shared_memory.detach()
|
||||
self.server.close_server()
|
||||
return result
|
||||
|
||||
def run(self, args):
|
||||
|
|
@ -135,23 +135,16 @@ class OpenLP(QtWidgets.QApplication, LogMixin):
|
|||
self.main_window.app_startup()
|
||||
return self.exec()
|
||||
|
||||
def is_already_running(self):
|
||||
@staticmethod
|
||||
def is_already_running():
|
||||
"""
|
||||
Look to see if OpenLP is already running and ask if a 2nd instance is to be started.
|
||||
Tell the user there is a 2nd instance running.
|
||||
"""
|
||||
self.shared_memory = QtCore.QSharedMemory('OpenLP')
|
||||
if self.shared_memory.attach():
|
||||
status = QtWidgets.QMessageBox.critical(None, UiStrings().Error, UiStrings().OpenLPStart,
|
||||
QtWidgets.QMessageBox.StandardButtons(QtWidgets.QMessageBox.Yes |
|
||||
QtWidgets.QMessageBox.No))
|
||||
if status == QtWidgets.QMessageBox.No:
|
||||
return True
|
||||
return False
|
||||
else:
|
||||
self.shared_memory.create(1)
|
||||
return False
|
||||
QtWidgets.QMessageBox.critical(None, UiStrings().Error, UiStrings().OpenLPStart,
|
||||
QtWidgets.QMessageBox.StandardButtons(QtWidgets.QMessageBox.Ok))
|
||||
|
||||
def is_data_path_missing(self):
|
||||
@staticmethod
|
||||
def is_data_path_missing():
|
||||
"""
|
||||
Check if the data folder path exists.
|
||||
"""
|
||||
|
|
@ -301,10 +294,7 @@ def parse_options(args=None):
|
|||
parser.add_argument('-l', '--log-level', dest='loglevel', default='warning', metavar='LEVEL',
|
||||
help='Set logging to LEVEL level. Valid values are "debug", "info", "warning".')
|
||||
parser.add_argument('-p', '--portable', dest='portable', action='store_true',
|
||||
help='Specify if this should be run as a portable app, '
|
||||
'off a USB flash drive (not implemented).')
|
||||
parser.add_argument('-d', '--dev-version', dest='dev_version', action='store_true',
|
||||
help='Ignore the version file and pull the version directly from Bazaar')
|
||||
help='Specify if this should be run as a portable app, ')
|
||||
parser.add_argument('-w', '--no-web-server', dest='no_web_server', action='store_true',
|
||||
help='Turn off the Web and Socket Server ')
|
||||
parser.add_argument('rargs', nargs='?', default=[])
|
||||
|
|
@ -383,11 +373,17 @@ def main(args=None):
|
|||
Registry().set_flag('no_web_server', args.no_web_server)
|
||||
application.setApplicationVersion(get_version()['version'])
|
||||
# Check if an instance of OpenLP is already running. Quit if there is a running instance and the user only wants one
|
||||
if application.is_already_running():
|
||||
server = Server()
|
||||
if server.is_another_instance_running():
|
||||
application.is_already_running()
|
||||
server.post_to_server(qt_args)
|
||||
sys.exit()
|
||||
else:
|
||||
server.start_server()
|
||||
application.server = server
|
||||
# If the custom data path is missing and the user wants to restore the data path, quit OpenLP.
|
||||
if application.is_data_path_missing():
|
||||
application.shared_memory.detach()
|
||||
server.close_server()
|
||||
sys.exit()
|
||||
# Upgrade settings.
|
||||
settings = Settings()
|
||||
|
|
|
|||
|
|
@ -415,7 +415,7 @@ class UiStrings(object):
|
|||
self.NoResults = translate('OpenLP.Ui', 'No Search Results')
|
||||
self.OpenLP = translate('OpenLP.Ui', 'OpenLP')
|
||||
self.OpenLPv2AndUp = translate('OpenLP.Ui', 'OpenLP 2.0 and up')
|
||||
self.OpenLPStart = translate('OpenLP.Ui', 'OpenLP is already running. Do you wish to continue?')
|
||||
self.OpenLPStart = translate('OpenLP.Ui', 'OpenLP is already running on this machine. \nClosing this instance')
|
||||
self.OpenService = translate('OpenLP.Ui', 'Open service.')
|
||||
self.OptionalShowInFooter = translate('OpenLP.Ui', 'Optional, this will be displayed in footer.')
|
||||
self.OptionalHideInFooter = translate('OpenLP.Ui', 'Optional, this won\'t be displayed in footer.')
|
||||
|
|
|
|||
|
|
@ -0,0 +1,109 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
# vim: autoindent shiftwidth=4 expandtab textwidth=120 tabstop=4 softtabstop=4
|
||||
|
||||
###############################################################################
|
||||
# OpenLP - Open Source Lyrics Projection #
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Copyright (c) 2008-2018 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; version 2 of the License. #
|
||||
# #
|
||||
# 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, write to the Free Software Foundation, Inc., 59 #
|
||||
# Temple Place, Suite 330, Boston, MA 02111-1307 USA #
|
||||
###############################################################################
|
||||
from PyQt5 import QtCore, QtNetwork
|
||||
|
||||
from openlp.core.common.registry import Registry
|
||||
from openlp.core.common.mixins import LogMixin
|
||||
|
||||
|
||||
class Server(QtCore.QObject, LogMixin):
|
||||
"""
|
||||
The local server to handle OpenLP running in more than one instance and allows file
|
||||
handles to be transferred from the new to the existing one.
|
||||
"""
|
||||
def __init__(self):
|
||||
super(Server, self).__init__()
|
||||
self.out_socket = QtNetwork.QLocalSocket()
|
||||
self.server = None
|
||||
self.id = 'OpenLPDual'
|
||||
|
||||
def is_another_instance_running(self):
|
||||
"""
|
||||
Check the see if an other instance is running
|
||||
:return: True of False
|
||||
"""
|
||||
# Is there another instance running?
|
||||
self.out_socket.connectToServer(self.id)
|
||||
return self.out_socket.waitForConnected()
|
||||
|
||||
def post_to_server(self, args):
|
||||
"""
|
||||
Post the file name to the over instance
|
||||
:param args: The passed arguments including maybe a file name
|
||||
"""
|
||||
if 'OpenLP' in args:
|
||||
args.remove('OpenLP')
|
||||
# Yes, there is.
|
||||
self.out_stream = QtCore.QTextStream(self.out_socket)
|
||||
self.out_stream.setCodec('UTF-8')
|
||||
self.out_socket.write(str.encode("".join(args)))
|
||||
if not self.out_socket.waitForBytesWritten(10):
|
||||
raise Exception(str(self.out_socket.errorString()))
|
||||
self.out_socket.disconnectFromServer()
|
||||
|
||||
def start_server(self):
|
||||
"""
|
||||
Start the socket server to allow inter app communication
|
||||
:return:
|
||||
"""
|
||||
self.out_socket = None
|
||||
self.out_stream = None
|
||||
self.in_socket = None
|
||||
self.in_stream = None
|
||||
self.server = QtNetwork.QLocalServer()
|
||||
self.server.listen(self.id)
|
||||
self.server.newConnection.connect(self._on_new_connection)
|
||||
return True
|
||||
|
||||
def _on_new_connection(self):
|
||||
"""
|
||||
Handle a new connection to the server
|
||||
:return:
|
||||
"""
|
||||
if self.in_socket:
|
||||
self.in_socket.readyRead.disconnect(self._on_ready_read)
|
||||
self.in_socket = self.server.nextPendingConnection()
|
||||
if not self.in_socket:
|
||||
return
|
||||
self.in_stream = QtCore.QTextStream(self.in_socket)
|
||||
self.in_stream.setCodec('UTF-8')
|
||||
self.in_socket.readyRead.connect(self._on_ready_read)
|
||||
|
||||
def _on_ready_read(self):
|
||||
"""
|
||||
Read a record passed to the server and pass to the service manager to handle
|
||||
:return:
|
||||
"""
|
||||
msg = self.in_stream.readLine()
|
||||
if msg:
|
||||
self.log_debug("socket msg = " + msg)
|
||||
Registry().get('service_manager').on_load_service_clicked(msg)
|
||||
|
||||
def close_server(self):
|
||||
"""
|
||||
Shutdown to local socket server and make sure the server is removed.
|
||||
:return:
|
||||
"""
|
||||
if self.server:
|
||||
self.server.close()
|
||||
# Make sure the server file is removed.
|
||||
QtNetwork.QLocalServer.removeServer(self.id)
|
||||
|
|
@ -136,48 +136,12 @@ def get_version():
|
|||
global APPLICATION_VERSION
|
||||
if APPLICATION_VERSION:
|
||||
return APPLICATION_VERSION
|
||||
if '--dev-version' in sys.argv or '-d' in sys.argv:
|
||||
# NOTE: The following code is a duplicate of the code in setup.py. Any fix applied here should also be applied
|
||||
# there.
|
||||
|
||||
# Get the revision of this tree.
|
||||
bzr = Popen(('bzr', 'revno'), stdout=PIPE)
|
||||
tree_revision, error = bzr.communicate()
|
||||
tree_revision = tree_revision.decode()
|
||||
code = bzr.wait()
|
||||
if code != 0:
|
||||
raise Exception('Error running bzr log')
|
||||
|
||||
# Get all tags.
|
||||
bzr = Popen(('bzr', 'tags'), stdout=PIPE)
|
||||
output, error = bzr.communicate()
|
||||
code = bzr.wait()
|
||||
if code != 0:
|
||||
raise Exception('Error running bzr tags')
|
||||
tags = list(map(bytes.decode, output.splitlines()))
|
||||
if not tags:
|
||||
tag_version = '0.0.0'
|
||||
tag_revision = '0'
|
||||
else:
|
||||
# Remove any tag that has "?" as revision number. A "?" as revision number indicates, that this tag is from
|
||||
# another series.
|
||||
tags = [tag for tag in tags if tag.split()[-1].strip() != '?']
|
||||
# Get the last tag and split it in a revision and tag name.
|
||||
tag_version, tag_revision = tags[-1].split()
|
||||
# If they are equal, then this tree is tarball with the source for the release. We do not want the revision
|
||||
# number in the full version.
|
||||
if tree_revision == tag_revision:
|
||||
full_version = tag_version.strip()
|
||||
else:
|
||||
full_version = '{tag}-bzr{tree}'.format(tag=tag_version.strip(), tree=tree_revision.strip())
|
||||
else:
|
||||
# We're not running the development version, let's use the file.
|
||||
file_path = AppLocation.get_directory(AppLocation.VersionDir) / '.version'
|
||||
try:
|
||||
full_version = file_path.read_text().rstrip()
|
||||
except OSError:
|
||||
log.exception('Error in version file.')
|
||||
full_version = '0.0.0-bzr000'
|
||||
file_path = AppLocation.get_directory(AppLocation.VersionDir) / '.version'
|
||||
try:
|
||||
full_version = file_path.read_text().rstrip()
|
||||
except OSError:
|
||||
log.exception('Error in version file.')
|
||||
full_version = '0.0.0-bzr000'
|
||||
bits = full_version.split('-')
|
||||
APPLICATION_VERSION = {
|
||||
'full': full_version,
|
||||
|
|
|
|||
|
|
@ -0,0 +1,151 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
# vim: autoindent shiftwidth=4 expandtab textwidth=120 tabstop=4 softtabstop=4
|
||||
|
||||
###############################################################################
|
||||
# OpenLP - Open Source Lyrics Projection #
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Copyright (c) 2008-2018 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; version 2 of the License. #
|
||||
# #
|
||||
# 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, write to the Free Software Foundation, Inc., 59 #
|
||||
# Temple Place, Suite 330, Boston, MA 02111-1307 USA #
|
||||
###############################################################################
|
||||
"""
|
||||
Functional tests to test the Http init.
|
||||
"""
|
||||
from unittest import TestCase
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from openlp.core.api.http import check_auth, requires_auth, authenticate
|
||||
from openlp.core.common.registry import Registry
|
||||
from openlp.core.common.settings import Settings
|
||||
|
||||
from tests.helpers.testmixin import TestMixin
|
||||
|
||||
|
||||
class TestInit(TestCase, TestMixin):
|
||||
"""
|
||||
A test suite to test the functions on the init
|
||||
"""
|
||||
|
||||
def setUp(self):
|
||||
"""
|
||||
Create the UI
|
||||
"""
|
||||
Registry().create()
|
||||
Registry().register('service_list', MagicMock())
|
||||
self.build_settings()
|
||||
self.password = 'c3VwZXJmbHk6bGFtYXM='
|
||||
|
||||
def tearDown(self):
|
||||
self.destroy_settings()
|
||||
|
||||
def test_auth(self):
|
||||
"""
|
||||
Test the check_auth method with a match
|
||||
:return:
|
||||
"""
|
||||
# GIVEN: a known user
|
||||
Settings().setValue('api/user id', "superfly")
|
||||
Settings().setValue('api/password', "lamas")
|
||||
|
||||
# WHEN : I check the authorisation
|
||||
is_valid = check_auth(['aaaaa', self.password])
|
||||
|
||||
# THEN:
|
||||
assert is_valid is True
|
||||
|
||||
def test_auth_falure(self):
|
||||
"""
|
||||
Test the check_auth method with a match
|
||||
:return:
|
||||
"""
|
||||
# GIVEN: a known user
|
||||
Settings().setValue('api/user id', 'superfly')
|
||||
Settings().setValue('api/password', 'lamas')
|
||||
|
||||
# WHEN : I check the authorisation
|
||||
is_valid = check_auth(['aaaaa', 'monkey123'])
|
||||
|
||||
# THEN:
|
||||
assert is_valid is False
|
||||
|
||||
def test_requires_auth_disabled(self):
|
||||
"""
|
||||
Test the requires_auth wrapper with disabled security
|
||||
:return:
|
||||
"""
|
||||
# GIVEN: A disabled security
|
||||
Settings().setValue('api/authentication enabled', False)
|
||||
|
||||
# WHEN: I call the function
|
||||
wrapped_function = requires_auth(func)
|
||||
value = wrapped_function()
|
||||
|
||||
# THEN: the result will be as expected
|
||||
assert value == 'called'
|
||||
|
||||
def test_requires_auth_enabled(self):
|
||||
"""
|
||||
Test the requires_auth wrapper with enabled security
|
||||
:return:
|
||||
"""
|
||||
# GIVEN: A disabled security
|
||||
Settings().setValue('api/authentication enabled', True)
|
||||
|
||||
# WHEN: I call the function
|
||||
wrapped_function = requires_auth(func)
|
||||
req = MagicMock()
|
||||
value = wrapped_function(req)
|
||||
|
||||
# THEN: the result will be as expected
|
||||
assert str(value) == str(authenticate())
|
||||
|
||||
def test_requires_auth_enabled_auth_error(self):
|
||||
"""
|
||||
Test the requires_auth wrapper with enabled security and authorization taken place and and error
|
||||
:return:
|
||||
"""
|
||||
# GIVEN: A enabled security
|
||||
Settings().setValue('api/authentication enabled', True)
|
||||
|
||||
# WHEN: I call the function with the wrong password
|
||||
wrapped_function = requires_auth(func)
|
||||
req = MagicMock()
|
||||
req.authorization = ['Basic', 'cccccccc']
|
||||
value = wrapped_function(req)
|
||||
|
||||
# THEN: the result will be as expected - try again
|
||||
assert str(value) == str(authenticate())
|
||||
|
||||
def test_requires_auth_enabled_auth(self):
|
||||
"""
|
||||
Test the requires_auth wrapper with enabled security and authorization taken place and and error
|
||||
:return:
|
||||
"""
|
||||
# GIVEN: An enabled security and a known user
|
||||
Settings().setValue('api/authentication enabled', True)
|
||||
Settings().setValue('api/user id', 'superfly')
|
||||
Settings().setValue('api/password', 'lamas')
|
||||
|
||||
# WHEN: I call the function with the wrong password
|
||||
wrapped_function = requires_auth(func)
|
||||
req = MagicMock()
|
||||
req.authorization = ['Basic', self.password]
|
||||
value = wrapped_function(req)
|
||||
|
||||
# THEN: the result will be as expected - try again
|
||||
assert str(value) == 'called'
|
||||
|
||||
|
||||
def func(field=None):
|
||||
return 'called'
|
||||
|
|
@ -41,7 +41,6 @@ def test_parse_options_basic():
|
|||
args = parse_options()
|
||||
|
||||
# THEN: the following fields will have been extracted.
|
||||
assert args.dev_version is False, 'The dev_version flag should be False'
|
||||
assert args.loglevel == 'warning', 'The log level should be set to warning'
|
||||
assert args.no_error_form is False, 'The no_error_form should be set to False'
|
||||
assert args.portable is False, 'The portable flag should be set to false'
|
||||
|
|
@ -59,7 +58,6 @@ def test_parse_options_debug():
|
|||
args = parse_options()
|
||||
|
||||
# THEN: the following fields will have been extracted.
|
||||
assert args.dev_version is False, 'The dev_version flag should be False'
|
||||
assert args.loglevel == ' debug', 'The log level should be set to debug'
|
||||
assert args.no_error_form is False, 'The no_error_form should be set to False'
|
||||
assert args.portable is False, 'The portable flag should be set to false'
|
||||
|
|
@ -77,7 +75,6 @@ def test_parse_options_debug_and_portable():
|
|||
args = parse_options()
|
||||
|
||||
# THEN: the following fields will have been extracted.
|
||||
assert args.dev_version is False, 'The dev_version flag should be False'
|
||||
assert args.loglevel == 'warning', 'The log level should be set to warning'
|
||||
assert args.no_error_form is False, 'The no_error_form should be set to False'
|
||||
assert args.portable is True, 'The portable flag should be set to true'
|
||||
|
|
@ -89,16 +86,15 @@ def test_parse_options_all_no_file():
|
|||
Test the parse options process works with two options
|
||||
"""
|
||||
# GIVEN: a a set of system arguments.
|
||||
sys.argv[1:] = ['-l debug', '-d']
|
||||
sys.argv[1:] = ['-l debug', '-p']
|
||||
|
||||
# WHEN: We we parse them to expand to options
|
||||
args = parse_options()
|
||||
|
||||
# THEN: the following fields will have been extracted.
|
||||
assert args.dev_version is True, 'The dev_version flag should be True'
|
||||
assert args.loglevel == ' debug', 'The log level should be set to debug'
|
||||
assert args.no_error_form is False, 'The no_error_form should be set to False'
|
||||
assert args.portable is False, 'The portable flag should be set to false'
|
||||
assert args.portable is True, 'The portable flag should be set to false'
|
||||
assert args.rargs == [], 'The service file should be blank'
|
||||
|
||||
|
||||
|
|
@ -113,7 +109,6 @@ def test_parse_options_file():
|
|||
args = parse_options()
|
||||
|
||||
# THEN: the following fields will have been extracted.
|
||||
assert args.dev_version is False, 'The dev_version flag should be False'
|
||||
assert args.loglevel == 'warning', 'The log level should be set to warning'
|
||||
assert args.no_error_form is False, 'The no_error_form should be set to False'
|
||||
assert args.portable is False, 'The portable flag should be set to false'
|
||||
|
|
@ -131,7 +126,6 @@ def test_parse_options_file_and_debug():
|
|||
args = parse_options()
|
||||
|
||||
# THEN: the following fields will have been extracted.
|
||||
assert args.dev_version is False, 'The dev_version flag should be False'
|
||||
assert args.loglevel == ' debug', 'The log level should be set to debug'
|
||||
assert args.no_error_form is False, 'The no_error_form should be set to False'
|
||||
assert args.portable is False, 'The portable flag should be set to false'
|
||||
|
|
|
|||
|
|
@ -0,0 +1,114 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
# vim: autoindent shiftwidth=4 expandtab textwidth=120 tabstop=4 softtabstop=4
|
||||
|
||||
###############################################################################
|
||||
# OpenLP - Open Source Lyrics Projection #
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Copyright (c) 2008-2018 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; version 2 of the License. #
|
||||
# #
|
||||
# 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, write to the Free Software Foundation, Inc., 59 #
|
||||
# Temple Place, Suite 330, Boston, MA 02111-1307 USA #
|
||||
###############################################################################
|
||||
from unittest import TestCase
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from openlp.core.server import Server
|
||||
from openlp.core.common.registry import Registry
|
||||
|
||||
from tests.helpers.testmixin import TestMixin
|
||||
|
||||
|
||||
class TestServer(TestCase, TestMixin):
|
||||
"""
|
||||
Test the Server Class used to check if OpenLP is running.
|
||||
"""
|
||||
def setUp(self):
|
||||
Registry.create()
|
||||
with patch('PyQt5.QtNetwork.QLocalSocket'):
|
||||
self.server = Server()
|
||||
|
||||
def tearDown(self):
|
||||
self.server.close_server()
|
||||
|
||||
def test_is_another_instance_running(self):
|
||||
"""
|
||||
Run a test as if this was the first time and no instance is running
|
||||
"""
|
||||
# GIVEN: A running Server
|
||||
|
||||
# WHEN: I ask for it to start
|
||||
value = self.server.is_another_instance_running()
|
||||
|
||||
# THEN the following is called
|
||||
self.server.out_socket.waitForConnected.assert_called_once_with()
|
||||
self.server.out_socket.connectToServer.assert_called_once_with(self.server.id)
|
||||
assert isinstance(value, MagicMock)
|
||||
|
||||
def test_is_another_instance_running_true(self):
|
||||
"""
|
||||
Run a test as if there is another instance running
|
||||
"""
|
||||
# GIVEN: A running Server
|
||||
self.server.out_socket.waitForConnected.return_value = True
|
||||
|
||||
# WHEN: I ask for it to start
|
||||
value = self.server.is_another_instance_running()
|
||||
|
||||
# THEN the following is called
|
||||
self.server.out_socket.waitForConnected.assert_called_once_with()
|
||||
self.server.out_socket.connectToServer.assert_called_once_with(self.server.id)
|
||||
assert value is True
|
||||
|
||||
def test_on_read_ready(self):
|
||||
"""
|
||||
Test the on_read_ready method calls the service_manager
|
||||
"""
|
||||
# GIVEN: A server with a service manager
|
||||
self.server.in_stream = MagicMock()
|
||||
service_manager = MagicMock()
|
||||
Registry().register('service_manager', service_manager)
|
||||
|
||||
# WHEN: a file is added to the socket and the method called
|
||||
file_name = '\\home\\superfly\\'
|
||||
self.server.in_stream.readLine.return_value = file_name
|
||||
self.server._on_ready_read()
|
||||
|
||||
# THEN: the service will be loaded
|
||||
assert service_manager.on_load_service_clicked.call_count == 1
|
||||
service_manager.on_load_service_clicked.assert_called_once_with(file_name)
|
||||
|
||||
@patch("PyQt5.QtCore.QTextStream")
|
||||
def test_post_to_server(self, mocked_stream):
|
||||
"""
|
||||
A Basic test with a post to the service
|
||||
:return:
|
||||
"""
|
||||
# GIVEN: A server
|
||||
# WHEN: I post to a server
|
||||
self.server.post_to_server(['l', 'a', 'm', 'a', 's'])
|
||||
|
||||
# THEN: the file should be passed out to the socket
|
||||
self.server.out_socket.write.assert_called_once_with(b'lamas')
|
||||
|
||||
@patch("PyQt5.QtCore.QTextStream")
|
||||
def test_post_to_server_openlp(self, mocked_stream):
|
||||
"""
|
||||
A Basic test with a post to the service with OpenLP
|
||||
:return:
|
||||
"""
|
||||
# GIVEN: A server
|
||||
# WHEN: I post to a server
|
||||
self.server.post_to_server(['l', 'a', 'm', 'a', 's', 'OpenLP'])
|
||||
|
||||
# THEN: the file should be passed out to the socket
|
||||
self.server.out_socket.write.assert_called_once_with(b'lamas')
|
||||
Loading…
Reference in New Issue