forked from openlp/openlp
start to add tests
This commit is contained in:
parent
fada29080c
commit
c6076e1054
@ -19,10 +19,7 @@
|
||||
# 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
|
||||
from PyQt5 import QtNetwork
|
||||
from PyQt5 import QtCore, QtNetwork
|
||||
|
||||
from openlp.core.common.registry import Registry
|
||||
from openlp.core.common.mixins import LogMixin
|
||||
@ -33,35 +30,38 @@ 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):
|
||||
self._id = 'OpenLPDual'
|
||||
# Is there another instance running?
|
||||
self._outSocket = QtNetwork.QLocalSocket()
|
||||
self._outSocket.connectToServer(self._id)
|
||||
return self._outSocket.waitForConnected()
|
||||
self.out_socket.connectToServer(self.id)
|
||||
return self.out_socket.waitForConnected()
|
||||
|
||||
def post_to_server(self, args):
|
||||
if 'OpenLP' in args:
|
||||
args.remove('OpenLP')
|
||||
# Yes, there is.
|
||||
self._outStream = QtCore.QTextStream(self._outSocket)
|
||||
self._outStream.setCodec('UTF-8')
|
||||
self._outSocket.write(str.encode("".join(args)))
|
||||
if not self._outSocket.waitForBytesWritten(10):
|
||||
raise Exception(str(self._outSocket.errorString()))
|
||||
self._outSocket.disconnectFromServer()
|
||||
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()
|
||||
return False
|
||||
|
||||
def start_server(self):
|
||||
# No, there isn't.
|
||||
self._outSocket = None
|
||||
self._outStream = None
|
||||
self._inSocket = None
|
||||
self._inStream = None
|
||||
self._server = QtNetwork.QLocalServer()
|
||||
self._server.listen(self._id)
|
||||
self._server.newConnection.connect(self._on_new_connection)
|
||||
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):
|
||||
@ -69,21 +69,21 @@ class Server(QtCore.QObject, LogMixin):
|
||||
Handle a new connection to the server
|
||||
:return:
|
||||
"""
|
||||
if self._inSocket:
|
||||
self._inSocket.readyRead.disconnect(self._on_ready_read)
|
||||
self._inSocket = self._server.nextPendingConnection()
|
||||
if not self._inSocket:
|
||||
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._inStream = QtCore.QTextStream(self._inSocket)
|
||||
self._inStream.setCodec('UTF-8')
|
||||
self._inSocket.readyRead.connect(self._on_ready_read)
|
||||
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 load a service
|
||||
:return:
|
||||
"""
|
||||
msg = self._inStream.readLine()
|
||||
msg = self.in_stream.readLine()
|
||||
if msg:
|
||||
self.log_debug("socket msg = " + msg)
|
||||
Registry().get('service_manager').on_load_service_clicked(msg)
|
||||
@ -93,7 +93,7 @@ class Server(QtCore.QObject, LogMixin):
|
||||
Shutdown to local socket server and make sure the server is removed.
|
||||
:return:
|
||||
"""
|
||||
if self._server:
|
||||
self._server.close()
|
||||
if self.server:
|
||||
self.server.close()
|
||||
# Make sure the server file is removed.
|
||||
QtNetwork.QLocalServer.removeServer(self._id)
|
||||
QtNetwork.QLocalServer.removeServer(self.id)
|
||||
|
59
tests/functional/openlp_core/test_server.py
Normal file
59
tests/functional/openlp_core/test_server.py
Normal file
@ -0,0 +1,59 @@
|
||||
# -*- 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 #
|
||||
###############################################################################
|
||||
import sys
|
||||
from unittest import TestCase, skip
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from PyQt5 import QtCore, QtWidgets
|
||||
|
||||
from openlp.core.app import OpenLP, parse_options
|
||||
from openlp.core.common.settings import Settings
|
||||
from openlp.core.server import Server
|
||||
from tests.utils.constants import RESOURCE_PATH
|
||||
from tests.helpers.testmixin import TestMixin
|
||||
|
||||
|
||||
class TestServer(TestCase, TestMixin):
|
||||
"""
|
||||
Test the OpenLP app class
|
||||
"""
|
||||
def setUp(self):
|
||||
#self.setup_application()
|
||||
#self.build_settings()
|
||||
#self.openlp = OpenLP([])
|
||||
with patch('PyQt5.QtNetwork.QLocalSocket'):
|
||||
self.server = Server()
|
||||
|
||||
def tearDown(self):
|
||||
#self.destroy_settings()
|
||||
#del self.openlp
|
||||
#self.openlp = None
|
||||
self.server.close_server()
|
||||
pass
|
||||
|
||||
def test_is_another_instance_running(self):
|
||||
# GIVEN: A running Server
|
||||
# WHEN: I ask for it to start
|
||||
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)
|
Loading…
Reference in New Issue
Block a user