iframe-proxy | Sunbelt Computer Software

History

166 lines (123 loc) · 3.7 KB

Raw

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

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

"""

# imported for _GLOBAL_DEFAULT_TIMEOUT

import socket as socket_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().

def reply_with(line):

global _reply_data

_reply_data = line

class MockFile:

"""Mock file object returned by MockSocket.makefile().

"""

def __init__(self, lines):

self.lines = lines

def readline(self, limit=-1):

result = self.lines.pop(0) + b'\r\n'

if limit >= 0:

# Re-insert the line, removing the \r\n we added.

self.lines.insert(0, result[limit:-2])

result = result[:limit]

return result

def close(self):

pass

class MockSocket:

"""Mock socket object used by smtpd and 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

def queue_recv(self, line):

self.lines.append(line)

def recv(self, bufsize, flags=None):

data = self.lines.pop(0) + b'\r\n'

return data

def fileno(self):

return 0

def settimeout(self, timeout):

if timeout is None:

self.timeout = _defaulttimeout

else:

self.timeout = timeout

def gettimeout(self):

return self.timeout

def setsockopt(self, level, optname, value):

pass

def getsockopt(self, level, optname, buflen=None):

return 0

def bind(self, address):

pass

def accept(self):

self.conn = MockSocket()

return self.conn, 'c'

def getsockname(self):

return ('0.0.0.0', 0)

def setblocking(self, flag):

pass

def listen(self, backlog):

pass

def makefile(self, mode='r', bufsize=-1):

handle = MockFile(self.lines)

return handle

def sendall(self, data, flags=None):

self.last = data

self.output.append(data)

return len(data)

def send(self, data, flags=None):

self.last = data

self.output.append(data)

return len(data)

def getpeername(self):

return ('peer-address', 'peer-port')

def close(self):

pass

def connect(self, host):

pass

def socket(family=None, type=None, proto=None):

return MockSocket(family)

def create_connection(address, timeout=socket_module._GLOBAL_DEFAULT_TIMEOUT,

source_address=None):

try:

int_port = int(address[1])

except ValueError:

raise error

ms = MockSocket()

if timeout is socket_module._GLOBAL_DEFAULT_TIMEOUT:

timeout = getdefaulttimeout()

ms.settimeout(timeout)

return ms

def setdefaulttimeout(timeout):

global _defaulttimeout

_defaulttimeout = timeout

def getdefaulttimeout():

return _defaulttimeout

def getfqdn():

return ""

def gethostname():

pass

def gethostbyname(name):

return ""

def getaddrinfo(*args, **kw):

return socket_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

if hasattr(socket_module, 'AF_UNIX'):

AF_UNIX = socket_module.AF_UNIX

Sunbelt Computer Software

PL/B Language Development and Support

Navigation Menu

Uh oh!

Sunbelt Computer Software

PL/B Language Development and Support

Uh oh!

FilesExpand file tree

mock_socket.py

Latest commit

History

mock_socket.py

File metadata and controls