tests for sendfile functionality import asyncio import errno import os import socket import sys import tempfile import unittest from asyncio import baseevents from asyncio import constants from unittest import mock from test import support from test support import oshelper from test support import sockethelper from test testasyncio import utils as testutils try import ssl except importerror ssl none def teardownmodule asyncio seteventlooppolicynone class mysendfileprotoasyncio protocol def initself loopnone closeafter0 self transport none self state initial self nbytes 0 if loop is not none self connected loop createfuture self done loop createfuture self data bytearray self closeafter closeafter def assertstateself expected if self state not in expected raise assertionerrorf state self state r expected expected r def connectionmadeself transport self transport transport self assertstate initial self state connected if self connected self connected setresultnone def eofreceivedself self assertstate connected self state eof def connectionlostself exc self assertstate connected eof self state closed if self done self done setresultnone def datareceivedself data self assertstate connected self nbytes lendata self data extenddata super datareceiveddata if self closeafter and self nbytes self closeafter self transport close class myprotoasyncio protocol def initself loop self started false self closed false self data bytearray self fut loop createfuture self transport none def connectionmadeself transport self started true self transport transport def datareceivedself data self data extenddata def connectionlostself exc self closed true self fut setresultnone async def waitclosedself await self fut class sendfilebase 256 kib plus small unaligned to buffer chunk newer versions of windows seems to have increased its internal buffer and tries to send as much of the data as it can as it has some form of buffering for this which is less than 256kib on newer server versions and windows 11 so data should be larger than 256 kib to make this test reliable data bx 1024 256 1 reduce socket buffer size to test on relative small data sets bufsize 4 1024 4 kib def createeventloopself raise notimplementederror classmethod def setupclasscls with openoshelper testfn wb as fp fp writecls data super setupclass classmethod def teardownclasscls oshelper unlinkoshelper testfn super teardownclass def setupself self file openoshelper testfn rb self addcleanupself file close self loop self createeventloop self seteventloopself loop super setup def teardownself just in case if we have transport close callbacks if not self loop isclosed testutils runbrieflyself loop self docleanups support gccollect super teardown def runloopself coro return self loop rununtilcompletecoro class socksendfilemixinsendfilebase classmethod def setupclasscls cls oldbufsize constants sendfilefallbackreadbuffersize constants sendfilefallbackreadbuffersize 1024 16 super setupclass classmethod def teardownclasscls constants sendfilefallbackreadbuffersize cls oldbufsize super teardownclass def makesocketself cleanuptrue sock socket socketsocket afinet socket sockstream sock setblockingfalse if cleanup self addcleanupsock close return sock def reducereceivebuffersizeself sock reduce receive socket buffer size to test on relative small data sets sock setsockoptsocket solsocket socket sorcvbuf self bufsize def reducesendbuffersizeself sock transportnone reduce send socket buffer size to test on relative small data sets on macos sosndbuf is reset by connect so this method should be called after the socket is connected sock setsockoptsocket solsocket socket sosndbuf self bufsize if transport is not none transport setwritebufferlimitshighself bufsize def preparesocksendfileself proto myprotoself loop port sockethelper findunusedport srvsock self makesocketcleanupfalse srvsock bindsockethelper host port server self runloopself loop createserver lambda proto socksrvsock self reducereceivebuffersizesrvsock sock self makesocket self runloopself loop sockconnectsock 127 0 0 1 port self reducesendbuffersizesock def cleanup if proto transport is not none can be none if the task was cancelled before connectionmade callback proto transport close self runloopproto waitclosed server close self runloopserver waitclosed self addcleanupcleanup return sock proto def testsocksendfilesuccessself sock proto self preparesocksendfile ret self runloopself loop socksendfilesock self file sock close self runloopproto waitclosed self assertequalret lenself data self assertequalproto data self data self assertequalself file tell lenself data def testsocksendfilewithoffsetandcountself sock proto self preparesocksendfile ret self runloopself loop socksendfilesock self file 1000 2000 sock close self runloopproto waitclosed self assertequalproto data self data1000 3000 self assertequalself file tell 3000 self assertequalret 2000 def testsocksendfilezerosizeself sock proto self preparesocksendfile with tempfile temporaryfile as f ret self runloopself loop socksendfilesock f 0 none sock close self runloopproto waitclosed self assertequalret 0 self assertequalself file tell 0 def testsocksendfilemixwithregularsendself buf bmixregularsend 4 1024 64 kib sock proto self preparesocksendfile self runloopself loop socksendallsock buf ret self runloopself loop socksendfilesock self file self runloopself loop socksendallsock buf sock close self runloopproto waitclosed self assertequalret lenself data expected buf self data buf self assertequalproto data expected self assertequalself file tell lenself data class sendfilemixinsendfilebase note sendfile via ssl transport is equal to sendfile fallback def preparesendfileself issslfalse closeafter0 port sockethelper findunusedport srvproto mysendfileprotoloopself loop closeaftercloseafter if isssl if not ssl self skiptestno ssl module srvctx testutils simpleserversslcontext clictx testutils simpleclientsslcontext else srvctx none clictx none srvsock socket socketsocket afinet socket sockstream srvsock bindsockethelper host port server self runloopself loop createserver lambda srvproto socksrvsock sslsrvctx self reducereceivebuffersizesrvsock if isssl serverhostname sockethelper host else serverhostname none clisock socket socketsocket afinet socket sockstream clisock connectsockethelper host port cliproto mysendfileprotoloopself loop tr pr self runloopself loop createconnection lambda cliproto sockclisock sslclictx serverhostnameserverhostname self reducesendbuffersizeclisock transporttr def cleanup srvproto transport close cliproto transport close self runloopsrvproto done self runloopcliproto done server close self runloopserver waitclosed self addcleanupcleanup return srvproto cliproto unittest skipifsys platform win32 udp sockets are not supported def testsendfilenotsupportedself tr pr self runloop self loop createdatagramendpoint asyncio datagramprotocol familysocket afinet try with self assertraisesregexruntimeerror not supported self runloop self loop sendfiletr self file self assertequal0 self file tell finally don t use self addcleanup because it produces resource warning tr close def testsendfileself srvproto cliproto self preparesendfile ret self runloop self loop sendfilecliproto transport self file cliproto transport close self runloopsrvproto done self assertequalret lenself data self assertequalsrvproto nbytes lenself data self assertequalsrvproto data self data self assertequalself file tell lenself data def testsendfileforcefallbackself srvproto cliproto self preparesendfile def sendfilenativetransp file offset count to raise sendfilenotavailableerror return baseevents baseeventloop sendfilenative self loop transp file offset count self loop sendfilenative sendfilenative ret self runloop self loop sendfilecliproto transport self file cliproto transport close self runloopsrvproto done self assertequalret lenself data self assertequalsrvproto nbytes lenself data self assertequalsrvproto data self data self assertequalself file tell lenself data def testsendfileforceunsupportednativeself if sys platform win32 if isinstanceself loop asyncio proactoreventloop self skiptestfails on proactor event loop srvproto cliproto self preparesendfile def sendfilenativetransp file offset count to raise sendfilenotavailableerror return baseevents baseeventloop sendfilenative self loop transp file offset count self loop sendfilenative sendfilenative with self assertraisesregexasyncio sendfilenotavailableerror not supported self runloop self loop sendfilecliproto transport self file fallbackfalse cliproto transport close self runloopsrvproto done self assertequalsrvproto nbytes 0 self assertequalself file tell 0 def testsendfilesslself srvproto cliproto self preparesendfileisssltrue ret self runloop self loop sendfilecliproto transport self file cliproto transport close self runloopsrvproto done self assertequalret lenself data self assertequalsrvproto nbytes lenself data self assertequalsrvproto data self data self assertequalself file tell lenself data def testsendfileforclosingtranspself srvproto cliproto self preparesendfile cliproto transport close with self assertraisesregexruntimeerror is closing self runloopself loop sendfilecliproto transport self file self runloopsrvproto done self assertequalsrvproto nbytes 0 self assertequalself file tell 0 def testsendfilepreandpostdataself srvproto cliproto self preparesendfile prefix b prefix 1024 8 kib suffix b suffix 1024 8 kib cliproto transport writeprefix ret self runloop self loop sendfilecliproto transport self file cliproto transport writesuffix cliproto transport close self runloopsrvproto done self assertequalret lenself data self assertequalsrvproto data prefix self data suffix self assertequalself file tell lenself data def testsendfilesslpreandpostdataself srvproto cliproto self preparesendfileisssltrue prefix b zxcvbnm 1024 suffix b 0987654321 1024 cliproto transport writeprefix ret self runloop self loop sendfilecliproto transport self file cliproto transport writesuffix cliproto transport close self runloopsrvproto done self assertequalret lenself data self assertequalsrvproto data prefix self data suffix self assertequalself file tell lenself data def testsendfilepartialself srvproto cliproto self preparesendfile ret self runloop self loop sendfilecliproto transport self file 1000 100 cliproto transport close self runloopsrvproto done self assertequalret 100 self assertequalsrvproto nbytes 100 self assertequalsrvproto data self data1000 1100 self assertequalself file tell 1100 def testsendfilesslpartialself srvproto cliproto self preparesendfileisssltrue ret self runloop self loop sendfilecliproto transport self file 1000 100 cliproto transport close self runloopsrvproto done self assertequalret 100 self assertequalsrvproto nbytes 100 self assertequalsrvproto data self data1000 1100 self assertequalself file tell 1100 def testsendfileclosepeerafterreceivingself srvproto cliproto self preparesendfile closeafterlenself data ret self runloop self loop sendfilecliproto transport self file cliproto transport close self runloopsrvproto done self assertequalret lenself data self assertequalsrvproto nbytes lenself data self assertequalsrvproto data self data self assertequalself file tell lenself data def testsendfilesslclosepeerafterreceivingself srvproto cliproto self preparesendfile isssltrue closeafterlenself data ret self runloop self loop sendfilecliproto transport self file self runloopsrvproto done self assertequalret lenself data self assertequalsrvproto nbytes lenself data self assertequalsrvproto data self data self assertequalself file tell lenself data on solaris lowering sorcvbuf on a tcp connection after it has been established has no effect due to its age this bug affects both oracle solaris as well as all other opensolaris forks unless they fixed it themselves unittest skipifsys platform startswith sunos doesn t work on solaris def testsendfileclosepeerinthemiddleofreceivingself srvproto cliproto self preparesendfilecloseafter1024 with self assertraisesconnectionerror self runloop self loop sendfilecliproto transport self file self runloopsrvproto done self asserttrue1024 srvproto nbytes lenself data srvproto nbytes if not sys platform win32 and isinstanceself loop asyncio proactoreventloop on windows proactor uses transmitfile which does not update tell self asserttrue1024 self file tell lenself data self file tell self asserttruecliproto transport isclosing def testsendfilefallbackclosepeerinthemiddleofreceivingself def sendfilenativetransp file offset count to raise sendfilenotavailableerror return baseevents baseeventloop sendfilenative self loop transp file offset count self loop sendfilenative sendfilenative srvproto cliproto self preparesendfilecloseafter1024 with self assertraisesconnectionerror try self runloop self loop sendfilecliproto transport self file except oserror as e macos may raise oserror of eprototype when writing to a socket that is in the process of closing down if e errno errno eprototype and sys platform darwin raise connectionerror else raise self runloopsrvproto done self asserttrue1024 srvproto nbytes lenself data srvproto nbytes self asserttrue1024 self file tell lenself data self file tell unittest skipifnot hasattros sendfile don t have native sendfile support def testsendfilepreventsbarewriteself srvproto cliproto self preparesendfile fut self loop createfuture async def coro fut setresultnone return await self loop sendfilecliproto transport self file t self loop createtaskcoro self runloopfut with self assertraisesregexruntimeerror sendfile is in progress cliproto transport writeb data ret self runloopt self assertequalret lenself data def testsendfilenofallbackforfallbacktransportself transport mock mock transport isclosing sideeffect lambda false transport sendfilecompatible constants sendfilemode fallback with self assertraisesregexruntimeerror fallback is disabled self loop rununtilcomplete self loop sendfiletransport none fallbackfalse class sendfiletestsbasesendfilemixin socksendfilemixin pass if sys platform win32 class selecteventlooptestssendfiletestsbase testutils testcase def createeventloopself return asyncio selectoreventloop class proactoreventlooptestssendfiletestsbase testutils testcase def createeventloopself return asyncio proactoreventloop else import selectors if hasattrselectors kqueueselector class kqueueeventlooptestssendfiletestsbase testutils testcase def createeventloopself return asyncio selectoreventloop selectors kqueueselector if hasattrselectors epollselector class epolleventlooptestssendfiletestsbase testutils testcase def createeventloopself return asyncio selectoreventloopselectors epollselector if hasattrselectors pollselector class polleventlooptestssendfiletestsbase testutils testcase def createeventloopself return asyncio selectoreventloopselectors pollselector should always exist class selecteventlooptestssendfiletestsbase testutils testcase def createeventloopself return asyncio selectoreventloopselectors selectselector if name main unittest main 256 kib plus small unaligned to buffer chunk newer versions of windows seems to have increased its internal buffer and tries to send as much of the data as it can as it has some form of buffering for this which is less than 256kib on newer server versions and windows 11 so data should be larger than 256 kib to make this test reliable reduce socket buffer size to test on relative small data sets 4 kib just in case if we have transport close callbacks reduce receive socket buffer size to test on relative small data sets reduce send socket buffer size to test on relative small data sets on macos so_sndbuf is reset by connect so this method should be called after the socket is connected can be none if the task was cancelled before connection_made callback 64 kib note sendfile via ssl transport is equal to sendfile fallback don t use self addcleanup because it produces resource warning to raise sendfilenotavailableerror to raise sendfilenotavailableerror 8 kib 8 kib on solaris lowering so_rcvbuf on a tcp connection after it has been established has no effect due to its age this bug affects both oracle solaris as well as all other opensolaris forks unless they fixed it themselves on windows proactor uses transmitfile which does not update tell to raise sendfilenotavailableerror macos may raise oserror of eprototype when writing to a socket that is in the process of closing down should always exist
import asyncio
import errno
import os
import socket
import sys
import tempfile
import unittest
from asyncio import base_events
from asyncio import constants
from unittest import mock
from test import support
from test.support import os_helper
from test.support import socket_helper
from test.test_asyncio import utils as test_utils

try:
    import ssl
except ImportError:
    ssl = None


def tearDownModule():
    asyncio.set_event_loop_policy(None)


class MySendfileProto(asyncio.Protocol):

    def __init__(self, loop=None, close_after=0):
        self.transport = None
        self.state = 'INITIAL'
        self.nbytes = 0
        if loop is not None:
            self.connected = loop.create_future()
            self.done = loop.create_future()
        self.data = bytearray()
        self.close_after = close_after

    def _assert_state(self, *expected):
        if self.state not in expected:
            raise AssertionError(f'state: {self.state!r}, expected: {expected!r}')

    def connection_made(self, transport):
        self.transport = transport
        self._assert_state('INITIAL')
        self.state = 'CONNECTED'
        if self.connected:
            self.connected.set_result(None)

    def eof_received(self):
        self._assert_state('CONNECTED')
        self.state = 'EOF'

    def connection_lost(self, exc):
        self._assert_state('CONNECTED', 'EOF')
        self.state = 'CLOSED'
        if self.done:
            self.done.set_result(None)

    def data_received(self, data):
        self._assert_state('CONNECTED')
        self.nbytes += len(data)
        self.data.extend(data)
        super().data_received(data)
        if self.close_after and self.nbytes >= self.close_after:
            self.transport.close()


class MyProto(asyncio.Protocol):

    def __init__(self, loop):
        self.started = False
        self.closed = False
        self.data = bytearray()
        self.fut = loop.create_future()
        self.transport = None

    def connection_made(self, transport):
        self.started = True
        self.transport = transport

    def data_received(self, data):
        self.data.extend(data)

    def connection_lost(self, exc):
        self.closed = True
        self.fut.set_result(None)

    async def wait_closed(self):
        await self.fut


class SendfileBase:

    
    
    
    
    
    
    DATA = b"x" * (1024 * 256 + 1)
    
    BUF_SIZE = 4 * 1024   

    def create_event_loop(self):
        raise NotImplementedError

    @classmethod
    def setUpClass(cls):
        with open(os_helper.TESTFN, 'wb') as fp:
            fp.write(cls.DATA)
        super().setUpClass()

    @classmethod
    def tearDownClass(cls):
        os_helper.unlink(os_helper.TESTFN)
        super().tearDownClass()

    def setUp(self):
        self.file = open(os_helper.TESTFN, 'rb')
        self.addCleanup(self.file.close)
        self.loop = self.create_event_loop()
        self.set_event_loop(self.loop)
        super().setUp()

    def tearDown(self):
        
        if not self.loop.is_closed():
            test_utils.run_briefly(self.loop)

        self.doCleanups()
        support.gc_collect()
        super().tearDown()

    def run_loop(self, coro):
        return self.loop.run_until_complete(coro)


class SockSendfileMixin(SendfileBase):

    @classmethod
    def setUpClass(cls):
        cls.__old_bufsize = constants.SENDFILE_FALLBACK_READBUFFER_SIZE
        constants.SENDFILE_FALLBACK_READBUFFER_SIZE = 1024 * 16
        super().setUpClass()

    @classmethod
    def tearDownClass(cls):
        constants.SENDFILE_FALLBACK_READBUFFER_SIZE = cls.__old_bufsize
        super().tearDownClass()

    def make_socket(self, cleanup=True):
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.setblocking(False)
        if cleanup:
            self.addCleanup(sock.close)
        return sock

    def reduce_receive_buffer_size(self, sock):
        
        
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, self.BUF_SIZE)

    def reduce_send_buffer_size(self, sock, transport=None):
        

        
        
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, self.BUF_SIZE)

        if transport is not None:
            transport.set_write_buffer_limits(high=self.BUF_SIZE)

    def prepare_socksendfile(self):
        proto = MyProto(self.loop)
        port = socket_helper.find_unused_port()
        srv_sock = self.make_socket(cleanup=False)
        srv_sock.bind((socket_helper.HOST, port))
        server = self.run_loop(self.loop.create_server(
            lambda: proto, sock=srv_sock))
        self.reduce_receive_buffer_size(srv_sock)

        sock = self.make_socket()
        self.run_loop(self.loop.sock_connect(sock, ('127.0.0.1', port)))
        self.reduce_send_buffer_size(sock)

        def cleanup():
            if proto.transport is not None:
                
                
                proto.transport.close()
                self.run_loop(proto.wait_closed())

            server.close()
            self.run_loop(server.wait_closed())

        self.addCleanup(cleanup)

        return sock, proto

    def test_sock_sendfile_success(self):
        sock, proto = self.prepare_socksendfile()
        ret = self.run_loop(self.loop.sock_sendfile(sock, self.file))
        sock.close()
        self.run_loop(proto.wait_closed())

        self.assertEqual(ret, len(self.DATA))
        self.assertEqual(proto.data, self.DATA)
        self.assertEqual(self.file.tell(), len(self.DATA))

    def test_sock_sendfile_with_offset_and_count(self):
        sock, proto = self.prepare_socksendfile()
        ret = self.run_loop(self.loop.sock_sendfile(sock, self.file,
                                                    1000, 2000))
        sock.close()
        self.run_loop(proto.wait_closed())

        self.assertEqual(proto.data, self.DATA[1000:3000])
        self.assertEqual(self.file.tell(), 3000)
        self.assertEqual(ret, 2000)

    def test_sock_sendfile_zero_size(self):
        sock, proto = self.prepare_socksendfile()
        with tempfile.TemporaryFile() as f:
            ret = self.run_loop(self.loop.sock_sendfile(sock, f,
                                                        0, None))
        sock.close()
        self.run_loop(proto.wait_closed())

        self.assertEqual(ret, 0)
        self.assertEqual(self.file.tell(), 0)

    def test_sock_sendfile_mix_with_regular_send(self):
        buf = b"mix_regular_send" * (4 * 1024)  
        sock, proto = self.prepare_socksendfile()
        self.run_loop(self.loop.sock_sendall(sock, buf))
        ret = self.run_loop(self.loop.sock_sendfile(sock, self.file))
        self.run_loop(self.loop.sock_sendall(sock, buf))
        sock.close()
        self.run_loop(proto.wait_closed())

        self.assertEqual(ret, len(self.DATA))
        expected = buf + self.DATA + buf
        self.assertEqual(proto.data, expected)
        self.assertEqual(self.file.tell(), len(self.DATA))


class SendfileMixin(SendfileBase):

    

    def prepare_sendfile(self, *, is_ssl=False, close_after=0):
        port = socket_helper.find_unused_port()
        srv_proto = MySendfileProto(loop=self.loop,
                                    close_after=close_after)
        if is_ssl:
            if not ssl:
                self.skipTest("No ssl module")
            srv_ctx = test_utils.simple_server_sslcontext()
            cli_ctx = test_utils.simple_client_sslcontext()
        else:
            srv_ctx = None
            cli_ctx = None
        srv_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        srv_sock.bind((socket_helper.HOST, port))
        server = self.run_loop(self.loop.create_server(
            lambda: srv_proto, sock=srv_sock, ssl=srv_ctx))
        self.reduce_receive_buffer_size(srv_sock)

        if is_ssl:
            server_hostname = socket_helper.HOST
        else:
            server_hostname = None
        cli_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        cli_sock.connect((socket_helper.HOST, port))

        cli_proto = MySendfileProto(loop=self.loop)
        tr, pr = self.run_loop(self.loop.create_connection(
            lambda: cli_proto, sock=cli_sock,
            ssl=cli_ctx, server_hostname=server_hostname))
        self.reduce_send_buffer_size(cli_sock, transport=tr)

        def cleanup():
            srv_proto.transport.close()
            cli_proto.transport.close()
            self.run_loop(srv_proto.done)
            self.run_loop(cli_proto.done)

            server.close()
            self.run_loop(server.wait_closed())

        self.addCleanup(cleanup)
        return srv_proto, cli_proto

    @unittest.skipIf(sys.platform == 'win32', "UDP sockets are not supported")
    def test_sendfile_not_supported(self):
        tr, pr = self.run_loop(
            self.loop.create_datagram_endpoint(
                asyncio.DatagramProtocol,
                family=socket.AF_INET))
        try:
            with self.assertRaisesRegex(RuntimeError, "not supported"):
                self.run_loop(
                    self.loop.sendfile(tr, self.file))
            self.assertEqual(0, self.file.tell())
        finally:
            
            tr.close()

    def test_sendfile(self):
        srv_proto, cli_proto = self.prepare_sendfile()
        ret = self.run_loop(
            self.loop.sendfile(cli_proto.transport, self.file))
        cli_proto.transport.close()
        self.run_loop(srv_proto.done)
        self.assertEqual(ret, len(self.DATA))
        self.assertEqual(srv_proto.nbytes, len(self.DATA))
        self.assertEqual(srv_proto.data, self.DATA)
        self.assertEqual(self.file.tell(), len(self.DATA))

    def test_sendfile_force_fallback(self):
        srv_proto, cli_proto = self.prepare_sendfile()

        def sendfile_native(transp, file, offset, count):
            
            return base_events.BaseEventLoop._sendfile_native(
                self.loop, transp, file, offset, count)

        self.loop._sendfile_native = sendfile_native

        ret = self.run_loop(
            self.loop.sendfile(cli_proto.transport, self.file))
        cli_proto.transport.close()
        self.run_loop(srv_proto.done)
        self.assertEqual(ret, len(self.DATA))
        self.assertEqual(srv_proto.nbytes, len(self.DATA))
        self.assertEqual(srv_proto.data, self.DATA)
        self.assertEqual(self.file.tell(), len(self.DATA))

    def test_sendfile_force_unsupported_native(self):
        if sys.platform == 'win32':
            if isinstance(self.loop, asyncio.ProactorEventLoop):
                self.skipTest("Fails on proactor event loop")
        srv_proto, cli_proto = self.prepare_sendfile()

        def sendfile_native(transp, file, offset, count):
            
            return base_events.BaseEventLoop._sendfile_native(
                self.loop, transp, file, offset, count)

        self.loop._sendfile_native = sendfile_native

        with self.assertRaisesRegex(asyncio.SendfileNotAvailableError,
                                    "not supported"):
            self.run_loop(
                self.loop.sendfile(cli_proto.transport, self.file,
                                   fallback=False))

        cli_proto.transport.close()
        self.run_loop(srv_proto.done)
        self.assertEqual(srv_proto.nbytes, 0)
        self.assertEqual(self.file.tell(), 0)

    def test_sendfile_ssl(self):
        srv_proto, cli_proto = self.prepare_sendfile(is_ssl=True)
        ret = self.run_loop(
            self.loop.sendfile(cli_proto.transport, self.file))
        cli_proto.transport.close()
        self.run_loop(srv_proto.done)
        self.assertEqual(ret, len(self.DATA))
        self.assertEqual(srv_proto.nbytes, len(self.DATA))
        self.assertEqual(srv_proto.data, self.DATA)
        self.assertEqual(self.file.tell(), len(self.DATA))

    def test_sendfile_for_closing_transp(self):
        srv_proto, cli_proto = self.prepare_sendfile()
        cli_proto.transport.close()
        with self.assertRaisesRegex(RuntimeError, "is closing"):
            self.run_loop(self.loop.sendfile(cli_proto.transport, self.file))
        self.run_loop(srv_proto.done)
        self.assertEqual(srv_proto.nbytes, 0)
        self.assertEqual(self.file.tell(), 0)

    def test_sendfile_pre_and_post_data(self):
        srv_proto, cli_proto = self.prepare_sendfile()
        PREFIX = b'PREFIX__' * 1024  
        SUFFIX = b'--SUFFIX' * 1024  
        cli_proto.transport.write(PREFIX)
        ret = self.run_loop(
            self.loop.sendfile(cli_proto.transport, self.file))
        cli_proto.transport.write(SUFFIX)
        cli_proto.transport.close()
        self.run_loop(srv_proto.done)
        self.assertEqual(ret, len(self.DATA))
        self.assertEqual(srv_proto.data, PREFIX + self.DATA + SUFFIX)
        self.assertEqual(self.file.tell(), len(self.DATA))

    def test_sendfile_ssl_pre_and_post_data(self):
        srv_proto, cli_proto = self.prepare_sendfile(is_ssl=True)
        PREFIX = b'zxcvbnm' * 1024
        SUFFIX = b'0987654321' * 1024
        cli_proto.transport.write(PREFIX)
        ret = self.run_loop(
            self.loop.sendfile(cli_proto.transport, self.file))
        cli_proto.transport.write(SUFFIX)
        cli_proto.transport.close()
        self.run_loop(srv_proto.done)
        self.assertEqual(ret, len(self.DATA))
        self.assertEqual(srv_proto.data, PREFIX + self.DATA + SUFFIX)
        self.assertEqual(self.file.tell(), len(self.DATA))

    def test_sendfile_partial(self):
        srv_proto, cli_proto = self.prepare_sendfile()
        ret = self.run_loop(
            self.loop.sendfile(cli_proto.transport, self.file, 1000, 100))
        cli_proto.transport.close()
        self.run_loop(srv_proto.done)
        self.assertEqual(ret, 100)
        self.assertEqual(srv_proto.nbytes, 100)
        self.assertEqual(srv_proto.data, self.DATA[1000:1100])
        self.assertEqual(self.file.tell(), 1100)

    def test_sendfile_ssl_partial(self):
        srv_proto, cli_proto = self.prepare_sendfile(is_ssl=True)
        ret = self.run_loop(
            self.loop.sendfile(cli_proto.transport, self.file, 1000, 100))
        cli_proto.transport.close()
        self.run_loop(srv_proto.done)
        self.assertEqual(ret, 100)
        self.assertEqual(srv_proto.nbytes, 100)
        self.assertEqual(srv_proto.data, self.DATA[1000:1100])
        self.assertEqual(self.file.tell(), 1100)

    def test_sendfile_close_peer_after_receiving(self):
        srv_proto, cli_proto = self.prepare_sendfile(
            close_after=len(self.DATA))
        ret = self.run_loop(
            self.loop.sendfile(cli_proto.transport, self.file))
        cli_proto.transport.close()
        self.run_loop(srv_proto.done)
        self.assertEqual(ret, len(self.DATA))
        self.assertEqual(srv_proto.nbytes, len(self.DATA))
        self.assertEqual(srv_proto.data, self.DATA)
        self.assertEqual(self.file.tell(), len(self.DATA))

    def test_sendfile_ssl_close_peer_after_receiving(self):
        srv_proto, cli_proto = self.prepare_sendfile(
            is_ssl=True, close_after=len(self.DATA))
        ret = self.run_loop(
            self.loop.sendfile(cli_proto.transport, self.file))
        self.run_loop(srv_proto.done)
        self.assertEqual(ret, len(self.DATA))
        self.assertEqual(srv_proto.nbytes, len(self.DATA))
        self.assertEqual(srv_proto.data, self.DATA)
        self.assertEqual(self.file.tell(), len(self.DATA))

    
    
    
    
    @unittest.skipIf(sys.platform.startswith('sunos'),
                     "Doesn't work on Solaris")
    def test_sendfile_close_peer_in_the_middle_of_receiving(self):
        srv_proto, cli_proto = self.prepare_sendfile(close_after=1024)
        with self.assertRaises(ConnectionError):
            self.run_loop(
                self.loop.sendfile(cli_proto.transport, self.file))
        self.run_loop(srv_proto.done)

        self.assertTrue(1024 <= srv_proto.nbytes < len(self.DATA),
                        srv_proto.nbytes)
        if not (sys.platform == 'win32'
                and isinstance(self.loop, asyncio.ProactorEventLoop)):
            
            self.assertTrue(1024 <= self.file.tell() < len(self.DATA),
                            self.file.tell())
        self.assertTrue(cli_proto.transport.is_closing())

    def test_sendfile_fallback_close_peer_in_the_middle_of_receiving(self):

        def sendfile_native(transp, file, offset, count):
            
            return base_events.BaseEventLoop._sendfile_native(
                self.loop, transp, file, offset, count)

        self.loop._sendfile_native = sendfile_native

        srv_proto, cli_proto = self.prepare_sendfile(close_after=1024)
        with self.assertRaises(ConnectionError):
            try:
                self.run_loop(
                    self.loop.sendfile(cli_proto.transport, self.file))
            except OSError as e:
                
                
                if e.errno == errno.EPROTOTYPE and sys.platform == "darwin":
                    raise ConnectionError
                else:
                    raise

        self.run_loop(srv_proto.done)

        self.assertTrue(1024 <= srv_proto.nbytes < len(self.DATA),
                        srv_proto.nbytes)
        self.assertTrue(1024 <= self.file.tell() < len(self.DATA),
                        self.file.tell())

    @unittest.skipIf(not hasattr(os, 'sendfile'),
                     "Don't have native sendfile support")
    def test_sendfile_prevents_bare_write(self):
        srv_proto, cli_proto = self.prepare_sendfile()
        fut = self.loop.create_future()

        async def coro():
            fut.set_result(None)
            return await self.loop.sendfile(cli_proto.transport, self.file)

        t = self.loop.create_task(coro())
        self.run_loop(fut)
        with self.assertRaisesRegex(RuntimeError,
                                    "sendfile is in progress"):
            cli_proto.transport.write(b'data')
        ret = self.run_loop(t)
        self.assertEqual(ret, len(self.DATA))

    def test_sendfile_no_fallback_for_fallback_transport(self):
        transport = mock.Mock()
        transport.is_closing.side_effect = lambda: False
        transport._sendfile_compatible = constants._SendfileMode.FALLBACK
        with self.assertRaisesRegex(RuntimeError, 'fallback is disabled'):
            self.loop.run_until_complete(
                self.loop.sendfile(transport, None, fallback=False))


class SendfileTestsBase(SendfileMixin, SockSendfileMixin):
    pass


if sys.platform == 'win32':

    class SelectEventLoopTests(SendfileTestsBase,
                               test_utils.TestCase):

        def create_event_loop(self):
            return asyncio.SelectorEventLoop()

    class ProactorEventLoopTests(SendfileTestsBase,
                                 test_utils.TestCase):

        def create_event_loop(self):
            return asyncio.ProactorEventLoop()

else:
    import selectors

    if hasattr(selectors, 'KqueueSelector'):
        class KqueueEventLoopTests(SendfileTestsBase,
                                   test_utils.TestCase):

            def create_event_loop(self):
                return asyncio.SelectorEventLoop(
                    selectors.KqueueSelector())

    if hasattr(selectors, 'EpollSelector'):
        class EPollEventLoopTests(SendfileTestsBase,
                                  test_utils.TestCase):

            def create_event_loop(self):
                return asyncio.SelectorEventLoop(selectors.EpollSelector())

    if hasattr(selectors, 'PollSelector'):
        class PollEventLoopTests(SendfileTestsBase,
                                 test_utils.TestCase):

            def create_event_loop(self):
                return asyncio.SelectorEventLoop(selectors.PollSelector())

    
    class SelectEventLoopTests(SendfileTestsBase,
                               test_utils.TestCase):

        def create_event_loop(self):
            return asyncio.SelectorEventLoop(selectors.SelectSelector())


if __name__ == '__main__':
    unittest.main()