ยปCore Development>Code coverage>Lib/test/mock_socket.py

Python code coverage for Lib/test/mock_socket.py

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