From: Daira Hopwood Date: Tue, 25 Jun 2013 18:15:05 +0000 (+0100) Subject: iputil.py: add tests for recent changes. refs #1381, #1988, #982, #1064, #1536, ... X-Git-Tag: allmydata-tahoe-1.10.1a1~233 X-Git-Url: https://git.rkrishnan.org/provisioning?a=commitdiff_plain;h=a493ee0bb641175ecf918e28fce4d25df15994b6;p=tahoe-lafs%2Ftahoe-lafs.git iputil.py: add tests for recent changes. refs #1381, #1988, #982, #1064, #1536, #1935, #898, #1707, #1918 Signed-off-by: Daira Hopwood --- diff --git a/src/allmydata/test/test_iputil.py b/src/allmydata/test/test_iputil.py index fd64de42..b56b8d23 100644 --- a/src/allmydata/test/test_iputil.py +++ b/src/allmydata/test/test_iputil.py @@ -1,11 +1,74 @@ -import re +import re, errno, subprocess, os + from twisted.trial import unittest + from allmydata.util import iputil import allmydata.test.common_util as testutil + +class Namespace: + pass + DOTTED_QUAD_RE=re.compile("^[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+$") +MOCK_IPADDR_OUTPUT = """\ +1: lo: mtu 16436 qdisc noqueue state UNKNOWN \n\ + link/loopback 00:00:00:00:00:00 brd 00:00:00:00:00:00 + inet 127.0.0.1/8 scope host lo + inet6 ::1/128 scope host \n\ + valid_lft forever preferred_lft forever +2: eth1: mtu 1500 qdisc pfifo_fast state UP qlen 1000 + link/ether d4:3d:7e:01:b4:3e brd ff:ff:ff:ff:ff:ff + inet 192.168.0.6/24 brd 192.168.0.255 scope global eth1 + inet6 fe80::d63d:7eff:fe01:b43e/64 scope link \n\ + valid_lft forever preferred_lft forever +3: wlan0: mtu 1500 qdisc mq state UP qlen 1000 + link/ether 90:f6:52:27:15:0a brd ff:ff:ff:ff:ff:ff + inet 192.168.0.2/24 brd 192.168.0.255 scope global wlan0 + inet6 fe80::92f6:52ff:fe27:150a/64 scope link \n\ + valid_lft forever preferred_lft forever +""" + +MOCK_IFCONFIG_OUTPUT = """\ +eth1 Link encap:Ethernet HWaddr d4:3d:7e:01:b4:3e \n\ + inet addr:192.168.0.6 Bcast:192.168.0.255 Mask:255.255.255.0 + inet6 addr: fe80::d63d:7eff:fe01:b43e/64 Scope:Link + UP BROADCAST RUNNING MULTICAST MTU:1500 Metric:1 + RX packets:154242234 errors:0 dropped:0 overruns:0 frame:0 + TX packets:155461891 errors:0 dropped:0 overruns:0 carrier:0 + collisions:0 txqueuelen:1000 \n\ + RX bytes:84367213640 (78.5 GiB) TX bytes:73401695329 (68.3 GiB) + Interrupt:20 Memory:f4f00000-f4f20000 \n\ + +lo Link encap:Local Loopback \n\ + inet addr:127.0.0.1 Mask:255.0.0.0 + inet6 addr: ::1/128 Scope:Host + UP LOOPBACK RUNNING MTU:16436 Metric:1 + RX packets:27449267 errors:0 dropped:0 overruns:0 frame:0 + TX packets:27449267 errors:0 dropped:0 overruns:0 carrier:0 + collisions:0 txqueuelen:0 \n\ + RX bytes:192643017823 (179.4 GiB) TX bytes:192643017823 (179.4 GiB) + +wlan0 Link encap:Ethernet HWaddr 90:f6:52:27:15:0a \n\ + inet addr:192.168.0.2 Bcast:192.168.0.255 Mask:255.255.255.0 + inet6 addr: fe80::92f6:52ff:fe27:150a/64 Scope:Link + UP BROADCAST RUNNING MULTICAST MTU:1500 Metric:1 + RX packets:12352750 errors:0 dropped:0 overruns:0 frame:0 + TX packets:4501451 errors:0 dropped:0 overruns:0 carrier:0 + collisions:0 txqueuelen:1000 \n\ + RX bytes:3916475942 (3.6 GiB) TX bytes:458353654 (437.1 MiB) +""" + + +class FakeProcess: + def __init__(self, output, err): + self.output = output + self.err = err + def communicate(self): + return (self.output, self.err) + + class ListAddresses(testutil.SignalMixin, unittest.TestCase): def test_get_local_ip_for(self): addr = iputil.get_local_ip_for('127.0.0.1') @@ -14,10 +77,41 @@ class ListAddresses(testutil.SignalMixin, unittest.TestCase): def test_list_async(self): d = iputil.get_local_addresses_async() def _check(addresses): - self.failUnless(len(addresses) >= 1) # always have localhost - self.failUnless("127.0.0.1" in addresses, addresses) - self.failIf("0.0.0.0" in addresses, addresses) + self.failUnlessIn("127.0.0.1", addresses) + self.failIfIn("0.0.0.0", addresses) d.addCallbacks(_check) return d # David A.'s OpenSolaris box timed out on this test one time when it was at 2s. test_list_async.timeout=4 + + def _test_list_async_mock(self, command, output): + ns = Namespace() + ns.first = True + + def call_Popen(args, bufsize=0, executable=None, stdin=None, stdout=None, stderr=None, + preexec_fn=None, close_fds=False, shell=False, cwd=None, env=None, + universal_newlines=False, startupinfo=None, creationflags=0): + if ns.first: + ns.first = False + e = OSError("EINTR") + e.errno = errno.EINTR + raise e + elif os.path.basename(args[0]) == command: + return FakeProcess(output, "") + else: + e = OSError("not found") + e.errno = errno.EEXIST + raise e + self.patch(subprocess, 'Popen', call_Popen) + + d = iputil.get_local_addresses_async() + def _check(addresses): + self.failUnlessEquals(set(addresses), set(["127.0.0.1", "192.168.0.6", "192.168.0.2"])) + d.addCallbacks(_check) + return d + + def test_list_async_mock_ip_addr(self): + return self._test_list_async_mock("ip", MOCK_IPADDR_OUTPUT) + + def test_list_async_mock_ifconfig(self): + return self._test_list_async_mock("ifconfig", MOCK_IFCONFIG_OUTPUT)