src/allmydata/test/test_iputil.py

   1
   2 import re, errno, subprocess, os, sys
   3
   4 from twisted.trial import unittest
   5
   6 from allmydata.util import iputil
   7 import allmydata.test.common_util as testutil
   8
   9
  10 class Namespace:
  11     pass
  12
  13 DOTTED_QUAD_RE=re.compile("^[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+$")
  14
  15 MOCK_IPADDR_OUTPUT = """\
  16 1: lo: <LOOPBACK,UP,LOWER_UP> mtu 16436 qdisc noqueue state UNKNOWN \n\
  17     link/loopback 00:00:00:00:00:00 brd 00:00:00:00:00:00
  18     inet 127.0.0.1/8 scope host lo
  19     inet6 ::1/128 scope host \n\
  20        valid_lft forever preferred_lft forever
  21 2: eth1: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc pfifo_fast state UP qlen 1000
  22     link/ether d4:3d:7e:01:b4:3e brd ff:ff:ff:ff:ff:ff
  23     inet 192.168.0.6/24 brd 192.168.0.255 scope global eth1
  24     inet6 fe80::d63d:7eff:fe01:b43e/64 scope link \n\
  25        valid_lft forever preferred_lft forever
  26 3: wlan0: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc mq state UP qlen 1000
  27     link/ether 90:f6:52:27:15:0a brd ff:ff:ff:ff:ff:ff
  28     inet 192.168.0.2/24 brd 192.168.0.255 scope global wlan0
  29     inet6 fe80::92f6:52ff:fe27:150a/64 scope link \n\
  30        valid_lft forever preferred_lft forever
  31 """
  32
  33 MOCK_IFCONFIG_OUTPUT = """\
  34 eth1      Link encap:Ethernet  HWaddr d4:3d:7e:01:b4:3e  \n\
  35           inet addr:192.168.0.6  Bcast:192.168.0.255  Mask:255.255.255.0
  36           inet6 addr: fe80::d63d:7eff:fe01:b43e/64 Scope:Link
  37           UP BROADCAST RUNNING MULTICAST  MTU:1500  Metric:1
  38           RX packets:154242234 errors:0 dropped:0 overruns:0 frame:0
  39           TX packets:155461891 errors:0 dropped:0 overruns:0 carrier:0
  40           collisions:0 txqueuelen:1000 \n\
  41           RX bytes:84367213640 (78.5 GiB)  TX bytes:73401695329 (68.3 GiB)
  42           Interrupt:20 Memory:f4f00000-f4f20000 \n\
  43
  44 lo        Link encap:Local Loopback  \n\
  45           inet addr:127.0.0.1  Mask:255.0.0.0
  46           inet6 addr: ::1/128 Scope:Host
  47           UP LOOPBACK RUNNING  MTU:16436  Metric:1
  48           RX packets:27449267 errors:0 dropped:0 overruns:0 frame:0
  49           TX packets:27449267 errors:0 dropped:0 overruns:0 carrier:0
  50           collisions:0 txqueuelen:0 \n\
  51           RX bytes:192643017823 (179.4 GiB)  TX bytes:192643017823 (179.4 GiB)
  52
  53 wlan0     Link encap:Ethernet  HWaddr 90:f6:52:27:15:0a  \n\
  54           inet addr:192.168.0.2  Bcast:192.168.0.255  Mask:255.255.255.0
  55           inet6 addr: fe80::92f6:52ff:fe27:150a/64 Scope:Link
  56           UP BROADCAST RUNNING MULTICAST  MTU:1500  Metric:1
  57           RX packets:12352750 errors:0 dropped:0 overruns:0 frame:0
  58           TX packets:4501451 errors:0 dropped:0 overruns:0 carrier:0
  59           collisions:0 txqueuelen:1000 \n\
  60           RX bytes:3916475942 (3.6 GiB)  TX bytes:458353654 (437.1 MiB)
  61 """
  62
  63
  64 class FakeProcess:
  65     def __init__(self, output, err):
  66         self.output = output
  67         self.err = err
  68     def communicate(self):
  69         return (self.output, self.err)
  70
  71
  72 class ListAddresses(testutil.SignalMixin, unittest.TestCase):
  73     def test_get_local_ip_for(self):
  74         addr = iputil.get_local_ip_for('127.0.0.1')
  75         self.failUnless(DOTTED_QUAD_RE.match(addr))
  76
  77     def test_list_async(self):
  78         d = iputil.get_local_addresses_async()
  79         def _check(addresses):
  80             self.failUnlessIn("127.0.0.1", addresses)
  81             self.failIfIn("0.0.0.0", addresses)
  82         d.addCallbacks(_check)
  83         return d
  84     # David A.'s OpenSolaris box timed out on this test one time when it was at 2s.
  85     test_list_async.timeout=4
  86
  87     def _test_list_async_mock(self, command, output):
  88         ns = Namespace()
  89         ns.first = True
  90
  91         def call_Popen(args, bufsize=0, executable=None, stdin=None, stdout=None, stderr=None,
  92                        preexec_fn=None, close_fds=False, shell=False, cwd=None, env=None,
  93                        universal_newlines=False, startupinfo=None, creationflags=0):
  94             if ns.first:
  95                 ns.first = False
  96                 e = OSError("EINTR")
  97                 e.errno = errno.EINTR
  98                 raise e
  99             elif os.path.basename(args[0]) == command:
 100                 return FakeProcess(output, "")
 101             else:
 102                 e = OSError("[Errno 2] No such file or directory")
 103                 e.errno = errno.ENOENT
 104                 raise e
 105         self.patch(subprocess, 'Popen', call_Popen)
 106
 107         def call_get_local_ip_for(target):
 108             if target in ("localhost", "127.0.0.1"):
 109                 return "127.0.0.1"
 110             else:
 111                 return "192.168.0.10"
 112         self.patch(iputil, 'get_local_ip_for', call_get_local_ip_for)
 113
 114         d = iputil.get_local_addresses_async()
 115         def _check(addresses):
 116             if sys.platform == "cygwin":
 117                 expected = set(["127.0.0.1", "192.168.0.10"])
 118             else:
 119                 expected = set(["127.0.0.1", "192.168.0.6", "192.168.0.2", "192.168.0.10"])
 120             self.failUnlessEquals(set(addresses), expected)
 121         d.addCallbacks(_check)
 122         return d
 123
 124     def test_list_async_mock_ip_addr(self):
 125         return self._test_list_async_mock("ip", MOCK_IPADDR_OUTPUT)
 126
 127     def test_list_async_mock_ifconfig(self):
 128         return self._test_list_async_mock("ifconfig", MOCK_IFCONFIG_OUTPUT)