Merge pull request #26 from migibert/discover_mtu

Discover mtu

Author: migibert

pynetlib/device.py

@@ -1,5 +1,5 @@
 from __future__ import absolute_import
-from .utils import execute_command, find_value
+from .utils import execute_command, get_devices_info
 from .exceptions import ObjectAlreadyExistsException, ObjectNotFoundException
 
 
@@ -12,6 +12,7 @@ def __init__(self, id, name, flags=[], namespace=None):
         self.state = None
         self.inet = []
         self.inet6 = []
+        self.mtu = None
 
     def is_loopback(self):
         return 'LOOPBACK' in self.flags
@@ -54,53 +55,30 @@ def disable(self):
             self.refresh()
 
     def refresh(self):
-        output = execute_command('ip addr show %s' % self.name, namespace=self.namespace)
-        devices = Device.parse_output(output, namespace=self.namespace)
-        if len(devices) != 1:
-            raise Exception('Only one device can be named %s inside the same namespace' % self.name)
-        dev = devices[0]
-        self.id = dev.id
-        self.name = dev.name
-        self.flags = dev.flags
-        self.namespace = dev.namespace
-        self.state = dev.state
-        self.inet = dev.inet
-        self.inet6 = dev.inet6
+        devices = Device.discover(namespace=self.namespace)
+        if self not in devices:
+            raise ObjectNotFoundException(self)
+        found = devices[devices.index(self)]
+        self.flags = found.flags
+        self.inet = found.inet
+        self.inet6 = found.inet6
+        self.mtu = found.mtu
 
     @staticmethod
     def discover(namespace=None):
-        output = execute_command('ip addr list', namespace=namespace)
-        return Device.parse_output(output, namespace=namespace)
-
-    @staticmethod
-    def parse_output(output, namespace=None):
         devices = []
-        current_device = None
-        for block in output.split('\n'):
-            if block and block[0].isdigit():
-                if current_device:
-                    devices.append(current_device)
-                prefixes = block.split(':')
-                id = prefixes[0]
-                name = prefixes[1].strip()
-                flags = block[block.index('<') + 1:block.index('>')].split(',')
-                current_device = Device(id, name, flags=flags, namespace=namespace)
-            words = block.strip().split(' ')
-            state = find_value(words, 'state')
-            if state is not None:
-                current_device.state = state
-            inet = find_value(words, 'inet')
-            if inet is not None:
-                current_device.inet.append(inet)
-            inet6 = find_value(words, 'inet6')
-            if inet6 is not None:
-                current_device.inet6.append(inet6)
-        if current_device:
-            devices.append(current_device)
+        output = execute_command('ip addr list', namespace=namespace)
+        for id, name, flags, state, inet, inet6, mtu in get_devices_info(output):
+            device = Device(id, name, flags=flags, namespace=namespace)
+            device.state = state
+            device.inet = inet
+            device.inet6 = inet6
+            device.mtu = mtu
+            devices.append(device)
         return devices
 
     def __eq__(self, other):
         return self.name == other.name and self.id == other.id
 
     def __repr__(self):
-        return '<' + ','.join([self.id, self.name, str(self.inet), str(self.inet6)]) + '>'
+        return '<' + ','.join([self.id, self.name, str(self.inet), str(self.inet6), str(self.mtu)]) + '>'

pynetlib/tests/fixtures/ip_addr_show

@@ -1,6 +1,4 @@
 from __future__ import absolute_import
-from nose_parameterized.parameterized import parameterized
-import os
 import mock
 import unittest
 from . import read_file
@@ -14,7 +12,6 @@ class TestDevice(unittest.TestCase):
 
     def setUp(self):
         self.ip_addr_list_output = read_file('ip_addr_list')
-        self.ip_addr_show_output = read_file('ip_addr_show')
 
     def test_init_device(self):
         id = '1'
@@ -26,6 +23,7 @@ def test_init_device(self):
         self.assertEqual(dev.inet, [])
         self.assertEqual(dev.inet6, [])
         self.assertIsNone(dev.state)
+        self.assertIsNone(dev.mtu)
 
     def test_init_loopback(self):
         id = '1'
@@ -207,17 +205,25 @@ def test_disable_down_device(self, execute_command):
 
     @mock.patch('pynetlib.device.execute_command')
     def test_refresh_device(self, execute_command):
-        execute_command.return_value = self.ip_addr_show_output
-        device = Device('1', 'eth0', flags=[])
+        execute_command.return_value = self.ip_addr_list_output
+        device = Device('2', 'eth0', flags=[])
         device.refresh()
-        execute_command.assert_called_once_with("ip addr show eth0", namespace=None)
+        execute_command.assert_called_once_with("ip addr list", namespace=None)
         self.assertEqual(device.id, '2')
         self.assertEqual(device.name, "eth0")
         self.assertEqual(device.flags, ['BROADCAST', 'MULTICAST', 'UP', 'LOWER_UP'])
-        self.assertEqual(device.inet, ['10.0.2.15/24'])
+        self.assertEqual(device.inet, ['10.0.2.15/24', '10.0.2.16/24'])
         self.assertEqual(device.inet6, ['fe80::a00:27ff:feea:67cf/64'])
+        self.assertEqual(device.mtu, '1500')
         self.assertIsNone(device.namespace)
 
+    @mock.patch('pynetlib.device.execute_command')
+    def test_refresh_non_existing_device(self, execute_command):
+        execute_command.return_value = self.ip_addr_list_output
+        device = Device('id', 'device', flags=[])
+        with self.assertRaises(ObjectNotFoundException):
+            device.refresh()
+            execute_command.assert_called_once_with("ip addr list", namespace=None)
 
 if __name__ == '__main__':
     unittest.main()

pynetlib/tests/test_device.py

@@ -1,9 +1,9 @@
 from __future__ import absolute_import
 import mock
 import unittest
-from pynetlib.utils import execute_command, find_value
+from pynetlib.utils import execute_command, find_values_or_default_value
 from pynetlib.namespace import Namespace
-from pynetlib.exceptions import ValueNotFoundException
+from nose_parameterized import parameterized
 
 
 class TestUtils(unittest.TestCase):
@@ -31,25 +31,21 @@ def test_execute_command_in_string_namespace(self, check_output):
         execute_command('command arg1 arg2', namespace=namespace)
         check_output.assert_called_once_with('ip netns exec namespace command arg1 arg2', shell=True)
 
-    def test_find_value(self):
-        values = ['key', 'value']
-        value = find_value(values, 'key')
+    def test_find_values(self):
+        values = 'key value'
+        value = find_values_or_default_value(values, 'key', single=True)
         self.assertEqual(value, 'value')
 
-    def test_find_non_existing_key(self):
-        values = ['key', 'value']
-        value = find_value(values, 'test')
-        self.assertIsNone(value)
-
-    def test_find_non_existing_value(self):
-        values = ['key']
-        with self.assertRaises(ValueNotFoundException):
-            find_value(values, 'key')
-
-    def test_find_two_values(self):
-        values = ['key', 'value', 'key', 'other']
-        value = find_value(values, 'key')
-        self.assertEqual(value, 'value')
+    @parameterized.expand([
+        ('key value', 'key', True, None, 'value'),
+        ('', 'test', True, 'value', 'value'),
+        ('key', 'key', True, None, None),
+        ('key', 'key', True, 'test', 'test'),
+        ('key value1 key value2', 'key', False, None, ['value1', 'value2'])
+    ])
+    def test_find_values_or_default_value(self, values, key, single, default, expected):
+        value = find_values_or_default_value(values, key, single=single, default_value=default)
+        self.assertEqual(value, expected)
 
 
 if __name__ == '__main__':

pynetlib/tests/test_utils.py

@@ -14,9 +14,60 @@ def execute_command(command, namespace=None):
     return subprocess.check_output(cmd, shell=True)
 
 
-def find_value(values, key):
-    if key in values:
-        if len(values) > values.index(key) + 1:
-            return values[values.index(key) + 1]
-        raise ValueNotFoundException('Value not found for key %s' % key)
-    return None
+def get_devices_info(output):
+    devices = []
+    blocks = parse_output(output)
+    for block in blocks:
+        id, name, flags = parse_header(block)
+        state = find_values_or_default_value(block, 'state', single=True, default_value='UNKNOWN')
+        inet = find_values_or_default_value(block, 'inet', default_value=[])
+        inet6 = find_values_or_default_value(block, 'inet6', default_value=[])
+        mtu = find_values_or_default_value(block, 'mtu', default_value=None, single=True)
+        devices.append((id, name, flags, state, inet, inet6, mtu))
+    return devices
+
+
+def parse_output(output):
+    blocks = []
+    current_block = ''
+
+    for line in output.split('\n'):
+        if line and line[0].isdigit():
+            if current_block:
+                blocks.append(current_block)
+                current_block = ''
+        current_block = current_block + line
+    if current_block:
+        blocks.append(current_block)
+    return blocks
+
+
+def parse_header(header):
+    prefixes = header.split(':')
+    id = prefixes[0]
+    name = prefixes[1].strip()
+    flags = header[header.index('<') + 1:header.index('>')].split(',')
+    return id, name, flags
+
+
+def find_values(data, key):
+    result = []
+    words = data.split(' ')
+    indexes = [index for index, word in enumerate(words) if word == key]
+    for index in indexes:
+        value_index = index + 1
+        if len(words) > value_index:
+            result.append(words[value_index])
+        else:
+            raise ValueNotFoundException(key)
+    if len(result) == 0:
+        raise ValueNotFoundException(key)
+    return result
+
+
+def find_values_or_default_value(data, key, default_value=None, single=False):
+    try:
+        result = find_values(data, key)
+        return result[0] if single else result
+    except ValueNotFoundException:
+        return default_value