• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1"""Set of Mocks and stubs for network utilities unit tests.
2
3Implement a set of mocks and stubs use to implement unit tests
4for the network libraries.
5"""
6
7import socket
8from autotest_lib.client.common_lib.test_utils import mock
9from autotest_lib.client.bin.net import net_utils
10
11
12def os_open(*args, **kwarg):
13    return os_stub('open')
14
15
16class 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
28def netutils_netif(iface):
29    return netif_stub(iface, 'net_utils', net_utils.netif)
30
31
32class 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
41class 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
84class 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