forked from openlp/openlp
110 lines
4.9 KiB
Python
110 lines
4.9 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
##########################################################################
|
|
# OpenLP - Open Source Lyrics Projection #
|
|
# ---------------------------------------------------------------------- #
|
|
# Copyright (c) 2008-2020 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/>. #
|
|
##########################################################################
|
|
from unittest.mock import MagicMock, call, patch
|
|
|
|
from openlp.core.api.zeroconf import ZeroconfWorker, start_zeroconf
|
|
|
|
|
|
@patch('openlp.core.api.zeroconf.socket.inet_aton')
|
|
def test_zeroconf_worker_constructor(mocked_inet_aton):
|
|
"""Test creating the Zeroconf worker object"""
|
|
# GIVEN: A ZeroconfWorker class and a mocked inet_aton
|
|
mocked_inet_aton.return_value = 'processed_ip'
|
|
|
|
# WHEN: An instance of the ZeroconfWorker is created
|
|
worker = ZeroconfWorker('127.0.0.1', 8000, 8001)
|
|
|
|
# THEN: The inet_aton function should have been called and the attrs should be set
|
|
mocked_inet_aton.assert_called_once_with('127.0.0.1')
|
|
assert worker.address == 'processed_ip'
|
|
assert worker.http_port == 8000
|
|
assert worker.ws_port == 8001
|
|
|
|
|
|
@patch('openlp.core.api.zeroconf.ServiceInfo')
|
|
@patch('openlp.core.api.zeroconf.Zeroconf')
|
|
def test_zeroconf_worker_start(MockedZeroconf, MockedServiceInfo):
|
|
"""Test the start() method of ZeroconfWorker"""
|
|
# GIVEN: A few mocks and a ZeroconfWorker instance with a mocked can_run method
|
|
mocked_http_info = MagicMock()
|
|
mocked_ws_info = MagicMock()
|
|
mocked_zc = MagicMock()
|
|
MockedServiceInfo.side_effect = [mocked_http_info, mocked_ws_info]
|
|
MockedZeroconf.return_value = mocked_zc
|
|
worker = ZeroconfWorker('127.0.0.1', 8000, 8001)
|
|
|
|
# WHEN: The start() method is called
|
|
with patch.object(worker, 'can_run') as mocked_can_run:
|
|
mocked_can_run.side_effect = [True, False]
|
|
worker.start()
|
|
|
|
# THEN: The correct calls are made
|
|
assert MockedServiceInfo.call_args_list == [
|
|
call('_http._tcp.local.', 'OpenLP._http._tcp.local.', address=b'\x7f\x00\x00\x01', port=8000, properties={}),
|
|
call('_ws._tcp.local.', 'OpenLP._ws._tcp.local.', address=b'\x7f\x00\x00\x01', port=8001, properties={})
|
|
]
|
|
assert MockedZeroconf.call_count == 1
|
|
assert mocked_zc.register_service.call_args_list == [call(mocked_http_info), call(mocked_ws_info)]
|
|
assert mocked_can_run.call_count == 2
|
|
assert mocked_zc.unregister_service.call_args_list == [call(mocked_http_info), call(mocked_ws_info)]
|
|
assert mocked_zc.close.call_count == 1
|
|
|
|
|
|
def test_zeroconf_worker_stop():
|
|
"""Test that the ZeroconfWorker.stop() method correctly stops the service"""
|
|
# GIVEN: A worker object with _can_run set to True
|
|
worker = ZeroconfWorker('127.0.0.1', 8000, 8001)
|
|
worker._can_run = True
|
|
|
|
# WHEN: stop() is called
|
|
worker.stop()
|
|
|
|
# THEN: _can_run should be False
|
|
assert worker.can_run() is False
|
|
|
|
|
|
@patch('openlp.core.api.zeroconf.get_network_interfaces')
|
|
@patch('openlp.core.api.zeroconf.Registry')
|
|
@patch('openlp.core.api.zeroconf.ZeroconfWorker')
|
|
@patch('openlp.core.api.zeroconf.run_thread')
|
|
def test_start_zeroconf(mocked_run_thread, MockedZeroconfWorker, MockedRegistry, mocked_get_network_interfaces):
|
|
"""Test the start_zeroconf() function"""
|
|
# GIVEN: A whole bunch of stuff that's mocked out
|
|
mocked_get_network_interfaces.return_value = {
|
|
'eth0': {
|
|
'ip': '192.168.1.191',
|
|
'broadcast': '192.168.1.255',
|
|
'netmask': '255.255.255.0',
|
|
'prefix': 24,
|
|
'localnet': '192.168.1.0'
|
|
}
|
|
}
|
|
MockedRegistry.return_value.get_flag.return_value = False
|
|
MockedRegistry.settings.return_value.value.side_effect = [8000, 8001]
|
|
mocked_worker = MagicMock()
|
|
MockedZeroconfWorker.return_value = mocked_worker
|
|
|
|
# WHEN: start_zeroconf() is called
|
|
start_zeroconf()
|
|
|
|
# THEN: A worker is added to the list of threads
|
|
mocked_run_thread.assert_called_once_with(mocked_worker, 'api_zeroconf_eth0')
|