mock_socket.py

"""Mock socket module used by the smtplib tests.
"""

# imported for _GLOBAL_DEFAULT_TIMEOUT
importsocketassocket_module

# Mock socket module
_defaulttimeout=None
_reply_data=None

# This is used to queue up data to be read through socket.makefile, typically
# *before* the socket object is even created. It is intended to handle a single
# line which the socket will feed on recv() or makefile().
defreply_with(line):
global_reply_data
_reply_data=line


classMockFile:
"""Mock file object returned by MockSocket.makefile().
 """
def__init__(self, lines):
self.lines=lines
defreadline(self, limit=-1):
result=self.lines.pop(0) +b'\r\n'
iflimit>=0:
# Re-insert the line, removing the \r\n we added.
self.lines.insert(0, result[limit:-2])
result=result[:limit]
returnresult
defclose(self):
pass


classMockSocket:
"""Mock socket object used by the smtplib tests.
 """
def__init__(self, family=None):
global_reply_data
self.family=family
self.output= []
self.lines= []
if_reply_data:
self.lines.append(_reply_data)
_reply_data=None
self.conn=None
self.timeout=None

defqueue_recv(self, line):
self.lines.append(line)

defrecv(self, bufsize, flags=None):
data=self.lines.pop(0) +b'\r\n'
returndata

deffileno(self):
return0

defsettimeout(self, timeout):
iftimeoutisNone:
self.timeout=_defaulttimeout
else:
self.timeout=timeout

defgettimeout(self):
returnself.timeout

defsetsockopt(self, level, optname, value):
pass

defgetsockopt(self, level, optname, buflen=None):
return0

defbind(self, address):
pass

defaccept(self):
self.conn=MockSocket()
returnself.conn, 'c'

defgetsockname(self):
return ('0.0.0.0', 0)

defsetblocking(self, flag):
pass

deflisten(self, backlog):
pass

defmakefile(self, mode='r', bufsize=-1):
handle=MockFile(self.lines)
returnhandle

defsendall(self, data, flags=None):
self.last=data
self.output.append(data)
returnlen(data)

defsend(self, data, flags=None):
self.last=data
self.output.append(data)
returnlen(data)

defgetpeername(self):
return ('peer-address', 'peer-port')

defclose(self):
pass

defconnect(self, host):
pass


defsocket(family=None, type=None, proto=None):
returnMockSocket(family)

defcreate_connection(address, timeout=socket_module._GLOBAL_DEFAULT_TIMEOUT,
source_address=None):
try:
int_port=int(address[1])
exceptValueError:
raiseerror
ms=MockSocket()
iftimeoutissocket_module._GLOBAL_DEFAULT_TIMEOUT:
timeout=getdefaulttimeout()
ms.settimeout(timeout)
returnms


defsetdefaulttimeout(timeout):
global_defaulttimeout
_defaulttimeout=timeout


defgetdefaulttimeout():
return_defaulttimeout


defgetfqdn():
return""


defgethostname():
pass


defgethostbyname(name):
return""

defgetaddrinfo(*args, **kw):
returnsocket_module.getaddrinfo(*args, **kw)

gaierror=socket_module.gaierror
error=socket_module.error


# Constants
_GLOBAL_DEFAULT_TIMEOUT=socket_module._GLOBAL_DEFAULT_TIMEOUT
AF_INET=socket_module.AF_INET
AF_INET6=socket_module.AF_INET6
SOCK_STREAM=socket_module.SOCK_STREAM
SOL_SOCKET=None
SO_REUSEADDR=None

ifhasattr(socket_module, 'AF_UNIX'):
AF_UNIX=socket_module.AF_UNIX