diff --git a/neutron/agent/windows/ip_lib.py b/neutron/agent/windows/ip_lib.py index 873f163117f..cd2439134b2 100644 --- a/neutron/agent/windows/ip_lib.py +++ b/neutron/agent/windows/ip_lib.py @@ -45,20 +45,41 @@ class IPWrapper(object): class IPDevice(object): def __init__(self, name): - self.device_name = name + self.name = name + self.link = IPLink(self) - def device_has_ip(self, ip): + def read_ifaddresses(self): try: - device_addresses = netifaces.ifaddresses(self.device_name) + device_addresses = netifaces.ifaddresses(self.name) except ValueError: LOG.error(_LE("The device does not exist on the system: %s."), - self.device_name) - return False + self.name) + return except OSError: - LOG.error(_LE("Failed to get ip addresses for interface: %s."), - self.device_name) + LOG.error(_LE("Failed to get interface addresses: %s."), + self.name) + return + return device_addresses + + def device_has_ip(self, ip): + device_addresses = self.read_ifaddresses() + if device_addresses is None: return False addresses = [ip_addr['addr'] for ip_addr in device_addresses.get(netifaces.AF_INET, [])] return ip in addresses + + +class IPLink(object): + + def __init__(self, parent): + self._parent = parent + + @property + def address(self): + device_addresses = self._parent.read_ifaddresses() + if device_addresses is None: + return False + return [eth_addr['addr'] for eth_addr in + device_addresses.get(netifaces.AF_LINK, [])] diff --git a/neutron/tests/functional/agent/windows/test_ip_lib.py b/neutron/tests/functional/agent/windows/test_ip_lib.py index 106af1283ac..123d7ef86fd 100644 --- a/neutron/tests/functional/agent/windows/test_ip_lib.py +++ b/neutron/tests/functional/agent/windows/test_ip_lib.py @@ -18,6 +18,7 @@ from neutron.tests import base WRONG_IP = '0.0.0.0' TEST_IP = '127.0.0.1' +TEST_MAC = '00:00:00:00:00:00' class IpLibTestCase(base.BaseTestCase): @@ -27,8 +28,17 @@ class IpLibTestCase(base.BaseTestCase): def test_ipwrapper_get_device_by_ip(self): ip_dev = ip_lib.IPWrapper().get_device_by_ip(TEST_IP) - self.assertEqual('lo', ip_dev.device_name) + self.assertEqual('lo', ip_dev.name) def test_device_has_ip(self): not_a_device = ip_lib.IPDevice('#!#._not_a_device_bleargh!!@@@') self.assertFalse(not_a_device.device_has_ip(TEST_IP)) + + def test_ip_link_read_mac_address(self): + ip_dev = ip_lib.IPWrapper().get_device_by_ip(TEST_IP) + self.assertEqual([TEST_MAC], ip_lib.IPLink(ip_dev).address) + + def test_ip_link_read_mac_address_wrong(self): + not_a_device = ip_lib.IPDevice('#!#._not_a_device_bleargh!!@@@') + mac_addr = ip_lib.IPLink(not_a_device).address + self.assertFalse(mac_addr) diff --git a/neutron/tests/unit/agent/windows/test_ip_lib.py b/neutron/tests/unit/agent/windows/test_ip_lib.py index c6051d7564f..bf4cd73c54e 100644 --- a/neutron/tests/unit/agent/windows/test_ip_lib.py +++ b/neutron/tests/unit/agent/windows/test_ip_lib.py @@ -43,8 +43,8 @@ class TestIpWrapper(base.BaseTestCase): mock.sentinel.dev2] ret = ip_lib.IPWrapper().get_devices() - self.assertEqual(mock.sentinel.dev1, ret[0].device_name) - self.assertEqual(mock.sentinel.dev2, ret[1].device_name) + self.assertEqual(mock.sentinel.dev1, ret[0].name) + self.assertEqual(mock.sentinel.dev2, ret[1].name) @mock.patch('netifaces.interfaces') def test_get_devices_error(self, mock_interfaces): @@ -57,38 +57,71 @@ class TestIpWrapper(base.BaseTestCase): class TestIpDevice(base.BaseTestCase): @mock.patch('netifaces.ifaddresses') - def test_device_has_ip(self, mock_netifaces): + def test_read_ifaddresses(self, mock_netifaces): mock_address = {'addr': mock.sentinel.fake_addr} mock_netifaces.return_value = {netifaces.AF_INET: [mock_address]} - - ret = ip_lib.IPDevice("fake_dev").device_has_ip( - mock.sentinel.fake_addr) - + ret = ip_lib.IPDevice("fake_dev").read_ifaddresses() self.assertTrue(ret) @mock.patch('netifaces.ifaddresses') - def test_device_has_ip_false(self, mock_netifaces): + def test_read_ifaddresses_no_ip(self, mock_netifaces): mock_netifaces.return_value = {} - - ret = ip_lib.IPDevice("fake_dev").device_has_ip( - mock.sentinel.fake_addr) - + ret = ip_lib.IPDevice("fake_dev").read_ifaddresses() self.assertFalse(ret) @mock.patch('netifaces.ifaddresses') - def test_device_has_ip_error(self, mock_netifaces): + def test_read_ifaddresses_ip_error(self, mock_netifaces): mock_netifaces.side_effect = OSError - - ret = ip_lib.IPDevice("fake_dev").device_has_ip( - mock.sentinel.fake_addr) - + ret = ip_lib.IPDevice("fake_dev").read_ifaddresses() self.assertFalse(ret) @mock.patch('netifaces.ifaddresses') - def test_device_not_found(self, mock_netifaces): + def test_read_faddresses_not_found(self, mock_netifaces): mock_netifaces.side_effect = ValueError - - ret = ip_lib.IPDevice("fake_dev").device_has_ip( - mock.sentinel.fake_addr) - + ret = ip_lib.IPDevice("fake_dev").read_ifaddresses() self.assertFalse(ret) + + def test_device_has_ip(self): + mock_address = {'addr': mock.sentinel.fake_addr} + ip_device = ip_lib.IPDevice("fake_dev") + with mock.patch.object(ip_device, "read_ifaddresses", return_value=( + {netifaces.AF_INET: [mock_address]})): + ret = ip_device.device_has_ip(mock.sentinel.fake_addr) + self.assertTrue(ret) + + def test_device_has_ip_false(self): + ip_device = ip_lib.IPDevice("fake_dev") + with mock.patch.object(ip_device, "read_ifaddresses", return_value={}): + ret = ip_device.device_has_ip(mock.sentinel.fake_addr) + self.assertFalse(ret) + + def test_device_has_ip_error(self): + ip_device = ip_lib.IPDevice("fake_dev") + with mock.patch.object(ip_device, "read_ifaddresses", + return_value=None): + ret = ip_device.device_has_ip(mock.sentinel.fake_addr) + self.assertFalse(ret) + + +class TestIPLink(base.BaseTestCase): + + def setUp(self): + super(TestIPLink, self).setUp() + parent = ip_lib.IPDevice("fake_dev") + self.ip_link = ip_lib.IPLink(parent) + self.ip_link._parent.read_ifaddresses = mock.Mock() + + def test_address(self): + mock_address = {'addr': mock.sentinel.fake_addr} + self.ip_link._parent.read_ifaddresses.return_value = { + netifaces.AF_LINK: [mock_address]} + self.assertEqual([mock_address['addr']], self.ip_link.address) + + def test_address_no_address(self): + self.ip_link._parent.read_ifaddresses.return_value = { + netifaces.AF_LINK: []} + self.assertEqual([], self.ip_link.address) + + def test_address_error(self): + self.ip_link._parent.read_ifaddresses.return_value = None + self.assertFalse(self.ip_link.address)