forked from openlp/openlp
Further refactor the code, and fix up the tests
This commit is contained in:
parent
f27e4f8821
commit
11ad9dbb22
@ -24,7 +24,7 @@ The :mod:`~openlp.core.api.tab` module contains the settings tab for the API
|
||||
"""
|
||||
from PyQt5 import QtCore, QtGui, QtWidgets
|
||||
|
||||
from openlp.core.common import get_local_ip4
|
||||
from openlp.core.common import get_network_interfaces
|
||||
from openlp.core.common.i18n import UiStrings, translate
|
||||
from openlp.core.common.registry import Registry
|
||||
from openlp.core.common.settings import Settings
|
||||
@ -194,8 +194,7 @@ class ApiTab(SettingsTab):
|
||||
http_url_temp = http_url + 'main'
|
||||
self.live_url.setText('<a href="{url}">{url}</a>'.format(url=http_url_temp))
|
||||
|
||||
@staticmethod
|
||||
def get_ip_address(ip_address):
|
||||
def get_ip_address(self, ip_address):
|
||||
"""
|
||||
returns the IP address in dependency of the passed address
|
||||
ip_address == 0.0.0.0: return the IP address of the first valid interface
|
||||
@ -203,9 +202,8 @@ class ApiTab(SettingsTab):
|
||||
"""
|
||||
if ip_address == ZERO_URL:
|
||||
# In case we have more than one interface
|
||||
ifaces = get_local_ip4()
|
||||
for key in iter(ifaces):
|
||||
ip_address = ifaces.get(key)['ip']
|
||||
for _, interface in get_network_interfaces().items():
|
||||
ip_address = interface['ip']
|
||||
# We only want the first interface returned
|
||||
break
|
||||
return ip_address
|
||||
|
@ -23,23 +23,16 @@
|
||||
The :mod:`~openlp.core.api.zeroconf` module runs a Zerconf server so that OpenLP can advertise the
|
||||
RESTful API for devices on the network to discover.
|
||||
"""
|
||||
import logging
|
||||
import re
|
||||
import socket
|
||||
from time import sleep
|
||||
|
||||
from zeroconf import ServiceInfo, Zeroconf
|
||||
|
||||
from openlp.core.common import get_local_ip4
|
||||
from openlp.core.common import get_network_interfaces
|
||||
from openlp.core.common.registry import Registry
|
||||
from openlp.core.common.settings import Settings
|
||||
from openlp.core.threading import ThreadWorker, run_thread
|
||||
|
||||
# Skip names like "docker0", "tun0", "loopback_0", etc
|
||||
INTERFACE_FILTER = re.compile('loopback|docker|tun', re.IGNORECASE)
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class ZeroconfWorker(ThreadWorker):
|
||||
"""
|
||||
@ -92,19 +85,6 @@ class ZeroconfWorker(ThreadWorker):
|
||||
self._can_run = False
|
||||
|
||||
|
||||
def filter_interfaces():
|
||||
"""
|
||||
Return a list of interfaces filtered down to valid interfaces
|
||||
"""
|
||||
all_interfaces = get_local_ip4()
|
||||
interfaces = {}
|
||||
for name, interface in all_interfaces.items():
|
||||
log.debug('Interface {name}: {interface}'.format(name=name, interface=interface))
|
||||
if not INTERFACE_FILTER.search(name):
|
||||
interfaces[name] = interface
|
||||
return interfaces
|
||||
|
||||
|
||||
def start_zeroconf():
|
||||
"""
|
||||
Start the Zeroconf service
|
||||
@ -114,6 +94,6 @@ def start_zeroconf():
|
||||
return
|
||||
http_port = Settings().value('api/port')
|
||||
ws_port = Settings().value('api/websocket port')
|
||||
for name, interface in filter_interfaces().items():
|
||||
for name, interface in get_network_interfaces().items():
|
||||
worker = ZeroconfWorker(interface['ip'], http_port, ws_port)
|
||||
run_thread(worker, 'api_zeroconf_{name}'.format(name=name))
|
||||
|
@ -51,9 +51,10 @@ REPLACMENT_CHARS_MAP = str.maketrans({'\u2018': '\'', '\u2019': '\'', '\u201c':
|
||||
'\u2013': '-', '\u2014': '-', '\v': '\n\n', '\f': '\n\n'})
|
||||
NEW_LINE_REGEX = re.compile(r' ?(\r\n?|\n) ?')
|
||||
WHITESPACE_REGEX = re.compile(r'[ \t]+')
|
||||
INTERFACE_FILTER = re.compile('lo|loopback|docker|tun', re.IGNORECASE)
|
||||
|
||||
|
||||
def get_local_ip4():
|
||||
def get_network_interfaces():
|
||||
"""
|
||||
Creates a dictionary of local IPv4 interfaces on local machine.
|
||||
If no active interfaces available, returns a dict of localhost IPv4 information
|
||||
@ -61,43 +62,33 @@ def get_local_ip4():
|
||||
:returns: Dict of interfaces
|
||||
"""
|
||||
log.debug('Getting local IPv4 interface(es) information')
|
||||
my_ip4 = {}
|
||||
for iface in QNetworkInterface.allInterfaces():
|
||||
interfaces = {}
|
||||
for interface in QNetworkInterface.allInterfaces():
|
||||
interface_name = interface.name()
|
||||
if not INTERFACE_FILTER.search(interface_name):
|
||||
log.debug('Filtering out interfaces we don\'t care about: {name}'.format(name=interface_name))
|
||||
continue
|
||||
log.debug('Checking for isValid and flags == IsUP | IsRunning')
|
||||
if not iface.isValid() or not (iface.flags() & (QNetworkInterface.IsUp | QNetworkInterface.IsRunning)):
|
||||
if not interface.isValid() or not (interface.flags() & (QNetworkInterface.IsUp | QNetworkInterface.IsRunning)):
|
||||
continue
|
||||
log.debug('Checking address(es) protocol')
|
||||
for address in iface.addressEntries():
|
||||
for address in interface.addressEntries():
|
||||
ip = address.ip()
|
||||
log.debug('Checking for protocol == IPv4Protocol')
|
||||
if ip.protocol() == QAbstractSocket.IPv4Protocol:
|
||||
log.debug('Getting interface information')
|
||||
my_ip4[iface.name()] = {'ip': ip.toString(),
|
||||
'broadcast': address.broadcast().toString(),
|
||||
'netmask': address.netmask().toString(),
|
||||
'prefix': address.prefixLength(),
|
||||
'localnet': QHostAddress(address.netmask().toIPv4Address() &
|
||||
ip.toIPv4Address()).toString()
|
||||
}
|
||||
log.debug('Adding {iface} to active list'.format(iface=iface.name()))
|
||||
if len(my_ip4) == 0:
|
||||
interfaces[interface_name] = {
|
||||
'ip': ip.toString(),
|
||||
'broadcast': address.broadcast().toString(),
|
||||
'netmask': address.netmask().toString(),
|
||||
'prefix': address.prefixLength(),
|
||||
'localnet': QHostAddress(address.netmask().toIPv4Address() &
|
||||
ip.toIPv4Address()).toString()
|
||||
}
|
||||
log.debug('Adding {interface} to active list'.format(interface=interface.name()))
|
||||
if len(interfaces) == 0:
|
||||
log.warning('No active IPv4 network interfaces detected')
|
||||
return my_ip4
|
||||
if 'localhost' in my_ip4:
|
||||
log.debug('Renaming windows localhost to lo')
|
||||
my_ip4['lo'] = my_ip4['localhost']
|
||||
my_ip4.pop('localhost')
|
||||
if len(my_ip4) == 1:
|
||||
if 'lo' in my_ip4:
|
||||
# No active interfaces - so leave localhost in there
|
||||
log.warning('No active IPv4 interfaces found except localhost')
|
||||
else:
|
||||
# Since we have a valid IP4 interface, remove localhost
|
||||
if 'lo' in my_ip4:
|
||||
log.debug('Found at least one IPv4 interface, removing localhost')
|
||||
my_ip4.pop('lo')
|
||||
|
||||
return my_ip4
|
||||
return interfaces
|
||||
|
||||
|
||||
def trace_error_handler(logger):
|
||||
|
@ -28,7 +28,7 @@ from unittest import TestCase
|
||||
from PyQt5 import QtWidgets
|
||||
|
||||
from openlp.core.api.tab import ApiTab
|
||||
from openlp.core.common import get_local_ip4
|
||||
from openlp.core.common import get_network_interfaces
|
||||
from openlp.core.common.registry import Registry
|
||||
from openlp.core.common.settings import Settings
|
||||
from tests.helpers.testmixin import TestMixin
|
||||
@ -62,7 +62,7 @@ class TestApiTab(TestCase, TestMixin):
|
||||
Registry().create()
|
||||
Registry().set_flag('website_version', '00-00-0000')
|
||||
self.form = ApiTab(self.parent)
|
||||
self.my_ip4_list = get_local_ip4()
|
||||
self.interfaces = get_network_interfaces()
|
||||
|
||||
def tearDown(self):
|
||||
"""
|
||||
@ -77,9 +77,9 @@ class TestApiTab(TestCase, TestMixin):
|
||||
Test the get_ip_address function with ZERO_URL
|
||||
"""
|
||||
# GIVEN: list of local IP addresses for this machine
|
||||
ip4_list = []
|
||||
for ip4 in iter(self.my_ip4_list):
|
||||
ip4_list.append(self.my_ip4_list.get(ip4)['ip'])
|
||||
ip_addresses = []
|
||||
for _, interface in self.interfaces.items():
|
||||
ip_addresses.append(interface['ip'])
|
||||
|
||||
# WHEN: the default ip address is given
|
||||
ip_address = self.form.get_ip_address(ZERO_URL)
|
||||
@ -87,7 +87,7 @@ class TestApiTab(TestCase, TestMixin):
|
||||
# THEN: the default ip address will be returned
|
||||
assert re.match(r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}', ip_address), \
|
||||
'The return value should be a valid ip address'
|
||||
assert ip_address in ip4_list, 'The return address should be in the list of local IP addresses'
|
||||
assert ip_address in ip_addresses, 'The return address should be in the list of local IP addresses'
|
||||
|
||||
def test_get_ip_address_with_ip(self):
|
||||
"""
|
||||
|
@ -82,14 +82,24 @@ def test_zeroconf_worker_stop():
|
||||
assert worker._can_run is False
|
||||
|
||||
|
||||
# @patch('openlp.core.api.zeroconf.get_local_ip4')
|
||||
@patch('openlp.core.api.zeroconf.get_network_interfaces')
|
||||
@patch('openlp.core.api.zeroconf.Registry')
|
||||
@patch('openlp.core.api.zeroconf.Settings')
|
||||
@patch('openlp.core.api.zeroconf.ZeroconfWorker')
|
||||
@patch('openlp.core.api.zeroconf.run_thread')
|
||||
def test_start_zeroconf(mocked_run_thread, MockedZeroconfWorker, MockedSettings, MockedRegistry):
|
||||
def test_start_zeroconf(mocked_run_thread, MockedZeroconfWorker, MockedSettings, 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
|
||||
MockedSettings.return_value.value.side_effect = [8000, 8001]
|
||||
mocked_worker = MagicMock()
|
||||
@ -99,4 +109,4 @@ def test_start_zeroconf(mocked_run_thread, MockedZeroconfWorker, MockedSettings,
|
||||
start_zeroconf()
|
||||
|
||||
# THEN: A worker is added to the list of threads
|
||||
mocked_run_thread.assert_called_once_with(mocked_worker, 'api_zeroconf')
|
||||
mocked_run_thread.assert_called_once_with(mocked_worker, 'api_zeroconf_eth0')
|
||||
|
@ -28,7 +28,7 @@ from PyQt5.QtCore import QObject
|
||||
from PyQt5.QtNetwork import QHostAddress, QNetworkAddressEntry, QNetworkInterface
|
||||
|
||||
import openlp.core.common
|
||||
from openlp.core.common import get_local_ip4
|
||||
from openlp.core.common import get_network_interfaces
|
||||
from tests.helpers.testmixin import TestMixin
|
||||
|
||||
|
||||
@ -112,12 +112,12 @@ class TestInterfaces(TestCase, TestMixin):
|
||||
# WHEN: get_local_ip4 is called
|
||||
with patch('openlp.core.common.QNetworkInterface') as mock_network_interface:
|
||||
mock_network_interface.allInterfaces.return_value = []
|
||||
ifaces = get_local_ip4()
|
||||
interfaces = get_network_interfaces()
|
||||
|
||||
# THEN: There should not be any interfaces detected
|
||||
mock_log.debug.assert_has_calls(call_debug)
|
||||
mock_log.warning.assert_has_calls(call_warning)
|
||||
assert not ifaces, 'There should have been no active interfaces listed'
|
||||
assert not interfaces, 'There should have been no active interfaces listed'
|
||||
|
||||
@patch.object(openlp.core.common, 'log')
|
||||
def test_ip4_lo(self, mock_log):
|
||||
@ -136,12 +136,12 @@ class TestInterfaces(TestCase, TestMixin):
|
||||
# WHEN: get_local_ip4 is called
|
||||
with patch('openlp.core.common.QNetworkInterface') as mock_network_interface:
|
||||
mock_network_interface.allInterfaces.return_value = [self.fake_lo]
|
||||
ifaces = get_local_ip4()
|
||||
interfaces = get_network_interfaces()
|
||||
|
||||
# THEN: There should be a fake 'lo' interface
|
||||
mock_log.debug.assert_has_calls(call_debug)
|
||||
mock_log.warning.assert_has_calls(call_warning)
|
||||
assert ifaces == self.fake_lo.fake_data, "There should have been an 'lo' interface listed"
|
||||
assert interfaces == self.fake_lo.fake_data, "There should have been an 'lo' interface listed"
|
||||
|
||||
@patch.object(openlp.core.common, 'log')
|
||||
def test_ip4_localhost(self, mock_log):
|
||||
@ -149,24 +149,25 @@ class TestInterfaces(TestCase, TestMixin):
|
||||
Test get_local_ip4 returns proper dictionary with 'lo' if interface is 'localhost'
|
||||
"""
|
||||
# GIVEN: Test environment
|
||||
call_debug = [call('Getting local IPv4 interface(es) information'),
|
||||
call('Checking for isValid and flags == IsUP | IsRunning'),
|
||||
call('Checking address(es) protocol'),
|
||||
call('Checking for protocol == IPv4Protocol'),
|
||||
call('Getting interface information'),
|
||||
call('Adding localhost to active list'),
|
||||
call('Renaming windows localhost to lo')]
|
||||
call_debug = [
|
||||
call('Getting local IPv4 interface(es) information'),
|
||||
call('Checking for isValid and flags == IsUP | IsRunning'),
|
||||
call('Checking address(es) protocol'),
|
||||
call('Checking for protocol == IPv4Protocol'),
|
||||
call('Getting interface information'),
|
||||
call('Adding localhost to active list')
|
||||
]
|
||||
call_warning = [call('No active IPv4 interfaces found except localhost')]
|
||||
|
||||
# WHEN: get_local_ip4 is called
|
||||
with patch('openlp.core.common.QNetworkInterface') as mock_network_interface:
|
||||
mock_network_interface.allInterfaces.return_value = [self.fake_localhost]
|
||||
ifaces = get_local_ip4()
|
||||
interfaces = get_network_interfaces()
|
||||
|
||||
# THEN: There should be a fake 'lo' interface
|
||||
mock_log.debug.assert_has_calls(call_debug)
|
||||
mock_log.warning.assert_has_calls(call_warning)
|
||||
assert ifaces == self.fake_lo.fake_data, "There should have been an 'lo' interface listed"
|
||||
assert interfaces == self.fake_lo.fake_data, "There should have been an 'lo' interface listed"
|
||||
|
||||
@patch.object(openlp.core.common, 'log')
|
||||
def test_ip4_eth25(self, mock_log):
|
||||
@ -185,12 +186,12 @@ class TestInterfaces(TestCase, TestMixin):
|
||||
# WHEN: get_local_ip4 is called
|
||||
with patch('openlp.core.common.QNetworkInterface') as mock_network_interface:
|
||||
mock_network_interface.allInterfaces.return_value = [self.fake_address]
|
||||
ifaces = get_local_ip4()
|
||||
interfaces = get_network_interfaces()
|
||||
|
||||
# THEN: There should be a fake 'eth25' interface
|
||||
mock_log.debug.assert_has_calls(call_debug)
|
||||
mock_log.warning.assert_has_calls(call_warning)
|
||||
assert ifaces == self.fake_address.fake_data
|
||||
assert interfaces == self.fake_address.fake_data
|
||||
|
||||
@patch.object(openlp.core.common, 'log')
|
||||
def test_ip4_lo_eth25(self, mock_log):
|
||||
@ -215,9 +216,9 @@ class TestInterfaces(TestCase, TestMixin):
|
||||
# WHEN: get_local_ip4 is called
|
||||
with patch('openlp.core.common.QNetworkInterface') as mock_network_interface:
|
||||
mock_network_interface.allInterfaces.return_value = [self.fake_lo, self.fake_address]
|
||||
ifaces = get_local_ip4()
|
||||
interfaces = get_network_interfaces()
|
||||
|
||||
# THEN: There should be a fake 'eth25' interface
|
||||
mock_log.debug.assert_has_calls(call_debug)
|
||||
mock_log.warning.assert_has_calls(call_warning)
|
||||
assert ifaces == self.fake_address.fake_data, "There should have been only 'eth25' interface listed"
|
||||
assert interfaces == self.fake_address.fake_data, "There should have been only 'eth25' interface listed"
|
||||
|
Loading…
Reference in New Issue
Block a user