blob: d09e78c1d518588ac50e4d9ff842b1a84c2fe5fb [file] [log] [blame]
Richard Jones64b02de2010-08-03 06:39:33 +00001"""Mock socket module used by the smtpd and smtplib tests.
2"""
3
4# imported for _GLOBAL_DEFAULT_TIMEOUT
5import socket as socket_module
6
7# Mock socket module
8_defaulttimeout = None
9_reply_data = None
10
11# This is used to queue up data to be read through socket.makefile, typically
12# *before* the socket object is even created. It is intended to handle a single
13# line which the socket will feed on recv() or makefile().
14def reply_with(line):
15 global _reply_data
16 _reply_data = line
17
18
19class MockFile:
20 """Mock file object returned by MockSocket.makefile().
21 """
22 def __init__(self, lines):
23 self.lines = lines
24 def readline(self):
25 return self.lines.pop(0) + b'\r\n'
26 def close(self):
27 pass
28
29
30class MockSocket:
31 """Mock socket object used by smtpd and smtplib tests.
32 """
33 def __init__(self):
34 global _reply_data
35 self.output = []
36 self.lines = []
37 if _reply_data:
38 self.lines.append(_reply_data)
Richard Jones6a9e6bb2010-08-04 12:27:36 +000039 _reply_data = None
Richard Jones64b02de2010-08-03 06:39:33 +000040 self.conn = None
41 self.timeout = None
42
43 def queue_recv(self, line):
44 self.lines.append(line)
45
46 def recv(self, bufsize, flags=None):
47 data = self.lines.pop(0) + b'\r\n'
48 return data
49
50 def fileno(self):
51 return 0
52
53 def settimeout(self, timeout):
54 if timeout is None:
55 self.timeout = _defaulttimeout
56 else:
57 self.timeout = timeout
58
59 def gettimeout(self):
60 return self.timeout
61
62 def setsockopt(self, level, optname, value):
63 pass
64
65 def getsockopt(self, level, optname, buflen=None):
66 return 0
67
68 def bind(self, address):
69 pass
70
71 def accept(self):
72 self.conn = MockSocket()
73 return self.conn, 'c'
74
75 def getsockname(self):
76 return ('0.0.0.0', 0)
77
78 def setblocking(self, flag):
79 pass
80
81 def listen(self, backlog):
82 pass
83
84 def makefile(self, mode='r', bufsize=-1):
85 handle = MockFile(self.lines)
86 return handle
87
88 def sendall(self, buffer, flags=None):
89 self.last = data
90 self.output.append(data)
91 return len(data)
92
93 def send(self, data, flags=None):
94 self.last = data
95 self.output.append(data)
96 return len(data)
97
98 def getpeername(self):
99 return 'peer'
100
101 def close(self):
102 pass
103
104
105def socket(family=None, type=None, proto=None):
106 return MockSocket()
107
108
Senthil Kumaran3d23fd62011-07-30 10:56:50 +0800109def create_connection(address, timeout=socket_module._GLOBAL_DEFAULT_TIMEOUT,
110 source_address=None):
Richard Jones64b02de2010-08-03 06:39:33 +0000111 try:
112 int_port = int(address[1])
113 except ValueError:
114 raise error
115 ms = MockSocket()
116 if timeout is socket_module._GLOBAL_DEFAULT_TIMEOUT:
117 timeout = getdefaulttimeout()
118 ms.settimeout(timeout)
119 return ms
120
121
122def setdefaulttimeout(timeout):
123 global _defaulttimeout
124 _defaulttimeout = timeout
125
126
127def getdefaulttimeout():
128 return _defaulttimeout
129
130
131def getfqdn():
132 return ""
133
134
135def gethostname():
136 pass
137
138
139def gethostbyname(name):
140 return ""
141
142
143class gaierror(Exception):
144 pass
145
146
147class error(Exception):
148 pass
149
150
151# Constants
152AF_INET = None
153SOCK_STREAM = None
154SOL_SOCKET = None
155SO_REUSEADDR = None