def fake_socket(tmpdir_factory, request):
packets = request.node.callspec.params.get('packets')
# write data to the "socket"; this must be an actual file, because
# select.select()() expects a real file descriptor
filename = str(tmpdir_factory.mktemp('fake-socket').join(str(uuid4())))
f = open(filename, 'w')
request.addfinalizer(f.close)
sockobj = socket._socket.socket if six.PY3 else socket._socketobject
class fakesocket(sockobj):
buff = six.BytesIO()
def _send(self, data):
self.buff.write(data)
return len(data) # number of bytes sent
sock = fakesocket()
sock.f = six.BytesIO(packets)
# socket.socket overrides these at instantiation time to the underlying
# C implementation; set them here so that we can mock send() and recv()
# calls
sock.send = types.MethodType(_send, sock)
sock.recv = types.MethodType(
lambda self, _bytes: self.f.read(_bytes),
sock
)
sock.fileno = types.MethodType(lambda self: f.fileno(), sock)
return sock
评论列表
文章目录