Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def serve_file(_, writer):
out_fd = writer.transport.get_extra_info('socket').fileno()
size = (yield from aiofiles.os.stat(filename)).st_size
in_file = yield from aiofiles.open(filename)
try:
in_fd = in_file.fileno()
yield from aiofiles.os.sendfile(out_fd, in_fd, 0, size)
finally:
yield from in_file.close()
yield from writer.drain()
writer.close()
"""Test the sendfile functionality, file-to-file."""
filename = join(dirname(__file__), 'resources', 'test_file1.txt')
tmp_filename = tmpdir.join('tmp.bin')
with open(filename) as f:
contents = f.read()
input_file = yield from aiofiles.open(filename)
output_file = yield from aiofiles.open(str(tmp_filename), mode='w+')
size = (yield from aiofiles.os.stat(filename)).st_size
input_fd = input_file.fileno()
output_fd = output_file.fileno()
yield from aiofiles.os.sendfile(output_fd, input_fd, 0, size)
yield from output_file.seek(0)
actual_contents = yield from output_file.read()
actual_size = (yield from aiofiles.os.stat(str(tmp_filename))).st_size
assert contents == actual_contents
assert size == actual_size