1  from unittest import mock
 2  
 3  SOCK_STREAM = 0
 4  
 5  set_interface = mock.Mock()
 6  interface = mock.MagicMock()
 7  getaddrinfo = mock.Mock()
 8  socket = mock.Mock()
 9  
10  
11  class Mocket:
12      def __init__(self, response):
13          self.settimeout = mock.Mock()
14          self.close = mock.Mock()
15          self.connect = mock.Mock()
16          self.send = mock.Mock(side_effect=self._send)
17          self.readline = mock.Mock(side_effect=self._readline)
18          self.recv = mock.Mock(side_effect=self._recv)
19          self.fail_next_send = False
20          self._response = response
21          self._position = 0
22  
23      def _send(self, data):
24          if self.fail_next_send:
25              self.fail_next_send = False
26              raise RuntimeError("Send failed")
27          return None
28  
29      def _readline(self):
30          i = self._response.find(b"\r\n", self._position)
31          r = self._response[self._position : i + 2]
32          self._position = i + 2
33          return r
34  
35      def _recv(self, count):
36          end = self._position + count
37          r = self._response[self._position : end]
38          self._position = end
39          print(r)
40          return r