diff --git a/.gitignore b/.gitignore index 85bc879..4416c6b 100644 --- a/.gitignore +++ b/.gitignore @@ -15,3 +15,5 @@ dist/ .vscode .bak .pyo + +.idea diff --git a/debinterface/__init__.py b/debinterface/__init__.py index c894d8d..8a163d4 100644 --- a/debinterface/__init__.py +++ b/debinterface/__init__.py @@ -2,14 +2,14 @@ """Imports for easier use""" from .adapter import NetworkAdapter from .adapterValidation import NetworkAdapterValidation -from .dnsmasqRange import (DnsmasqRange, - DEFAULT_CONFIG as DNSMASQ_DEFAULT_CONFIG) +from .dnsmasqRange import (DEFAULT_CONFIG as DNSMASQ_DEFAULT_CONFIG, DnsmasqRange) from .hostapd import Hostapd from .interfaces import Interfaces from .interfacesReader import InterfacesReader +from .interfacesValidation import InterfacesValidation from .interfacesWriter import InterfacesWriter -__version__ = '3.5.0' +__version__ = '3.6.0' __all__ = [ 'NetworkAdapter', @@ -18,6 +18,7 @@ 'DNSMASQ_DEFAULT_CONFIG', 'Hostapd', 'Interfaces', + 'InterfacesValidation', 'InterfacesReader', 'InterfacesWriter' ] diff --git a/debinterface/adapter.py b/debinterface/adapter.py index 8df61cd..eb4a8a8 100644 --- a/debinterface/adapter.py +++ b/debinterface/adapter.py @@ -5,13 +5,14 @@ It has setter for many common options, but it is impossible to have setter for every options on earth ! """ -from __future__ import print_function, with_statement, absolute_import +from __future__ import absolute_import, print_function, with_statement import copy import socket import warnings from .adapterValidation import NetworkAdapterValidation, VALID_OPTS + try: import typing as tp except ImportError: @@ -483,10 +484,108 @@ def appendPostDown(self, cmd): cmd = cmd.split() self._ensure_list(self._ifAttributes, "post-down", cmd) + def setBondMaster(self, master_adapter_name): + # type: (str)->None + """Set bond-master reference on slave. + + Args: + master_adapter_name (str): name of master iface + """ + self._ifAttributes['bond-master'] = master_adapter_name + + def setBondMode(self, mode): + # type: (tp.Union[str, int])->None + """Set bond-mode. + + Args: + mode (str, int): mode of bonding + """ + if isinstance(mode, str): + mode = mode.lower().replace('_', '-') + self._ifAttributes['bond-mode'] = mode + + def setBondMiimon(self, miimon): + # type: (int)->None + """Set bond-miimon parameter. + Specifies the MII link monitoring frequency in milliseconds. + + Args: + miimon (int): miimon + """ + self._ifAttributes['bond-miimon'] = miimon + + def setBondUpDelay(self, delay): + # type: (int)->None + """Set bond-updelay parameter. + Specifies the time, in milliseconds, to wait before enabling a slave after a link recovery has been detected. + This option is only valid for the miimon link monitor. The updelay value should be a multiple of the miimon value. + + Args: + delay (int): delay in milliseconds + """ + miimon = self._ifAttributes.get('bond-miimon', 0) + if miimon and delay % miimon != 0: + raise ValueError("Updelay should be multiple of miimon value") + self._ifAttributes['bond-updelay'] = delay + + def setBondDownDelay(self, delay): + # type: (int)->None + """Set bond-downdelay parameter. + Specifies the time, in milliseconds, to wait before disabling a slave after a link failure has been detected. + This option is only valid for the miimon link monitor. The updelay value should be a multiple of the miimon value. + + Args: + delay (int): delay in milliseconds + """ + miimon = self._ifAttributes.get('bond-miimon', 0) + if miimon and delay % miimon != 0: + raise ValueError("Downdelay should be multiple of miimon value") + self._ifAttributes['bond-downdelay'] = delay + + def setBondPrimary(self, primary_name): + # type: (str)->None + """Set bond-primary parameter. + A string specifying which slave is the primary device. + The specified device will always be the active slave while it is available. + Only when the primary is off-line will alternate devices be used. + This is useful when one slave is preferred over another, e.g., when one slave has higher throughput than another. + The primary option is only valid for active-backup (1) mode. + + Args: + primary_name (str): name of primary slave + """ + self._ifAttributes['bond-primary'] = primary_name + + def setBondSlaves(self, slaves): + # type: (tp.Union[str, tp.List[str]])->None + """Set bond-primary parameter. + Slave interfaces names + + Args: + slaves (list, optional): names of slaves + """ + if not isinstance(slaves, list): + slaves = [slaves] + self._ifAttributes['bond-slaves'] = slaves + def setUnknown(self, key, val): # type: (str, tp.Any)->None """Stores uncommon options as there are with no special handling It's impossible to know about all available options + Format key with lower case. Replaces '_' with '-' + + Args: + key (str): the option name + val (any): the option value + """ + key = key.lower().replace('_', '-') + self.setUnknownUnformatted(key, val) + + def setUnknownUnformatted(self, key, val): + # type: (str, tp.Any)->None + """Stores uncommon options as there are with no special handling + It's impossible to know about all available options + Stores key as is WARNING: duplicated directives are overwriten. TODO better Args: @@ -607,6 +706,13 @@ def set_options(self, options): 'dns-search': self.setDnsSearch, 'nameservers': self.setNameservers, 'wpa-conf': self.setWpaConf, + 'bond-master': self.setBondMaster, + 'bond-slaves': self.setBondSlaves, + 'bond-miimon': self.setBondMiimon, + 'bond-mode': self.setBondMode, + 'bond-updelay': self.setBondUpDelay, + 'bond-downdelay': self.setBondDownDelay, + 'bond-primary': self.setBondPrimary } # type: tp.Dict[str, tp.Callable[[tp.Any], None]] for key, value in options.items(): if key in roseta: diff --git a/debinterface/adapterValidation.py b/debinterface/adapterValidation.py index 01d13d8..6787b20 100644 --- a/debinterface/adapterValidation.py +++ b/debinterface/adapterValidation.py @@ -3,14 +3,15 @@ but it is by no means fool proof as it is impossible to have checks for everything as any package can add its keys. """ -from __future__ import print_function, with_statement, absolute_import +from __future__ import absolute_import, print_function, with_statement + import socket + try: import typing as tp except ImportError: pass - VALID_OPTS = { 'hotplug': {'type': bool}, # Beware, option is really called allow-hotplug 'auto': {'type': bool}, @@ -54,7 +55,13 @@ 'post-up': {'type': list}, 'down': {'type': list}, 'pre-down': {'type': list}, - 'post-down': {'type': list} + 'post-down': {'type': list}, + 'bond-mode': { + 'in': ['balance-rr', 'active-backup', 'balance-xor', 'broadcast', '802.3ad', 'balance-tlb', 'balance-alb', 0, 1, + 2, 3, 4, 5, 6]}, + 'bond-miimon': {'type': int}, + 'bond-updelay': {'type': int}, + 'bond-downdelay': {'type': int} } # type: tp.Dict[str, tp.Dict[str, tp.Any]] REQUIRED_FAMILY_OPTS = { "inet": { @@ -81,7 +88,7 @@ "static": (), "manual": (), "dhcp": (), - "v4tunnel": ("address", ), + "v4tunnel": ("address",), "6to4": () }, "can": { diff --git a/debinterface/dnsmasqRange.py b/debinterface/dnsmasqRange.py index 5410afa..3712931 100644 --- a/debinterface/dnsmasqRange.py +++ b/debinterface/dnsmasqRange.py @@ -1,17 +1,18 @@ # -*- coding: utf-8 -*- -from __future__ import print_function, with_statement, absolute_import +from __future__ import absolute_import, print_function, with_statement + import copy import os import shutil from socket import inet_aton from . import toolutils + try: import typing as tp except ImportError: pass - DEFAULT_CONFIG = { 'dhcp-range': [ { diff --git a/debinterface/hostapd.py b/debinterface/hostapd.py index 413b22f..2666064 100644 --- a/debinterface/hostapd.py +++ b/debinterface/hostapd.py @@ -1,9 +1,11 @@ # -*- coding: utf-8 -*- -from __future__ import print_function, with_statement, absolute_import +from __future__ import absolute_import, print_function, with_statement + import os import shutil from . import toolutils + try: import typing as tp except ImportError: diff --git a/debinterface/interfaces.py b/debinterface/interfaces.py index c5c075f..a8d60e8 100644 --- a/debinterface/interfaces.py +++ b/debinterface/interfaces.py @@ -1,10 +1,12 @@ # -*- coding: utf-8 -*- # A class representing the contents of /etc/network/interfaces -from __future__ import print_function, with_statement, absolute_import -from .interfacesWriter import InterfacesWriter -from .interfacesReader import InterfacesReader -from .adapter import NetworkAdapter +from __future__ import absolute_import, print_function, with_statement + from . import toolutils +from .adapter import NetworkAdapter +from .interfacesReader import InterfacesReader +from .interfacesWriter import InterfacesWriter + try: import typing as tp except ImportError: @@ -198,3 +200,32 @@ def _set_paths(self, interfaces_path, backup_path): else: # self._interfaces_path is never None self._backup_path = self._interfaces_path + ".bak" + + def addBond(self, name="bond0", mode=0, slaves=None): + # type: (str, tp.Union[str, int], tp.Union[None, str, tp.List[str]])->NetworkAdapter + """Add new bond interface + + Args: + name (str): name of new interfaces + mode (str/int): mode of bonding + slaves (list, optional): list of names of bond slaves + + Returns: + NetworkAdapter: the new adapter + """ + if not slaves: + slaves = [] + + if self.getAdapter(name) is not None: + raise ValueError("Interface {} already exists".format(name)) + + for slave in slaves: + adapter = self.getAdapter(slave) + if adapter is None: + raise ValueError("Interface {} does not exists".format(slave)) + if 'bond-master' in adapter.attributes and adapter.attributes['bond-master'] != name: + raise ValueError( + "Interface {} already has master iface {}".format(slave, adapter.attributes['bond-master'])) + adapter.attributes['bond-master'] = name + + return self.addAdapter({"name": name, 'bond-mode': mode, 'bond-slaves': ' '.join(slaves)}) diff --git a/debinterface/interfacesReader.py b/debinterface/interfacesReader.py index c7c30e5..70367cf 100644 --- a/debinterface/interfacesReader.py +++ b/debinterface/interfacesReader.py @@ -1,8 +1,11 @@ # -*- coding: utf-8 -*- # A class representing the contents of /etc/network/interfaces -from __future__ import print_function, with_statement, absolute_import +from __future__ import absolute_import, print_function, with_statement + import os + from .adapter import NetworkAdapter + try: import typing as tp except ImportError: @@ -118,6 +121,8 @@ def _parse_details(self, line): if line[0].isspace() is True: keyword, value = line.strip().split(None, 1) + keyword = keyword.lower().replace('_', '-') + if keyword == 'address': self._adapters[self._context].setAddress(value) elif keyword == 'netmask': @@ -132,6 +137,25 @@ def _parse_details(self, line): self._adapters[self._context].setHostapd(value) elif keyword == 'wpa-conf': self._adapters[self._context].setWpaConf(value) + elif keyword == 'bond-mode': + mode = value # type: tp.Union[str, int] + try: + mode = int(mode) + except ValueError: + pass + self._adapters[self._context].setBondMode(mode) + elif keyword == 'bond-miimon': + self._adapters[self._context].setBondMiimon(int(value)) + elif keyword == 'bond-updelay': + self._adapters[self._context].setBondUpDelay(int(value)) + elif keyword == 'bond-downdelay': + self._adapters[self._context].setBondDownDelay(int(value)) + elif keyword == 'bond-primary': + self._adapters[self._context].setBondPrimary(value) + elif keyword == 'bond-master': + self._adapters[self._context].setBondMaster(value) + elif keyword == 'bond-slaves': + self._adapters[self._context].setBondSlaves(value.split()) elif keyword == 'dns-nameservers': self._adapters[self._context].setDnsNameservers(value) elif keyword == 'dns-search': diff --git a/debinterface/interfacesValidation.py b/debinterface/interfacesValidation.py new file mode 100644 index 0000000..cec6e58 --- /dev/null +++ b/debinterface/interfacesValidation.py @@ -0,0 +1,66 @@ +from .interfaces import Interfaces + + +class InterfacesValidation: + def __init__(self, interfaces): + # type: (Interfaces)->None + self.interfaces = interfaces + + def validate_bonding(self): + # type: ()->None + """Validate bond network settings with debian bonding standard (https://wiki.debian.org/Bonding) + + Raises: + ValueError: human-readable description of error + """ + for adapter in self.interfaces.adapters: + attrs = adapter.attributes + + if 'bond-mode' in attrs: + mode = attrs['bond-mode'] + if isinstance(mode, str): + mode = mode.lower().replace('_', '-') + + if not 'bond-slaves' in attrs: + raise ValueError("Interface {} have no slaves".format(attrs['name'])) + + if mode == 1 or mode == 'active-backup': + if not 'bond-primary' in attrs: + raise ValueError("Interface {} have no primary".format(attrs['name'])) + primary = self.interfaces.getAdapter(attrs['bond-primary']) + if not primary: + raise ValueError( + "Interface {0} have {1} as primary, but {1} was not found".format(attrs['name'], + attrs[ + 'bond-primary'])) + if primary.attributes['name'] not in attrs['bond-slaves']: + raise ValueError( + "Primary interface {} is not bond slave of {}".format(attrs['bond-primary'], + attrs['name'])) + + if 'bond-slaves' in attrs: + slaves = attrs['bond-slaves'] + if slaves != ['none']: + for slave in slaves: + slave_adapter = self.interfaces.getAdapter(slave) + if not slave_adapter or slave_adapter.attributes.get('bond-master', None) != attrs[ + 'name']: + raise ValueError("Interface {} have no {} as master".format(slave, attrs['name'])) + + if 'bond-master' in attrs: + master = self.interfaces.getAdapter(attrs['bond-master']) + if not master: + raise ValueError( + "Interface {} have no {} as master".format(attrs['name'], attrs['bond-master'])) + master_mode = master.attributes['bond-mode'] + if master_mode == 1 or master_mode == 'active-backup': + if not 'bond-primary' in attrs: + raise ValueError( + "Interface {} have no primary when bond master mode is active-backup".format( + attrs['name'])) + primary = self.interfaces.getAdapter(attrs['bond-primary']) + if not primary: + raise ValueError( + "Interface {0} have {1} as primary, but {1} was not found".format(attrs['name'], + attrs[ + 'bond-primary'])) diff --git a/debinterface/interfacesWriter.py b/debinterface/interfacesWriter.py index e98cdfb..99542ec 100644 --- a/debinterface/interfacesWriter.py +++ b/debinterface/interfacesWriter.py @@ -1,9 +1,10 @@ # -*- coding: utf-8 -*- # Write interface -from __future__ import print_function, with_statement, absolute_import +from __future__ import absolute_import, print_function, with_statement + +import os import shutil from collections import defaultdict -import os from string import Template from . import toolutils @@ -31,6 +32,8 @@ class InterfacesWriter(object): ] _prepFields = ['pre-up', 'pre-down', 'up', 'down', 'post-up', 'post-down'] _bridgeFields = ['ports', 'fd', 'hello', 'maxage', 'stp', 'maxwait'] + _bondFields = ['bond-master', 'bond-slaves', 'bond-miimon', 'bond-updelay', 'bond-downdelay', + 'bond-primary', 'bond-mode'] _plugins = ['hostapd', 'wpa-conf'] def __init__(self, adapters, interfaces_path, backup_path=None, @@ -163,6 +166,7 @@ def _write_adapter(self, interfaces, adapter): self._write_bridge(interfaces, adapter, ifAttributes) self._write_plugins(interfaces, adapter, ifAttributes) self._write_callbacks(interfaces, adapter, ifAttributes) + self._write_bond(interfaces, adapter, ifAttributes) self._write_unknown(interfaces, adapter, ifAttributes) interfaces.write("\n") @@ -176,6 +180,22 @@ def _write_auto(self, interfaces, adapter, ifAttributes): except KeyError: pass + def _write_bond(self, interfaces, adapter, ifAttributes): + # type: (tp.IO[str], NetworkAdapter, tp.Dict[str, tp.Any])->None + for field in self._bondFields: + try: + value = ifAttributes[field] + if value is not None: + if isinstance(value, list): + d = dict(varient=field, + value=" ".join(ifAttributes[field])) + else: + d = dict(varient=field, value=ifAttributes[field]) + interfaces.write(self._cmd.substitute(d)) + # Keep going if a field isn't provided. + except KeyError: + pass + def _write_hotplug(self, interfaces, adapter, ifAttributes): # type: (tp.IO[str], NetworkAdapter, tp.Dict[str, tp.Any])->None """ Write if applicable """ diff --git a/debinterface/toolutils.py b/debinterface/toolutils.py index c5ebc65..9a55734 100644 --- a/debinterface/toolutils.py +++ b/debinterface/toolutils.py @@ -1,10 +1,12 @@ # -*- coding: utf-8 -*- -from __future__ import print_function, with_statement, absolute_import +from __future__ import absolute_import, print_function, with_statement + import os import stat +import subprocess import tempfile from contextlib import contextmanager -import subprocess + try: import typing as tp except ImportError: diff --git a/test/interfaces_bond1.txt b/test/interfaces_bond1.txt new file mode 100644 index 0000000..28ae29e --- /dev/null +++ b/test/interfaces_bond1.txt @@ -0,0 +1,33 @@ +# Example of bond interfaces + +auto bond0 +iface bond0 inet static + address 0.0.0.0 + netmask 0.0.0.0 + bond-slaves enp1s0 enp2s0 + bond-mode 1 + bond-primary enp1s0 + up /sbin/ifenslave bond0 enp1s0 enp2s0 + down /sbin/ifenslave -d bond0 enp1s0 enp2s0 + +auto enp1s0 +iface enp1s0 inet manual + bond-master bond0 + bond-primary enp1s0 + +auto enp2s0 +iface enp2s0 inet manual + bond-master bond0 + bond-primary enp1s0 + +auto bond0.10 +iface bond0.10 inet static + address 192.168.10.1 + netmask 255.255.255.0 + vlan_raw_device bond0 + +auto bond0.20 +iface bond0.20 inet static + address 192.168.20.1 + netmask 255.255.255.0 + vlan_raw_device bond0 diff --git a/test/test_adapter.py b/test/test_adapter.py index a940026..683ffab 100644 --- a/test/test_adapter.py +++ b/test/test_adapter.py @@ -1,5 +1,6 @@ # -*- coding: utf-8 -*- import unittest + from ..debinterface import NetworkAdapter @@ -51,7 +52,7 @@ def test_assign_all_items(self): 'auto': True, 'hotplug': True, 'bridge-opts': {'maxwait': 1}, - 'up' : cmds, + 'up': cmds, 'down': cmds, 'pre-up': cmds, 'pre-down': cmds, diff --git a/test/test_dnsmasqRange.py b/test/test_dnsmasqRange.py index 98a2caa..ee527c6 100644 --- a/test/test_dnsmasqRange.py +++ b/test/test_dnsmasqRange.py @@ -1,11 +1,11 @@ # -*- coding: utf-8 -*- -import unittest -import socket import copy import filecmp +import socket import tempfile -from ..debinterface import DnsmasqRange, DNSMASQ_DEFAULT_CONFIG +import unittest +from ..debinterface import DNSMASQ_DEFAULT_CONFIG, DnsmasqRange DEFAULT_CONTENT = ''' dhcp-range=interface:wlan0,10.1.10.11,10.1.10.250,24h @@ -35,7 +35,8 @@ def test_write(self): dns._config = copy.deepcopy(DNSMASQ_DEFAULT_CONFIG) dns.write() source.flush() - content = open(source.name).read().replace("\n", "") + with open(source.name) as f: + content = f.read().replace("\n", "") for line in DEFAULT_CONTENT.split("\n"): if not line or line == "\n": continue diff --git a/test/test_hostapd.py b/test/test_hostapd.py index 7b41fd7..b42054d 100644 --- a/test/test_hostapd.py +++ b/test/test_hostapd.py @@ -1,9 +1,9 @@ # -*- coding: utf-8 -*- -import unittest import filecmp import tempfile -from ..debinterface import Hostapd +import unittest +from ..debinterface import Hostapd DEFAULT_CONTENT = ''' interface=wlan0 @@ -87,7 +87,8 @@ def test_write(self): dns._config = DEFAULT_CONFIG dns.write() source.flush() - content = open(source.name).read().replace("\n", "") + with open(source.name) as f: + content = f.read().replace("\n", "") for line in DEFAULT_CONTENT.split("\n"): if not line or line == "\n": continue diff --git a/test/test_interfaces.py b/test/test_interfaces.py index 7b0243e..9236ba5 100644 --- a/test/test_interfaces.py +++ b/test/test_interfaces.py @@ -1,11 +1,12 @@ # -*- coding: utf-8 -*- import os import unittest -from ..debinterface import Interfaces +from ..debinterface import Interfaces, InterfacesValidation INF_PATH = os.path.join(os.path.abspath(os.path.dirname(__file__)), "interfaces.txt") INF2_PATH = os.path.join(os.path.abspath(os.path.dirname(__file__)), "interfaces2.txt") +INFBOND1_PATH = os.path.join(os.path.abspath(os.path.dirname(__file__)), "interfaces_bond1.txt") class TestInterfaces(unittest.TestCase): @@ -79,7 +80,6 @@ def test_export_adapter_settings(self): with self.assertRaises(KeyError): adapter.get_attr("attr") # type: ignore - def test_remove_adapter(self): # type: ()->None itfs = Interfaces(interfaces_path=INF_PATH) @@ -98,3 +98,41 @@ def test_remove_adapter_name(self): self.assertEqual(len(itfs.adapters), nb_adapters - 1) for adapter in itfs.adapters: self.assertNotEqual("eth0", adapter.attributes["name"]) + + def test_bond_remove_bond(self): + # type: ()->None + itfs = Interfaces(interfaces_path=INFBOND1_PATH) + validator = InterfacesValidation(itfs) + validator.validate_bonding() + itfs.removeAdapterByName("bond0") + with self.assertRaises(ValueError): + validator.validate_bonding() + + def test_bond_configure_bond(self): + # type: ()->None + itfs = Interfaces(interfaces_path=INFBOND1_PATH) + # Slave has another interface as master + with self.assertRaises(ValueError): + itfs.addBond(name="bond1", slaves=["enp1s0"]) + # Slave does not exist + with self.assertRaises(ValueError): + itfs.addBond(name="bond1", slaves=["enp3s0"]) + + itfs.addAdapter({"name": "enp3s0"}) + itfs.addBond(name="bond1", slaves=["enp3s0"], mode=0) + validator = InterfacesValidation(itfs) + validator.validate_bonding() + + def test_bond_miimon(self): + # type: ()->None + itfs = Interfaces(interfaces_path=INFBOND1_PATH) + bond0 = itfs.getAdapter("bond0") + self.assertIsNotNone(bond0) + bond0.setBondMode("bAlAnCe_XOR") # type: ignore + validator = InterfacesValidation(itfs) + validator.validate_bonding() + bond0.setBondMiimon(100) # type: ignore + bond0.setBondUpDelay(200) # type: ignore + # Delay should be multiple of the miimon value + with self.assertRaises(ValueError): + bond0.setBondDownDelay(205) # type: ignore diff --git a/test/test_interfacesReader.py b/test/test_interfacesReader.py index 1c931b6..2e57f54 100644 --- a/test/test_interfacesReader.py +++ b/test/test_interfacesReader.py @@ -1,8 +1,8 @@ # -*- coding: utf-8 -*- import os import unittest -from ..debinterface import InterfacesReader +from ..debinterface import InterfacesReader INF_PATH = os.path.join(os.path.abspath(os.path.dirname(__file__)), "interfaces.txt") INF2_PATH = os.path.join(os.path.abspath(os.path.dirname(__file__)), "interfaces2.txt") @@ -198,7 +198,7 @@ def test_source_other_file(self): self.assertEqual(adapters[0].attributes["name"], "eth_main_file_1") # before the source self.assertEqual(adapters[1].attributes["name"], "eth0") # in the sourced file self.assertEqual(adapters[2].attributes["name"], "eth2") # in the sourced file - self.assertEqual(adapters[3].attributes["name"], "eth_main_file_2") # after the source + self.assertEqual(adapters[3].attributes["name"], "eth_main_file_2") # after the source # Check paths self.assertEqual(adapters[0].interfaces_path, INF5_PATH) # before the source @@ -214,4 +214,5 @@ def test_attrs_with_spaces(self): None ) self.assertIsNotNone(eth1) - self.assertEqual(eth1.attributes['unknown']['some-attr'], 'value contains space\ttab_underscore-dash') # type: ignore + self.assertEqual(eth1.attributes['unknown']['some-attr'], + 'value contains space\ttab_underscore-dash') # type: ignore diff --git a/test/test_interfacesWriter.py b/test/test_interfacesWriter.py index ad3e3a4..391405d 100644 --- a/test/test_interfacesWriter.py +++ b/test/test_interfacesWriter.py @@ -1,16 +1,18 @@ # -*- coding: utf-8 -*- -from __future__ import print_function, with_statement, absolute_import +from __future__ import absolute_import, print_function, with_statement + import os -import unittest import tempfile -from ..debinterface import InterfacesWriter, NetworkAdapter +import typing as tp +import unittest +from ..debinterface import InterfacesWriter, NetworkAdapter INF_PATH = os.path.join(os.path.abspath(os.path.dirname(__file__)), "interfaces.txt") class TestInterfacesWriter(unittest.TestCase): - def compare_iterface_with_expected(self, content_text, expected, number_of_stict_lines=2): + def compare_interface_with_expected(self, content_text, expected, number_of_stict_lines): # type: (str, tp.List[str], int)->None content = list(map(lambda x: x.strip(), content_text.strip().split("\n"))) @@ -20,9 +22,15 @@ def compare_iterface_with_expected(self, content_text, expected, number_of_stict self.assertEqual(content[i], expected[i]) content = list(filter(lambda x: x, content)) - for line_written, line_expected in zip(sorted(content[number_of_stict_lines:]), sorted(expected[number_of_stict_lines:])): + for line_written, line_expected in zip(sorted(content[number_of_stict_lines:]), + sorted(expected[number_of_stict_lines:])): self.assertEqual(line_written, line_expected) + def compare_interfaces_file_with_expected(self, filename, expected, number_of_stict_lines=2): + # type: (str, tp.List[str], int)->None + with open(filename) as f: + self.compare_interface_with_expected(f.read(), expected, number_of_stict_lines) + def test_write_complete(self): # type: ()->None """Should work""" @@ -96,7 +104,7 @@ def test_write_complete(self): writer = InterfacesWriter([adapter], interfaces_path=tempf.name, backup_path="/tmp") writer.write_interfaces() - self.compare_iterface_with_expected(open(tempf.name).read(), expected) + self.compare_interfaces_file_with_expected(tempf.name, expected) def test_write_complete2(self): # type: ()->None @@ -132,7 +140,7 @@ def test_write_complete2(self): writer = InterfacesWriter([adapter], interfaces_path=tempf.name, backup_path="/tmp") writer.write_interfaces() - self.compare_iterface_with_expected(open(tempf.name).read(), expected) + self.compare_interfaces_file_with_expected(tempf.name, expected) def test_supplicant_conf_write(self): # type: ()->None @@ -156,7 +164,7 @@ def test_supplicant_conf_write(self): writer = InterfacesWriter([adapter], interfaces_path=tempf.name, backup_path="/tmp") writer.write_interfaces() - self.compare_iterface_with_expected(open(tempf.name).read(), expected) + self.compare_interfaces_file_with_expected(tempf.name, expected) def test_multiDns_write(self): # type: ()->None @@ -190,7 +198,7 @@ def test_multiDns_write(self): writer = InterfacesWriter([adapter], interfaces_path=tempf.name, backup_path="/tmp") writer.write_interfaces() - self.compare_iterface_with_expected(open(tempf.name).read(), expected) + self.compare_interfaces_file_with_expected(tempf.name, expected) def test_header_comment_no_symbol_write(self): # type: ()->None @@ -221,7 +229,7 @@ def test_header_comment_no_symbol_write(self): header_comment=header_comment, backup_path="/tmp") writer.write_interfaces() - self.compare_iterface_with_expected(open(tempf.name).read(), expected, number_of_stict_lines=len(expected)) + self.compare_interfaces_file_with_expected(tempf.name, expected, number_of_stict_lines=len(expected)) def test_header_comment_symbol_write(self): # type: ()->None @@ -252,4 +260,4 @@ def test_header_comment_symbol_write(self): header_comment=header_comment, backup_path="/tmp") writer.write_interfaces() - self.compare_iterface_with_expected(open(tempf.name).read(), expected, number_of_stict_lines=len(expected)) + self.compare_interfaces_file_with_expected(tempf.name, expected, number_of_stict_lines=len(expected))