Python code coverage for Lib/test/mock_socket.py

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