Home | History | Annotate | Download | only in net 
      1 """Set of Mocks and stubs for network utilities unit tests.
      2 
      3 Implement a set of mocks and stubs use to implement unit tests
      4 for the network libraries.
      5 """
      6 
      7 import socket
      8 from autotest_lib.client.common_lib.test_utils import mock
      9 from autotest_lib.client.bin.net import net_utils
     10 
     11 
     12 def os_open(*args, **kwarg):
     13     return os_stub('open')
     14 
     15 
     16 class os_stub(mock.mock_function):
     17     def __init__(self, symbol, **kwargs):
     18         mock.mock_function.__init__(self, symbol, *kwargs)
     19 
     20     readval = ""
     21     def open(self, *args, **kwargs):
     22         return self
     23 
     24     def read(self, *args, **kwargs):
     25         return os_stub.readval
     26 
     27 
     28 def netutils_netif(iface):
     29     return netif_stub(iface, 'net_utils', net_utils.netif)
     30 
     31 
     32 class netif_stub(mock.mock_class):
     33     def __init__(self, iface, cls, name, *args, **kwargs):
     34         mock.mock_class.__init__(self, cls, name, args, *kwargs)
     35 
     36 
     37     def wait_for_carrier(self, timeout):
     38         return
     39 
     40 
     41 class socket_stub(mock.mock_class):
     42     """Class use to mock sockets."""
     43     def __init__(self, iface, cls, name, *args, **kwargs):
     44         mock.mock_class.__init__(self, cls, name, args, *kwargs)
     45         self.recv_val = ''
     46         self.throw_timeout = False
     47         self.send_val = None
     48         self.timeout = None
     49         self.family = None
     50         self.type = None
     51 
     52 
     53     def close(self):
     54         pass
     55 
     56 
     57     def socket(self, family, type):
     58         self.family = family
     59         self.type = type
     60 
     61 
     62     def settimeout(self, timeout):
     63         self.timeout = timeout
     64         return
     65 
     66 
     67     def send(self, buf):
     68         self.send_val = buf
     69 
     70 
     71     def recv(self, size):
     72         if self.throw_timeout:
     73             raise socket.timeout
     74 
     75         if len(self.recv_val) > size:
     76             return self.recv_val[:size]
     77         return self.recv_val
     78 
     79 
     80     def bind(self, arg):
     81         pass
     82 
     83 
     84 class network_interface_mock(net_utils.network_interface):
     85     def __init__(self, iface='some_name', test_init=False):
     86         self._test_init = test_init # test network_interface __init__()
     87         if self._test_init:
     88             super(network_interface_mock, self).__init__(iface)
     89             return
     90 
     91         self.ethtool = '/mock/ethtool'
     92         self._name = iface
     93         self.was_down = False
     94         self.orig_ipaddr = '1.2.3.4'
     95         self.was_loopback_enabled = False
     96         self._socket = socket_stub(iface, socket, socket)
     97 
     98         self.loopback_enabled = False
     99         self.driver = 'mock_driver'
    100 
    101 
    102     def is_down(self):
    103         if self._test_init:
    104             return 'is_down'
    105         return super(network_interface_mock, self).is_down()
    106 
    107 
    108     def get_ipaddr(self):
    109         if self._test_init:
    110             return 'get_ipaddr'
    111         return super(network_interface_mock, self).get_ipaddr()
    112 
    113 
    114     def is_loopback_enabled(self):
    115         if self._test_init:
    116             return 'is_loopback_enabled'
    117         return self.loopback_enabled
    118 
    119 
    120     def get_driver(self):
    121         return self.driver
    122 
    123 
    124     def wait_for_carrier(self, timeout=1):
    125         return
    126