tests for selectorevents py import collections import selectors import socket import sys import unittest from asyncio import selectorevents from unittest import mock try import ssl except importerror ssl none import asyncio from asyncio selectorevents import baseselectoreventloop selectordatagramtransport selectorsockettransport selectortransport from test testasyncio import utils as testutils mockany mock any def teardownmodule asyncio seteventlooppolicynone class testbaseselectoreventloopbaseselectoreventloop def makeselfpipeself self ssock mock mock self csock mock mock self internalfds 1 def closeselfpipeself pass def listtobufferl buffer collections deque buffer extendmemoryviewi for i in l return buffer def closetransporttransport don t call transport close because the event loop and the selector are mocked if transport sock is none return transport sock close transport sock none class baseselectoreventloopteststestutils testcase def setupself super setup self selector mock mock self selector select returnvalue self loop testbaseselectoreventloopself selector self seteventloopself loop def testmakesockettransportself m mock mock self loop addreader mock mock self loop ensurefdnotransport mock mock transport self loop makesockettransportm asyncio protocol self assertisinstancetransport selectorsockettransport self assertequalself loop ensurefdnotransport callcount 1 calling repr must not fail when the event loop is closed self loop close reprtransport closetransporttransport mock patch asyncio selectorevents ssl none mock patch asyncio sslproto ssl none def testmakessltransportwithoutsslerrorself m mock mock self loop addreader mock mock self loop addwriter mock mock self loop removereader mock mock self loop removewriter mock mock self loop ensurefdnotransport mock mock with self assertraisesruntimeerror self loop makessltransportm m m m self assertequalself loop ensurefdnotransport callcount 1 def testcloseself class eventloopbaseselectoreventloop def makeselfpipeself self ssock mock mock self csock mock mock self internalfds 1 self loop eventloopself selector self seteventloopself loop ssock self loop ssock ssock fileno returnvalue 7 csock self loop csock csock fileno returnvalue 1 removereader self loop removereader mock mock self loop selector close self loop selector selector mock mock self assertfalseself loop isclosed self loop close self asserttrueself loop isclosed self assertisnoneself loop selector self assertisnoneself loop csock self assertisnoneself loop ssock selector close assertcalledwith ssock close assertcalledwith csock close assertcalledwith removereader assertcalledwith7 it should be possible to call close more than once self loop close self loop close operation blocked when the loop is closed f self loop createfuture self assertraisesruntimeerror self loop runforever self assertraisesruntimeerror self loop rununtilcomplete f fd 0 def callback pass self assertraisesruntimeerror self loop addreader fd callback self assertraisesruntimeerror self loop addwriter fd callback def testclosenoselectorself self loop removereader mock mock self loop selector close self loop selector none self loop close self assertisnoneself loop selector def testreadfromselftryagainself self loop ssock recv sideeffect blockingioerror self assertisnoneself loop readfromself def testreadfromselfexceptionself self loop ssock recv sideeffect oserror self assertraisesoserror self loop readfromself def testwritetoselftryagainself self loop csock send sideeffect blockingioerror with testutils disablelogger self assertisnoneself loop writetoself def testwritetoselfexceptionself writetoself swallows oserror self loop csock send sideeffect runtimeerror self assertraisesruntimeerror self loop writetoself mock patch socket getaddrinfo def testsockconnectresolveusingsocketparamsself mgai addr needresolution com 8080 for socktype in socket sockstream socket sockdgram with self subtestsocktype sock testutils mocknonblockingsockettypesocktype mgai sideeffect lambda args none none none none 127 0 0 1 0 con self loop createtaskself loop sockconnectsock addr self loop rununtilcompletecon mgai assertcalledwith addr0 addr1 sock family sock type sock proto 0 self loop rununtilcompletecon sock connect assertcalledwith 127 0 0 1 0 def testaddreaderself self loop selector getmap returnvalue cb lambda true self loop addreader1 cb self asserttrueself loop selector register called fd mask r w self loop selector register callargs0 self assertequal1 fd self assertequalselectors eventread mask self assertequalcb r callback self assertisnonew def testaddreaderexistingself reader mock mock writer mock mock self loop selector getmap returnvalue 1 selectors selectorkey 1 1 selectors eventwrite reader writer cb lambda true self loop addreader1 cb self asserttruereader cancel called self assertfalseself loop selector register called self asserttrueself loop selector modify called fd mask r w self loop selector modify callargs0 self assertequal1 fd self assertequalselectors eventwrite selectors eventread mask self assertequalcb r callback self assertequalwriter w def testaddreaderexistingwriterself writer mock mock self loop selector getmap returnvalue 1 selectors selectorkey 1 1 selectors eventwrite none writer cb lambda true self loop addreader1 cb self assertfalseself loop selector register called self asserttrueself loop selector modify called fd mask r w self loop selector modify callargs0 self assertequal1 fd self assertequalselectors eventwrite selectors eventread mask self assertequalcb r callback self assertequalwriter w def testremovereaderself self loop selector getmap returnvalue 1 selectors selectorkey 1 1 selectors eventread none none self assertfalseself loop removereader1 self asserttrueself loop selector unregister called def testremovereaderreadwriteself reader mock mock writer mock mock self loop selector getmap returnvalue 1 selectors selectorkey 1 1 selectors eventread selectors eventwrite reader writer self asserttrue self loop removereader1 self assertfalseself loop selector unregister called self assertequal 1 selectors eventwrite none writer self loop selector modify callargs0 def testremovereaderunknownself self loop selector getmap returnvalue self assertfalse self loop removereader1 def testaddwriterself self loop selector getmap returnvalue cb lambda true self loop addwriter1 cb self asserttrueself loop selector register called fd mask r w self loop selector register callargs0 self assertequal1 fd self assertequalselectors eventwrite mask self assertisnoner self assertequalcb w callback def testaddwriterexistingself reader mock mock writer mock mock self loop selector getmap returnvalue 1 selectors selectorkey 1 1 selectors eventread reader writer cb lambda true self loop addwriter1 cb self asserttruewriter cancel called self assertfalseself loop selector register called self asserttrueself loop selector modify called fd mask r w self loop selector modify callargs0 self assertequal1 fd self assertequalselectors eventwrite selectors eventread mask self assertequalreader r self assertequalcb w callback def testremovewriterself self loop selector getmap returnvalue 1 selectors selectorkey 1 1 selectors eventwrite none none self assertfalseself loop removewriter1 self asserttrueself loop selector unregister called def testremovewriterreadwriteself reader mock mock writer mock mock self loop selector getmap returnvalue 1 selectors selectorkey 1 1 selectors eventread selectors eventwrite reader writer self asserttrue self loop removewriter1 self assertfalseself loop selector unregister called self assertequal 1 selectors eventread reader none self loop selector modify callargs0 def testremovewriterunknownself self loop selector getmap returnvalue self assertfalse self loop removewriter1 def testprocesseventsreadself reader mock mock reader cancelled false self loop addcallback mock mock self loop processevents selectors selectorkey 1 1 selectors eventread reader none selectors eventread self asserttrueself loop addcallback called self loop addcallback assertcalledwithreader def testprocesseventsreadcancelledself reader mock mock reader cancelled true self loop removereader mock mock self loop processevents selectors selectorkey 1 1 selectors eventread reader none selectors eventread self loop removereader assertcalledwith1 def testprocesseventswriteself writer mock mock writer cancelled false self loop addcallback mock mock self loop processevents selectors selectorkey1 1 selectors eventwrite none writer selectors eventwrite self loop addcallback assertcalledwithwriter def testprocesseventswritecancelledself writer mock mock writer cancelled true self loop removewriter mock mock self loop processevents selectors selectorkey1 1 selectors eventwrite none writer selectors eventwrite self loop removewriter assertcalledwith1 def testacceptconnectionmultipleself sock mock mock sock accept returnvalue mock mock mock mock backlog 100 mock the coroutine generation for a connection to prevent warnings related to unawaited coroutines acceptconnection2 is an async function that is patched with asyncmock createtask creates a task out of coroutine returned by asyncmock so use asyncio sleep0 to ensure created tasks are complete to avoid task pending warnings mockobj mock patch object with mockobjself loop acceptconnection2 as accept2mock self loop acceptconnection mock mock sock backlogbacklog self loop rununtilcompleteasyncio sleep0 self assertequalsock accept callcount backlog class selectortransportteststestutils testcase def setupself super setup self loop self newtestloop self protocol testutils maketestprotocolasyncio protocol self sock mock mocksocket socket self sock fileno returnvalue 7 def createtransportself transport selectortransportself loop self sock self protocol none self addcleanupclosetransport transport return transport def testctorself tr self createtransport self assertistr loop self loop self assertistr sock self sock self assertistr sockfd 7 def testabortself tr self createtransport tr forceclose mock mock tr abort tr forceclose assertcalledwithnone def testcloseself tr self createtransport tr close self asserttruetr isclosing self assertequal1 self loop removereadercount7 self protocol connectionlostnone self assertequaltr connlost 1 tr close self assertequaltr connlost 1 self assertequal1 self loop removereadercount7 def testclosewritebufferself tr self createtransport tr buffer extendb data tr close self assertfalseself loop readers testutils runbrieflyself loop self assertfalseself protocol connectionlost called def testforcecloseself tr self createtransport tr buffer extendb 1 self loop addreader7 mock sentinel self loop addwriter7 mock sentinel tr forceclosenone self asserttruetr isclosing self assertequaltr buffer listtobuffer self assertfalseself loop readers self assertfalseself loop writers second close should not remove reader tr forceclosenone self assertfalseself loop readers self assertequal1 self loop removereadercount7 mock patch asyncio log logger error def testfatalerrorself mexc exc oserror tr self createtransport tr forceclose mock mock tr fatalerrorexc mexc assertnotcalled tr forceclose assertcalledwithexc mock patch asyncio log logger error def testfatalerrorcustomexceptionself mexc class myerrorexception pass exc myerror tr self createtransport tr forceclose mock mock tr fatalerrorexc mexc assertcalledwith testutils mockpattern fatal error on transportnprotocol ntransport excinfomyerror mockany mockany tr forceclose assertcalledwithexc def testconnectionlostself exc oserror tr self createtransport self assertisnotnonetr protocol self assertisnotnonetr loop tr callconnectionlostexc self protocol connectionlost assertcalledwithexc self sock close assertcalledwith self assertisnonetr sock self assertisnonetr protocol self assertisnonetr loop def testaddreaderself tr self createtransport tr buffer extendb 1 tr addreader7 mock sentinel self asserttrueself loop readers tr forceclosenone self asserttruetr isclosing self assertfalseself loop readers can not add readers after closing tr addreader7 mock sentinel self assertfalseself loop readers class selectorsockettransportteststestutils testcase def setupself super setup self loop self newtestloop self protocol testutils maketestprotocolasyncio protocol self sock mock mocksocket socket self sockfd self sock fileno returnvalue 7 def sockettransportself waiternone sendmsgfalse transport selectorsockettransportself loop self sock self protocol waiterwaiter if sendmsg transport writeready transport writesendmsg else transport writeready transport writesend self addcleanupclosetransport transport return transport def testctorself waiter self loop createfuture tr self sockettransportwaiterwaiter self loop rununtilcompletewaiter self loop assertreader7 tr readready testutils runbrieflyself loop self protocol connectionmade assertcalledwithtr def testctorwithwaiterself waiter self loop createfuture self sockettransportwaiterwaiter self loop rununtilcompletewaiter self assertisnonewaiter result def testpauseresumereadingself tr self sockettransport testutils runbrieflyself loop self assertfalsetr paused self asserttruetr isreading self loop assertreader7 tr readready tr pausereading tr pausereading self asserttruetr paused self assertfalsetr isreading self loop assertnoreader7 tr resumereading tr resumereading self assertfalsetr paused self asserttruetr isreading self loop assertreader7 tr readready tr close self assertfalsetr isreading self loop assertnoreader7 def testpausereadingconnectionmadeself tr self sockettransport self protocol connectionmade sideeffect lambda tr pausereading testutils runbrieflyself loop self assertfalsetr isreading self loop assertnoreader7 tr resumereading self asserttruetr isreading self loop assertreader7 tr readready tr close self assertfalsetr isreading self loop assertnoreader7 def testreadeofreceivederrorself transport self sockettransport transport close mock mock transport fatalerror mock mock self loop callexceptionhandler mock mock self protocol eofreceived sideeffect lookuperror self sock recv returnvalue b transport readready self protocol eofreceived assertcalledwith self asserttruetransport fatalerror called def testdatareceivederrorself transport self sockettransport transport fatalerror mock mock self loop callexceptionhandler mock mock self protocol datareceived sideeffect lookuperror self sock recv returnvalue b data transport readready self asserttruetransport fatalerror called self asserttrueself protocol datareceived called def testreadreadyself transport self sockettransport self sock recv returnvalue b data transport readready self protocol datareceived assertcalledwithb data def testreadreadyeofself transport self sockettransport transport close mock mock self sock recv returnvalue b transport readready self protocol eofreceived assertcalledwith transport close assertcalledwith def testreadreadyeofkeepopenself transport self sockettransport transport close mock mock self sock recv returnvalue b self protocol eofreceived returnvalue true transport readready self protocol eofreceived assertcalledwith self assertfalsetransport close called mock patch logging exception def testreadreadytryagainself mexc self sock recv sideeffect blockingioerror transport self sockettransport transport fatalerror mock mock transport readready self assertfalsetransport fatalerror called mock patch logging exception def testreadreadytryagaininterruptedself mexc self sock recv sideeffect interruptederror transport self sockettransport transport fatalerror mock mock transport readready self assertfalsetransport fatalerror called mock patch logging exception def testreadreadyconnresetself mexc err self sock recv sideeffect connectionreseterror transport self sockettransport transport forceclose mock mock with testutils disablelogger transport readready transport forceclose assertcalledwitherr mock patch logging exception def testreadreadyerrself mexc err self sock recv sideeffect oserror transport self sockettransport transport fatalerror mock mock transport readready transport fatalerror assertcalledwith err fatal read error on socket transport def testwriteself data b data self sock send returnvalue lendata transport self sockettransport transport writedata self sock send assertcalledwithdata def testwritebytearrayself data bytearrayb data self sock send returnvalue lendata transport self sockettransport transport writedata self sock send assertcalledwithdata self assertequaldata bytearrayb data hasn t been mutated def testwritememoryviewself data memoryviewb data self sock send returnvalue lendata transport self sockettransport transport writedata self sock send assertcalledwithdata def testwritenodataself transport self sockettransport transport buffer appendmemoryviewb data transport writeb self assertfalseself sock send called self assertequallisttobufferb data transport buffer def testwritebufferself transport self sockettransport transport buffer appendb data1 transport writeb data2 self assertfalseself sock send called self assertequallisttobufferb data1 b data2 transport buffer def testwritepartialself data b data self sock send returnvalue 2 transport self sockettransport transport writedata self loop assertwriter7 transport writeready self assertequallisttobufferb ta transport buffer def testwritepartialbytearrayself data bytearrayb data self sock send returnvalue 2 transport self sockettransport transport writedata self loop assertwriter7 transport writeready self assertequallisttobufferb ta transport buffer self assertequaldata bytearrayb data hasn t been mutated def testwritepartialmemoryviewself data memoryviewb data self sock send returnvalue 2 transport self sockettransport transport writedata self loop assertwriter7 transport writeready self assertequallisttobufferb ta transport buffer def testwritepartialnoneself data b data self sock send returnvalue 0 self sock fileno returnvalue 7 transport self sockettransport transport writedata self loop assertwriter7 transport writeready self assertequallisttobufferb data transport buffer def testwritetryagainself self sock send sideeffect blockingioerror data b data transport self sockettransport transport writedata self loop assertwriter7 transport writeready self assertequallisttobufferb data transport buffer def testwritesendmsgnodataself self sock sendmsg mock mock self sock sendmsg returnvalue 0 transport self sockettransportsendmsgtrue transport buffer appendmemoryviewb data transport writeb self assertfalseself sock sendmsg called self assertequallisttobufferb data transport buffer unittest skipunlessselectorevents hassendmsg no sendmsg def testwritelinessendmsgfullself data memoryviewb data self sock sendmsg mock mock self sock sendmsg returnvalue lendata transport self sockettransportsendmsgtrue transport writelinesdata self asserttrueself sock sendmsg called self assertfalseself loop writers unittest skipunlessselectorevents hassendmsg no sendmsg def testwritelinessendmsgpartialself data memoryviewb data self sock sendmsg mock mock self sock sendmsg returnvalue 2 transport self sockettransportsendmsgtrue transport writelinesdata self asserttrueself sock sendmsg called self asserttrueself loop writers def testwritelinessendfullself data memoryviewb data self sock send returnvalue lendata self sock send fileno returnvalue 7 transport self sockettransport transport writelinesdata self asserttrueself sock send called self assertfalseself loop writers def testwritelinessendpartialself data memoryviewb data self sock send returnvalue 2 self sock send fileno returnvalue 7 transport self sockettransport transport writelinesdata self asserttrueself sock send called self asserttrueself loop writers unittest skipunlessselectorevents hassendmsg no sendmsg def testwritesendmsgfullself data memoryviewb data self sock sendmsg mock mock self sock sendmsg returnvalue lendata transport self sockettransportsendmsgtrue transport buffer appenddata self loop addwriter7 transport writeready transport writeready self asserttrueself sock sendmsg called self assertfalseself loop writers unittest skipunlessselectorevents hassendmsg no sendmsg def testwritesendmsgpartialself data memoryviewb data self sock sendmsg mock mock sent partial data self sock sendmsg returnvalue 2 transport self sockettransportsendmsgtrue transport buffer appenddata self loop addwriter7 transport writeready transport writeready self asserttrueself sock sendmsg called self asserttrueself loop writers self assertequallisttobufferb ta transport buffer unittest skipunlessselectorevents hassendmsg no sendmsg def testwritesendmsghalfbufferself data memoryviewb data1 memoryviewb data2 self sock sendmsg mock mock sent partial data self sock sendmsg returnvalue 2 transport self sockettransportsendmsgtrue transport buffer extenddata self loop addwriter7 transport writeready transport writeready self asserttrueself sock sendmsg called self asserttrueself loop writers self assertequallisttobufferb ta1 b data2 transport buffer unittest skipunlessselectorevents hassendmsg no sendmsg def testwritesendmsgoserrorself data memoryviewb data self sock sendmsg mock mock err self sock sendmsg sideeffect oserror transport self sockettransportsendmsgtrue transport fatalerror mock mock transport buffer extenddata calls fatalerror and clears the buffer transport writeready self asserttrueself sock sendmsg called self assertfalseself loop writers self assertequallisttobuffer transport buffer transport fatalerror assertcalledwith err fatal write error on socket transport mock patch asyncio selectorevents logger def testwriteexceptionself mlog err self sock send sideeffect oserror data b data transport self sockettransport transport fatalerror mock mock transport writedata transport fatalerror assertcalledwith err fatal write error on socket transport transport connlost 1 self sock resetmock transport writedata self assertfalseself sock send called self assertequaltransport connlost 2 transport writedata transport writedata transport writedata transport writedata mlog warning assertcalledwith socket send raised exception def testwritestrself transport self sockettransport self assertraisestypeerror transport write str def testwriteclosingself transport self sockettransport transport close self assertequaltransport connlost 1 transport writeb data self assertequaltransport connlost 2 def testwritereadyself data b data self sock send returnvalue lendata transport self sockettransport transport buffer appenddata self loop addwriter7 transport writeready transport writeready self asserttrueself sock send called self assertfalseself loop writers def testwritereadyclosingself data memoryviewb data self sock send returnvalue lendata transport self sockettransport transport closing true transport buffer appenddata self loop addwriter7 transport writeready transport writeready self asserttrueself sock send called self assertfalseself loop writers self sock close assertcalledwith self protocol connectionlost assertcalledwithnone unittest skipifsys flags optimize assertions are disabled in optimized mode def testwritereadynodataself transport self sockettransport this is an internal error self assertraisesassertionerror transport writeready def testwritereadypartialself data memoryviewb data self sock send returnvalue 2 transport self sockettransport transport buffer appenddata self loop addwriter7 transport writeready transport writeready self loop assertwriter7 transport writeready self assertequallisttobufferb ta transport buffer def testwritereadypartialnoneself data b data self sock send returnvalue 0 transport self sockettransport transport buffer appenddata self loop addwriter7 transport writeready transport writeready self loop assertwriter7 transport writeready self assertequallisttobufferb data transport buffer def testwritereadytryagainself self sock send sideeffect blockingioerror transport self sockettransport buffer listtobufferb data1 b data2 transport buffer buffer self loop addwriter7 transport writeready transport writeready self loop assertwriter7 transport writeready self assertequalbuffer transport buffer def testwritereadyexceptionself err self sock send sideeffect oserror transport self sockettransport transport fatalerror mock mock transport buffer extendb data transport writeready transport fatalerror assertcalledwith err fatal write error on socket transport def testwriteeofself tr self sockettransport self asserttruetr canwriteeof tr writeeof self sock shutdown assertcalledwithsocket shutwr tr writeeof self assertequalself sock shutdown callcount 1 tr close def testwriteeofbufferself tr self sockettransport self sock send sideeffect blockingioerror tr writeb data tr writeeof self assertequaltr buffer listtobufferb data self asserttruetr eof self assertfalseself sock shutdown called self sock send sideeffect lambda 4 tr writeready self asserttrueself sock send called self sock shutdown assertcalledwithsocket shutwr tr close def testwriteeofaftercloseself tr self sockettransport tr close self loop rununtilcompleteasyncio sleep0 tr writeeof mock patch asyncio baseevents logger def testtransportcloseremovewriterself mlog removewriter self loop removewriter mock mock transport self sockettransport transport close removewriter assertcalledwithself sockfd class selectorsockettransportbufferedprotocolteststestutils testcase def setupself super setup self loop self newtestloop self protocol testutils maketestprotocolasyncio bufferedprotocol self buf bytearray1 self protocol getbuffer sideeffect lambda hint self buf self sock mock mocksocket socket self sockfd self sock fileno returnvalue 7 def sockettransportself waiternone transport selectorsockettransportself loop self sock self protocol waiterwaiter self addcleanupclosetransport transport return transport def testctorself waiter self loop createfuture tr self sockettransportwaiterwaiter self loop rununtilcompletewaiter self loop assertreader7 tr readready testutils runbrieflyself loop self protocol connectionmade assertcalledwithtr def testgetbuffererrorself transport self sockettransport transport fatalerror mock mock self loop callexceptionhandler mock mock self protocol getbuffer sideeffect lookuperror transport readready self asserttruetransport fatalerror called self asserttrueself protocol getbuffer called self assertfalseself protocol bufferupdated called def testgetbufferzerosizedself transport self sockettransport transport fatalerror mock mock self loop callexceptionhandler mock mock self protocol getbuffer sideeffect lambda hint bytearray0 transport readready self asserttruetransport fatalerror called self asserttrueself protocol getbuffer called self assertfalseself protocol bufferupdated called def testprototypeswitchself self protocol testutils maketestprotocolasyncio protocol transport self sockettransport self sock recv returnvalue b data transport readready self protocol datareceived assertcalledwithb data switch protocol to a bufferedprotocol bufproto testutils maketestprotocolasyncio bufferedprotocol buf bytearray4 bufproto getbuffer sideeffect lambda hint buf transport setprotocolbufproto self sock recvinto returnvalue 10 transport readready bufproto getbuffer assertcalledwith1 bufproto bufferupdated assertcalledwith10 def testbufferupdatederrorself transport self sockettransport transport fatalerror mock mock self loop callexceptionhandler mock mock self protocol bufferupdated sideeffect lookuperror self sock recvinto returnvalue 10 transport readready self asserttruetransport fatalerror called self asserttrueself protocol getbuffer called self asserttrueself protocol bufferupdated called def testreadeofreceivederrorself transport self sockettransport transport close mock mock transport fatalerror mock mock self loop callexceptionhandler mock mock self protocol eofreceived sideeffect lookuperror self sock recvinto returnvalue 0 transport readready self protocol eofreceived assertcalledwith self asserttruetransport fatalerror called def testreadreadyself transport self sockettransport self sock recvinto returnvalue 10 transport readready self protocol getbuffer assertcalledwith1 self protocol bufferupdated assertcalledwith10 def testreadreadyeofself transport self sockettransport transport close mock mock self sock recvinto returnvalue 0 transport readready self protocol eofreceived assertcalledwith transport close assertcalledwith def testreadreadyeofkeepopenself transport self sockettransport transport close mock mock self sock recvinto returnvalue 0 self protocol eofreceived returnvalue true transport readready self protocol eofreceived assertcalledwith self assertfalsetransport close called mock patch logging exception def testreadreadytryagainself mexc self sock recvinto sideeffect blockingioerror transport self sockettransport transport fatalerror mock mock transport readready self assertfalsetransport fatalerror called mock patch logging exception def testreadreadytryagaininterruptedself mexc self sock recvinto sideeffect interruptederror transport self sockettransport transport fatalerror mock mock transport readready self assertfalsetransport fatalerror called mock patch logging exception def testreadreadyconnresetself mexc err self sock recvinto sideeffect connectionreseterror transport self sockettransport transport forceclose mock mock with testutils disablelogger transport readready transport forceclose assertcalledwitherr mock patch logging exception def testreadreadyerrself mexc err self sock recvinto sideeffect oserror transport self sockettransport transport fatalerror mock mock transport readready transport fatalerror assertcalledwith err fatal read error on socket transport class selectordatagramtransportteststestutils testcase def setupself super setup self loop self newtestloop self protocol testutils maketestprotocolasyncio datagramprotocol self sock mock mockspecsetsocket socket self sock fileno returnvalue 7 def datagramtransportself addressnone self sock getpeername sideeffect none if address else oserror transport selectordatagramtransportself loop self sock self protocol addressaddress self addcleanupclosetransport transport return transport def testreadreadyself transport self datagramtransport self sock recvfrom returnvalue b data 0 0 0 0 1234 transport readready self protocol datagramreceived assertcalledwith b data 0 0 0 0 1234 def testtransportinheritanceself transport self datagramtransport self assertisinstancetransport asyncio datagramtransport def testreadreadytryagainself transport self datagramtransport self sock recvfrom sideeffect blockingioerror transport fatalerror mock mock transport readready self assertfalsetransport fatalerror called def testreadreadyerrself transport self datagramtransport err self sock recvfrom sideeffect runtimeerror transport fatalerror mock mock transport readready transport fatalerror assertcalledwith err fatal read error on datagram transport def testreadreadyoserrself transport self datagramtransport err self sock recvfrom sideeffect oserror transport fatalerror mock mock transport readready self assertfalsetransport fatalerror called self protocol errorreceived assertcalledwitherr def testsendtoself data b data transport self datagramtransport transport sendtodata 0 0 0 0 1234 self asserttrueself sock sendto called self assertequal self sock sendto callargs0 data 0 0 0 0 1234 def testsendtobytearrayself data bytearrayb data transport self datagramtransport transport sendtodata 0 0 0 0 1234 self asserttrueself sock sendto called self assertequal self sock sendto callargs0 data 0 0 0 0 1234 def testsendtomemoryviewself data memoryviewb data transport self datagramtransport transport sendtodata 0 0 0 0 1234 self asserttrueself sock sendto called self assertequal self sock sendto callargs0 data 0 0 0 0 1234 def testsendtonodataself transport self datagramtransport transport buffer appendb data 0 0 0 0 12345 transport sendtob self assertfalseself sock sendto called self assertequal b data 0 0 0 0 12345 listtransport buffer def testsendtobufferself transport self datagramtransport transport buffer appendb data1 0 0 0 0 12345 transport sendtob data2 0 0 0 0 12345 self assertfalseself sock sendto called self assertequal b data1 0 0 0 0 12345 b data2 0 0 0 0 12345 listtransport buffer def testsendtobufferbytearrayself data2 bytearrayb data2 transport self datagramtransport transport buffer appendb data1 0 0 0 0 12345 transport sendtodata2 0 0 0 0 12345 self assertfalseself sock sendto called self assertequal b data1 0 0 0 0 12345 b data2 0 0 0 0 12345 listtransport buffer self assertisinstancetransport buffer10 bytes def testsendtobuffermemoryviewself data2 memoryviewb data2 transport self datagramtransport transport buffer appendb data1 0 0 0 0 12345 transport sendtodata2 0 0 0 0 12345 self assertfalseself sock sendto called self assertequal b data1 0 0 0 0 12345 b data2 0 0 0 0 12345 listtransport buffer self assertisinstancetransport buffer10 bytes def testsendtotryagainself data b data self sock sendto sideeffect blockingioerror transport self datagramtransport transport sendtodata 0 0 0 0 12345 self loop assertwriter7 transport sendtoready self assertequal b data 0 0 0 0 12345 listtransport buffer mock patch asyncio selectorevents logger def testsendtoexceptionself mlog data b data err self sock sendto sideeffect runtimeerror transport self datagramtransport transport fatalerror mock mock transport sendtodata self asserttruetransport fatalerror called transport fatalerror assertcalledwith err fatal write error on datagram transport transport connlost 1 transport address 123 transport sendtodata transport sendtodata transport sendtodata transport sendtodata transport sendtodata mlog warning assertcalledwith socket send raised exception def testsendtoerrorreceivedself data b data self sock sendto sideeffect connectionrefusederror transport self datagramtransport transport fatalerror mock mock transport sendtodata self assertequaltransport connlost 0 self assertfalsetransport fatalerror called def testsendtoerrorreceivedconnectedself data b data self sock send sideeffect connectionrefusederror transport self datagramtransportaddress 0 0 0 0 1 transport fatalerror mock mock transport sendtodata self assertfalsetransport fatalerror called self asserttrueself protocol errorreceived called def testsendtostrself transport self datagramtransport self assertraisestypeerror transport sendto str def testsendtoconnectedaddrself transport self datagramtransportaddress 0 0 0 0 1 self assertraises valueerror transport sendto b str 0 0 0 0 2 def testsendtoclosingself transport self datagramtransportaddress1 transport close self assertequaltransport connlost 1 transport sendtob data 1 self assertequaltransport connlost 2 def testsendtoreadyself data b data self sock sendto returnvalue lendata transport self datagramtransport transport buffer appenddata 0 0 0 0 12345 self loop addwriter7 transport sendtoready transport sendtoready self asserttrueself sock sendto called self assertequal self sock sendto callargs0 data 0 0 0 0 12345 self assertfalseself loop writers def testsendtoreadyclosingself data b data self sock send returnvalue lendata transport self datagramtransport transport closing true transport buffer appenddata self loop addwriter7 transport sendtoready transport sendtoready self sock sendto assertcalledwithdata self assertfalseself loop writers self sock close assertcalledwith self protocol connectionlost assertcalledwithnone def testsendtoreadynodataself transport self datagramtransport self loop addwriter7 transport sendtoready transport sendtoready self assertfalseself sock sendto called self assertfalseself loop writers def testsendtoreadytryagainself self sock sendto sideeffect blockingioerror transport self datagramtransport transport buffer extendb data1 b data2 self loop addwriter7 transport sendtoready transport sendtoready self loop assertwriter7 transport sendtoready self assertequal b data1 b data2 listtransport buffer def testsendtoreadyexceptionself err self sock sendto sideeffect runtimeerror transport self datagramtransport transport fatalerror mock mock transport buffer appendb data transport sendtoready transport fatalerror assertcalledwith err fatal write error on datagram transport def testsendtoreadyerrorreceivedself self sock sendto sideeffect connectionrefusederror transport self datagramtransport transport fatalerror mock mock transport buffer appendb data transport sendtoready self assertfalsetransport fatalerror called def testsendtoreadyerrorreceivedconnectionself self sock send sideeffect connectionrefusederror transport self datagramtransportaddress 0 0 0 0 1 transport fatalerror mock mock transport buffer appendb data transport sendtoready self assertfalsetransport fatalerror called self asserttrueself protocol errorreceived called mock patch asyncio baseevents logger error def testfatalerrorconnectedself mexc transport self datagramtransportaddress 0 0 0 0 1 err connectionrefusederror transport fatalerrorerr self assertfalseself protocol errorreceived called mexc assertnotcalled mock patch asyncio baseevents logger error def testfatalerrorconnectedcustomerrorself mexc class myexceptionexception pass transport self datagramtransportaddress 0 0 0 0 1 err myexception transport fatalerrorerr self assertfalseself protocol errorreceived called mexc assertcalledwith testutils mockpattern fatal error on transportnprotocol ntransport excinfomyexception mockany mockany if name main unittest main don t call transport close because the event loop and the selector are mocked calling repr must not fail when the event loop is closed it should be possible to call close more than once operation blocked when the loop is closed _write_to_self swallows oserror mock the coroutine generation for a connection to prevent warnings related to un awaited coroutines _accept_connection2 is an async function that is patched with asyncmock create_task creates a task out of coroutine returned by asyncmock so use asyncio sleep 0 to ensure created tasks are complete to avoid task pending warnings second close should not remove reader can not add readers after closing hasn t been mutated hasn t been mutated sent partial data sent partial data calls _fatal_error and clears the buffer this is an internal error switch protocol to a bufferedprotocol
import collections
import selectors
import socket
import sys
import unittest
from asyncio import selector_events
from unittest import mock

try:
    import ssl
except ImportError:
    ssl = None

import asyncio
from asyncio.selector_events import (BaseSelectorEventLoop,
                                     _SelectorDatagramTransport,
                                     _SelectorSocketTransport,
                                     _SelectorTransport)
from test.test_asyncio import utils as test_utils

MOCK_ANY = mock.ANY


def tearDownModule():
    asyncio.set_event_loop_policy(None)


class TestBaseSelectorEventLoop(BaseSelectorEventLoop):

    def _make_self_pipe(self):
        self._ssock = mock.Mock()
        self._csock = mock.Mock()
        self._internal_fds += 1

    def _close_self_pipe(self):
        pass


def list_to_buffer(l=()):
    buffer = collections.deque()
    buffer.extend((memoryview(i) for i in l))
    return buffer



def close_transport(transport):
    
    
    if transport._sock is None:
        return
    transport._sock.close()
    transport._sock = None


class BaseSelectorEventLoopTests(test_utils.TestCase):

    def setUp(self):
        super().setUp()
        self.selector = mock.Mock()
        self.selector.select.return_value = []
        self.loop = TestBaseSelectorEventLoop(self.selector)
        self.set_event_loop(self.loop)

    def test_make_socket_transport(self):
        m = mock.Mock()
        self.loop.add_reader = mock.Mock()
        self.loop._ensure_fd_no_transport = mock.Mock()
        transport = self.loop._make_socket_transport(m, asyncio.Protocol())
        self.assertIsInstance(transport, _SelectorSocketTransport)
        self.assertEqual(self.loop._ensure_fd_no_transport.call_count, 1)

        
        self.loop.close()
        repr(transport)

        close_transport(transport)

    @mock.patch('asyncio.selector_events.ssl', None)
    @mock.patch('asyncio.sslproto.ssl', None)
    def test_make_ssl_transport_without_ssl_error(self):
        m = mock.Mock()
        self.loop.add_reader = mock.Mock()
        self.loop.add_writer = mock.Mock()
        self.loop.remove_reader = mock.Mock()
        self.loop.remove_writer = mock.Mock()
        self.loop._ensure_fd_no_transport = mock.Mock()
        with self.assertRaises(RuntimeError):
            self.loop._make_ssl_transport(m, m, m, m)
        self.assertEqual(self.loop._ensure_fd_no_transport.call_count, 1)

    def test_close(self):
        class EventLoop(BaseSelectorEventLoop):
            def _make_self_pipe(self):
                self._ssock = mock.Mock()
                self._csock = mock.Mock()
                self._internal_fds += 1

        self.loop = EventLoop(self.selector)
        self.set_event_loop(self.loop)

        ssock = self.loop._ssock
        ssock.fileno.return_value = 7
        csock = self.loop._csock
        csock.fileno.return_value = 1
        remove_reader = self.loop._remove_reader = mock.Mock()

        self.loop._selector.close()
        self.loop._selector = selector = mock.Mock()
        self.assertFalse(self.loop.is_closed())

        self.loop.close()
        self.assertTrue(self.loop.is_closed())
        self.assertIsNone(self.loop._selector)
        self.assertIsNone(self.loop._csock)
        self.assertIsNone(self.loop._ssock)
        selector.close.assert_called_with()
        ssock.close.assert_called_with()
        csock.close.assert_called_with()
        remove_reader.assert_called_with(7)

        
        self.loop.close()
        self.loop.close()

        
        f = self.loop.create_future()
        self.assertRaises(RuntimeError, self.loop.run_forever)
        self.assertRaises(RuntimeError, self.loop.run_until_complete, f)
        fd = 0
        def callback():
            pass
        self.assertRaises(RuntimeError, self.loop.add_reader, fd, callback)
        self.assertRaises(RuntimeError, self.loop.add_writer, fd, callback)

    def test_close_no_selector(self):
        self.loop.remove_reader = mock.Mock()
        self.loop._selector.close()
        self.loop._selector = None
        self.loop.close()
        self.assertIsNone(self.loop._selector)

    def test_read_from_self_tryagain(self):
        self.loop._ssock.recv.side_effect = BlockingIOError
        self.assertIsNone(self.loop._read_from_self())

    def test_read_from_self_exception(self):
        self.loop._ssock.recv.side_effect = OSError
        self.assertRaises(OSError, self.loop._read_from_self)

    def test_write_to_self_tryagain(self):
        self.loop._csock.send.side_effect = BlockingIOError
        with test_utils.disable_logger():
            self.assertIsNone(self.loop._write_to_self())

    def test_write_to_self_exception(self):
        
        self.loop._csock.send.side_effect = RuntimeError()
        self.assertRaises(RuntimeError, self.loop._write_to_self)

    @mock.patch('socket.getaddrinfo')
    def test_sock_connect_resolve_using_socket_params(self, m_gai):
        addr = ('need-resolution.com', 8080)
        for sock_type in [socket.SOCK_STREAM, socket.SOCK_DGRAM]:
            with self.subTest(sock_type):
                sock = test_utils.mock_nonblocking_socket(type=sock_type)

                m_gai.side_effect = \
                    lambda *args: [(None, None, None, None, ('127.0.0.1', 0))]

                con = self.loop.create_task(self.loop.sock_connect(sock, addr))
                self.loop.run_until_complete(con)
                m_gai.assert_called_with(
                    addr[0], addr[1], sock.family, sock.type, sock.proto, 0)

                self.loop.run_until_complete(con)
                sock.connect.assert_called_with(('127.0.0.1', 0))

    def test_add_reader(self):
        self.loop._selector.get_map.return_value = {}
        cb = lambda: True
        self.loop.add_reader(1, cb)

        self.assertTrue(self.loop._selector.register.called)
        fd, mask, (r, w) = self.loop._selector.register.call_args[0]
        self.assertEqual(1, fd)
        self.assertEqual(selectors.EVENT_READ, mask)
        self.assertEqual(cb, r._callback)
        self.assertIsNone(w)

    def test_add_reader_existing(self):
        reader = mock.Mock()
        writer = mock.Mock()
        self.loop._selector.get_map.return_value = {1: selectors.SelectorKey(
            1, 1, selectors.EVENT_WRITE, (reader, writer))}
        cb = lambda: True
        self.loop.add_reader(1, cb)

        self.assertTrue(reader.cancel.called)
        self.assertFalse(self.loop._selector.register.called)
        self.assertTrue(self.loop._selector.modify.called)
        fd, mask, (r, w) = self.loop._selector.modify.call_args[0]
        self.assertEqual(1, fd)
        self.assertEqual(selectors.EVENT_WRITE | selectors.EVENT_READ, mask)
        self.assertEqual(cb, r._callback)
        self.assertEqual(writer, w)

    def test_add_reader_existing_writer(self):
        writer = mock.Mock()
        self.loop._selector.get_map.return_value = {1: selectors.SelectorKey(
            1, 1, selectors.EVENT_WRITE, (None, writer))}
        cb = lambda: True
        self.loop.add_reader(1, cb)

        self.assertFalse(self.loop._selector.register.called)
        self.assertTrue(self.loop._selector.modify.called)
        fd, mask, (r, w) = self.loop._selector.modify.call_args[0]
        self.assertEqual(1, fd)
        self.assertEqual(selectors.EVENT_WRITE | selectors.EVENT_READ, mask)
        self.assertEqual(cb, r._callback)
        self.assertEqual(writer, w)

    def test_remove_reader(self):
        self.loop._selector.get_map.return_value = {1: selectors.SelectorKey(
            1, 1, selectors.EVENT_READ, (None, None))}
        self.assertFalse(self.loop.remove_reader(1))

        self.assertTrue(self.loop._selector.unregister.called)

    def test_remove_reader_read_write(self):
        reader = mock.Mock()
        writer = mock.Mock()
        self.loop._selector.get_map.return_value = {1: selectors.SelectorKey(
            1, 1, selectors.EVENT_READ | selectors.EVENT_WRITE,
            (reader, writer))}
        self.assertTrue(
            self.loop.remove_reader(1))

        self.assertFalse(self.loop._selector.unregister.called)
        self.assertEqual(
            (1, selectors.EVENT_WRITE, (None, writer)),
            self.loop._selector.modify.call_args[0])

    def test_remove_reader_unknown(self):
        self.loop._selector.get_map.return_value = {}
        self.assertFalse(
            self.loop.remove_reader(1))

    def test_add_writer(self):
        self.loop._selector.get_map.return_value = {}
        cb = lambda: True
        self.loop.add_writer(1, cb)

        self.assertTrue(self.loop._selector.register.called)
        fd, mask, (r, w) = self.loop._selector.register.call_args[0]
        self.assertEqual(1, fd)
        self.assertEqual(selectors.EVENT_WRITE, mask)
        self.assertIsNone(r)
        self.assertEqual(cb, w._callback)

    def test_add_writer_existing(self):
        reader = mock.Mock()
        writer = mock.Mock()
        self.loop._selector.get_map.return_value = {1: selectors.SelectorKey(
            1, 1, selectors.EVENT_READ, (reader, writer))}
        cb = lambda: True
        self.loop.add_writer(1, cb)

        self.assertTrue(writer.cancel.called)
        self.assertFalse(self.loop._selector.register.called)
        self.assertTrue(self.loop._selector.modify.called)
        fd, mask, (r, w) = self.loop._selector.modify.call_args[0]
        self.assertEqual(1, fd)
        self.assertEqual(selectors.EVENT_WRITE | selectors.EVENT_READ, mask)
        self.assertEqual(reader, r)
        self.assertEqual(cb, w._callback)

    def test_remove_writer(self):
        self.loop._selector.get_map.return_value = {1: selectors.SelectorKey(
            1, 1, selectors.EVENT_WRITE, (None, None))}
        self.assertFalse(self.loop.remove_writer(1))

        self.assertTrue(self.loop._selector.unregister.called)

    def test_remove_writer_read_write(self):
        reader = mock.Mock()
        writer = mock.Mock()
        self.loop._selector.get_map.return_value = {1: selectors.SelectorKey(
            1, 1, selectors.EVENT_READ | selectors.EVENT_WRITE,
            (reader, writer))}
        self.assertTrue(
            self.loop.remove_writer(1))

        self.assertFalse(self.loop._selector.unregister.called)
        self.assertEqual(
            (1, selectors.EVENT_READ, (reader, None)),
            self.loop._selector.modify.call_args[0])

    def test_remove_writer_unknown(self):
        self.loop._selector.get_map.return_value = {}
        self.assertFalse(
            self.loop.remove_writer(1))

    def test_process_events_read(self):
        reader = mock.Mock()
        reader._cancelled = False

        self.loop._add_callback = mock.Mock()
        self.loop._process_events(
            [(selectors.SelectorKey(
                1, 1, selectors.EVENT_READ, (reader, None)),
              selectors.EVENT_READ)])
        self.assertTrue(self.loop._add_callback.called)
        self.loop._add_callback.assert_called_with(reader)

    def test_process_events_read_cancelled(self):
        reader = mock.Mock()
        reader.cancelled = True

        self.loop._remove_reader = mock.Mock()
        self.loop._process_events(
            [(selectors.SelectorKey(
                1, 1, selectors.EVENT_READ, (reader, None)),
             selectors.EVENT_READ)])
        self.loop._remove_reader.assert_called_with(1)

    def test_process_events_write(self):
        writer = mock.Mock()
        writer._cancelled = False

        self.loop._add_callback = mock.Mock()
        self.loop._process_events(
            [(selectors.SelectorKey(1, 1, selectors.EVENT_WRITE,
                                    (None, writer)),
              selectors.EVENT_WRITE)])
        self.loop._add_callback.assert_called_with(writer)

    def test_process_events_write_cancelled(self):
        writer = mock.Mock()
        writer.cancelled = True
        self.loop._remove_writer = mock.Mock()

        self.loop._process_events(
            [(selectors.SelectorKey(1, 1, selectors.EVENT_WRITE,
                                    (None, writer)),
              selectors.EVENT_WRITE)])
        self.loop._remove_writer.assert_called_with(1)

    def test_accept_connection_multiple(self):
        sock = mock.Mock()
        sock.accept.return_value = (mock.Mock(), mock.Mock())
        backlog = 100
        
        
        
        
        
        
        mock_obj = mock.patch.object
        with mock_obj(self.loop, '_accept_connection2') as accept2_mock:
            self.loop._accept_connection(
                mock.Mock(), sock, backlog=backlog)
        self.loop.run_until_complete(asyncio.sleep(0))
        self.assertEqual(sock.accept.call_count, backlog)


class SelectorTransportTests(test_utils.TestCase):

    def setUp(self):
        super().setUp()
        self.loop = self.new_test_loop()
        self.protocol = test_utils.make_test_protocol(asyncio.Protocol)
        self.sock = mock.Mock(socket.socket)
        self.sock.fileno.return_value = 7

    def create_transport(self):
        transport = _SelectorTransport(self.loop, self.sock, self.protocol,
                                       None)
        self.addCleanup(close_transport, transport)
        return transport

    def test_ctor(self):
        tr = self.create_transport()
        self.assertIs(tr._loop, self.loop)
        self.assertIs(tr._sock, self.sock)
        self.assertIs(tr._sock_fd, 7)

    def test_abort(self):
        tr = self.create_transport()
        tr._force_close = mock.Mock()

        tr.abort()
        tr._force_close.assert_called_with(None)

    def test_close(self):
        tr = self.create_transport()
        tr.close()

        self.assertTrue(tr.is_closing())
        self.assertEqual(1, self.loop.remove_reader_count[7])
        self.protocol.connection_lost(None)
        self.assertEqual(tr._conn_lost, 1)

        tr.close()
        self.assertEqual(tr._conn_lost, 1)
        self.assertEqual(1, self.loop.remove_reader_count[7])

    def test_close_write_buffer(self):
        tr = self.create_transport()
        tr._buffer.extend(b'data')
        tr.close()

        self.assertFalse(self.loop.readers)
        test_utils.run_briefly(self.loop)
        self.assertFalse(self.protocol.connection_lost.called)

    def test_force_close(self):
        tr = self.create_transport()
        tr._buffer.extend(b'1')
        self.loop._add_reader(7, mock.sentinel)
        self.loop._add_writer(7, mock.sentinel)
        tr._force_close(None)

        self.assertTrue(tr.is_closing())
        self.assertEqual(tr._buffer, list_to_buffer())
        self.assertFalse(self.loop.readers)
        self.assertFalse(self.loop.writers)

        
        tr._force_close(None)
        self.assertFalse(self.loop.readers)
        self.assertEqual(1, self.loop.remove_reader_count[7])

    @mock.patch('asyncio.log.logger.error')
    def test_fatal_error(self, m_exc):
        exc = OSError()
        tr = self.create_transport()
        tr._force_close = mock.Mock()
        tr._fatal_error(exc)

        m_exc.assert_not_called()

        tr._force_close.assert_called_with(exc)

    @mock.patch('asyncio.log.logger.error')
    def test_fatal_error_custom_exception(self, m_exc):
        class MyError(Exception):
            pass
        exc = MyError()
        tr = self.create_transport()
        tr._force_close = mock.Mock()
        tr._fatal_error(exc)

        m_exc.assert_called_with(
            test_utils.MockPattern(
                'Fatal error on transport\nprotocol:.*\ntransport:.*'),
            exc_info=(MyError, MOCK_ANY, MOCK_ANY))

        tr._force_close.assert_called_with(exc)

    def test_connection_lost(self):
        exc = OSError()
        tr = self.create_transport()
        self.assertIsNotNone(tr._protocol)
        self.assertIsNotNone(tr._loop)
        tr._call_connection_lost(exc)

        self.protocol.connection_lost.assert_called_with(exc)
        self.sock.close.assert_called_with()
        self.assertIsNone(tr._sock)

        self.assertIsNone(tr._protocol)
        self.assertIsNone(tr._loop)

    def test__add_reader(self):
        tr = self.create_transport()
        tr._buffer.extend(b'1')
        tr._add_reader(7, mock.sentinel)
        self.assertTrue(self.loop.readers)

        tr._force_close(None)

        self.assertTrue(tr.is_closing())
        self.assertFalse(self.loop.readers)

        
        tr._add_reader(7, mock.sentinel)
        self.assertFalse(self.loop.readers)


class SelectorSocketTransportTests(test_utils.TestCase):

    def setUp(self):
        super().setUp()
        self.loop = self.new_test_loop()
        self.protocol = test_utils.make_test_protocol(asyncio.Protocol)
        self.sock = mock.Mock(socket.socket)
        self.sock_fd = self.sock.fileno.return_value = 7

    def socket_transport(self, waiter=None, sendmsg=False):
        transport = _SelectorSocketTransport(self.loop, self.sock,
                                             self.protocol, waiter=waiter)
        if sendmsg:
            transport._write_ready = transport._write_sendmsg
        else:
            transport._write_ready = transport._write_send
        self.addCleanup(close_transport, transport)
        return transport

    def test_ctor(self):
        waiter = self.loop.create_future()
        tr = self.socket_transport(waiter=waiter)
        self.loop.run_until_complete(waiter)

        self.loop.assert_reader(7, tr._read_ready)
        test_utils.run_briefly(self.loop)
        self.protocol.connection_made.assert_called_with(tr)

    def test_ctor_with_waiter(self):
        waiter = self.loop.create_future()
        self.socket_transport(waiter=waiter)
        self.loop.run_until_complete(waiter)

        self.assertIsNone(waiter.result())

    def test_pause_resume_reading(self):
        tr = self.socket_transport()
        test_utils.run_briefly(self.loop)
        self.assertFalse(tr._paused)
        self.assertTrue(tr.is_reading())
        self.loop.assert_reader(7, tr._read_ready)

        tr.pause_reading()
        tr.pause_reading()
        self.assertTrue(tr._paused)
        self.assertFalse(tr.is_reading())
        self.loop.assert_no_reader(7)

        tr.resume_reading()
        tr.resume_reading()
        self.assertFalse(tr._paused)
        self.assertTrue(tr.is_reading())
        self.loop.assert_reader(7, tr._read_ready)

        tr.close()
        self.assertFalse(tr.is_reading())
        self.loop.assert_no_reader(7)

    def test_pause_reading_connection_made(self):
        tr = self.socket_transport()
        self.protocol.connection_made.side_effect = lambda _: tr.pause_reading()
        test_utils.run_briefly(self.loop)
        self.assertFalse(tr.is_reading())
        self.loop.assert_no_reader(7)

        tr.resume_reading()
        self.assertTrue(tr.is_reading())
        self.loop.assert_reader(7, tr._read_ready)

        tr.close()
        self.assertFalse(tr.is_reading())
        self.loop.assert_no_reader(7)


    def test_read_eof_received_error(self):
        transport = self.socket_transport()
        transport.close = mock.Mock()
        transport._fatal_error = mock.Mock()

        self.loop.call_exception_handler = mock.Mock()

        self.protocol.eof_received.side_effect = LookupError()

        self.sock.recv.return_value = b''
        transport._read_ready()

        self.protocol.eof_received.assert_called_with()
        self.assertTrue(transport._fatal_error.called)

    def test_data_received_error(self):
        transport = self.socket_transport()
        transport._fatal_error = mock.Mock()

        self.loop.call_exception_handler = mock.Mock()
        self.protocol.data_received.side_effect = LookupError()

        self.sock.recv.return_value = b'data'
        transport._read_ready()

        self.assertTrue(transport._fatal_error.called)
        self.assertTrue(self.protocol.data_received.called)

    def test_read_ready(self):
        transport = self.socket_transport()

        self.sock.recv.return_value = b'data'
        transport._read_ready()

        self.protocol.data_received.assert_called_with(b'data')

    def test_read_ready_eof(self):
        transport = self.socket_transport()
        transport.close = mock.Mock()

        self.sock.recv.return_value = b''
        transport._read_ready()

        self.protocol.eof_received.assert_called_with()
        transport.close.assert_called_with()

    def test_read_ready_eof_keep_open(self):
        transport = self.socket_transport()
        transport.close = mock.Mock()

        self.sock.recv.return_value = b''
        self.protocol.eof_received.return_value = True
        transport._read_ready()

        self.protocol.eof_received.assert_called_with()
        self.assertFalse(transport.close.called)

    @mock.patch('logging.exception')
    def test_read_ready_tryagain(self, m_exc):
        self.sock.recv.side_effect = BlockingIOError

        transport = self.socket_transport()
        transport._fatal_error = mock.Mock()
        transport._read_ready()

        self.assertFalse(transport._fatal_error.called)

    @mock.patch('logging.exception')
    def test_read_ready_tryagain_interrupted(self, m_exc):
        self.sock.recv.side_effect = InterruptedError

        transport = self.socket_transport()
        transport._fatal_error = mock.Mock()
        transport._read_ready()

        self.assertFalse(transport._fatal_error.called)

    @mock.patch('logging.exception')
    def test_read_ready_conn_reset(self, m_exc):
        err = self.sock.recv.side_effect = ConnectionResetError()

        transport = self.socket_transport()
        transport._force_close = mock.Mock()
        with test_utils.disable_logger():
            transport._read_ready()
        transport._force_close.assert_called_with(err)

    @mock.patch('logging.exception')
    def test_read_ready_err(self, m_exc):
        err = self.sock.recv.side_effect = OSError()

        transport = self.socket_transport()
        transport._fatal_error = mock.Mock()
        transport._read_ready()

        transport._fatal_error.assert_called_with(
                                   err,
                                   'Fatal read error on socket transport')

    def test_write(self):
        data = b'data'
        self.sock.send.return_value = len(data)

        transport = self.socket_transport()
        transport.write(data)
        self.sock.send.assert_called_with(data)

    def test_write_bytearray(self):
        data = bytearray(b'data')
        self.sock.send.return_value = len(data)

        transport = self.socket_transport()
        transport.write(data)
        self.sock.send.assert_called_with(data)
        self.assertEqual(data, bytearray(b'data'))  

    def test_write_memoryview(self):
        data = memoryview(b'data')
        self.sock.send.return_value = len(data)

        transport = self.socket_transport()
        transport.write(data)
        self.sock.send.assert_called_with(data)

    def test_write_no_data(self):
        transport = self.socket_transport()
        transport._buffer.append(memoryview(b'data'))
        transport.write(b'')
        self.assertFalse(self.sock.send.called)
        self.assertEqual(list_to_buffer([b'data']), transport._buffer)

    def test_write_buffer(self):
        transport = self.socket_transport()
        transport._buffer.append(b'data1')
        transport.write(b'data2')
        self.assertFalse(self.sock.send.called)
        self.assertEqual(list_to_buffer([b'data1', b'data2']),
                         transport._buffer)

    def test_write_partial(self):
        data = b'data'
        self.sock.send.return_value = 2

        transport = self.socket_transport()
        transport.write(data)

        self.loop.assert_writer(7, transport._write_ready)
        self.assertEqual(list_to_buffer([b'ta']), transport._buffer)

    def test_write_partial_bytearray(self):
        data = bytearray(b'data')
        self.sock.send.return_value = 2

        transport = self.socket_transport()
        transport.write(data)

        self.loop.assert_writer(7, transport._write_ready)
        self.assertEqual(list_to_buffer([b'ta']), transport._buffer)
        self.assertEqual(data, bytearray(b'data'))  

    def test_write_partial_memoryview(self):
        data = memoryview(b'data')
        self.sock.send.return_value = 2

        transport = self.socket_transport()
        transport.write(data)

        self.loop.assert_writer(7, transport._write_ready)
        self.assertEqual(list_to_buffer([b'ta']), transport._buffer)

    def test_write_partial_none(self):
        data = b'data'
        self.sock.send.return_value = 0
        self.sock.fileno.return_value = 7

        transport = self.socket_transport()
        transport.write(data)

        self.loop.assert_writer(7, transport._write_ready)
        self.assertEqual(list_to_buffer([b'data']), transport._buffer)

    def test_write_tryagain(self):
        self.sock.send.side_effect = BlockingIOError

        data = b'data'
        transport = self.socket_transport()
        transport.write(data)

        self.loop.assert_writer(7, transport._write_ready)
        self.assertEqual(list_to_buffer([b'data']), transport._buffer)

    def test_write_sendmsg_no_data(self):
        self.sock.sendmsg = mock.Mock()
        self.sock.sendmsg.return_value = 0
        transport = self.socket_transport(sendmsg=True)
        transport._buffer.append(memoryview(b'data'))
        transport.write(b'')
        self.assertFalse(self.sock.sendmsg.called)
        self.assertEqual(list_to_buffer([b'data']), transport._buffer)

    @unittest.skipUnless(selector_events._HAS_SENDMSG, 'no sendmsg')
    def test_writelines_sendmsg_full(self):
        data = memoryview(b'data')
        self.sock.sendmsg = mock.Mock()
        self.sock.sendmsg.return_value = len(data)

        transport = self.socket_transport(sendmsg=True)
        transport.writelines([data])
        self.assertTrue(self.sock.sendmsg.called)
        self.assertFalse(self.loop.writers)

    @unittest.skipUnless(selector_events._HAS_SENDMSG, 'no sendmsg')
    def test_writelines_sendmsg_partial(self):
        data = memoryview(b'data')
        self.sock.sendmsg = mock.Mock()
        self.sock.sendmsg.return_value = 2

        transport = self.socket_transport(sendmsg=True)
        transport.writelines([data])
        self.assertTrue(self.sock.sendmsg.called)
        self.assertTrue(self.loop.writers)

    def test_writelines_send_full(self):
        data = memoryview(b'data')
        self.sock.send.return_value = len(data)
        self.sock.send.fileno.return_value = 7

        transport = self.socket_transport()
        transport.writelines([data])
        self.assertTrue(self.sock.send.called)
        self.assertFalse(self.loop.writers)

    def test_writelines_send_partial(self):
        data = memoryview(b'data')
        self.sock.send.return_value = 2
        self.sock.send.fileno.return_value = 7

        transport = self.socket_transport()
        transport.writelines([data])
        self.assertTrue(self.sock.send.called)
        self.assertTrue(self.loop.writers)

    @unittest.skipUnless(selector_events._HAS_SENDMSG, 'no sendmsg')
    def test_write_sendmsg_full(self):
        data = memoryview(b'data')
        self.sock.sendmsg = mock.Mock()
        self.sock.sendmsg.return_value = len(data)

        transport = self.socket_transport(sendmsg=True)
        transport._buffer.append(data)
        self.loop._add_writer(7, transport._write_ready)
        transport._write_ready()
        self.assertTrue(self.sock.sendmsg.called)
        self.assertFalse(self.loop.writers)

    @unittest.skipUnless(selector_events._HAS_SENDMSG, 'no sendmsg')
    def test_write_sendmsg_partial(self):

        data = memoryview(b'data')
        self.sock.sendmsg = mock.Mock()
        
        self.sock.sendmsg.return_value = 2

        transport = self.socket_transport(sendmsg=True)
        transport._buffer.append(data)
        self.loop._add_writer(7, transport._write_ready)
        transport._write_ready()
        self.assertTrue(self.sock.sendmsg.called)
        self.assertTrue(self.loop.writers)
        self.assertEqual(list_to_buffer([b'ta']), transport._buffer)

    @unittest.skipUnless(selector_events._HAS_SENDMSG, 'no sendmsg')
    def test_write_sendmsg_half_buffer(self):
        data = [memoryview(b'data1'), memoryview(b'data2')]
        self.sock.sendmsg = mock.Mock()
        
        self.sock.sendmsg.return_value = 2

        transport = self.socket_transport(sendmsg=True)
        transport._buffer.extend(data)
        self.loop._add_writer(7, transport._write_ready)
        transport._write_ready()
        self.assertTrue(self.sock.sendmsg.called)
        self.assertTrue(self.loop.writers)
        self.assertEqual(list_to_buffer([b'ta1', b'data2']), transport._buffer)

    @unittest.skipUnless(selector_events._HAS_SENDMSG, 'no sendmsg')
    def test_write_sendmsg_OSError(self):
        data = memoryview(b'data')
        self.sock.sendmsg = mock.Mock()
        err = self.sock.sendmsg.side_effect = OSError()

        transport = self.socket_transport(sendmsg=True)
        transport._fatal_error = mock.Mock()
        transport._buffer.extend(data)
        
        transport._write_ready()
        self.assertTrue(self.sock.sendmsg.called)
        self.assertFalse(self.loop.writers)
        self.assertEqual(list_to_buffer([]), transport._buffer)
        transport._fatal_error.assert_called_with(
                                   err,
                                   'Fatal write error on socket transport')

    @mock.patch('asyncio.selector_events.logger')
    def test_write_exception(self, m_log):
        err = self.sock.send.side_effect = OSError()

        data = b'data'
        transport = self.socket_transport()
        transport._fatal_error = mock.Mock()
        transport.write(data)
        transport._fatal_error.assert_called_with(
                                   err,
                                   'Fatal write error on socket transport')
        transport._conn_lost = 1

        self.sock.reset_mock()
        transport.write(data)
        self.assertFalse(self.sock.send.called)
        self.assertEqual(transport._conn_lost, 2)
        transport.write(data)
        transport.write(data)
        transport.write(data)
        transport.write(data)
        m_log.warning.assert_called_with('socket.send() raised exception.')

    def test_write_str(self):
        transport = self.socket_transport()
        self.assertRaises(TypeError, transport.write, 'str')

    def test_write_closing(self):
        transport = self.socket_transport()
        transport.close()
        self.assertEqual(transport._conn_lost, 1)
        transport.write(b'data')
        self.assertEqual(transport._conn_lost, 2)

    def test_write_ready(self):
        data = b'data'
        self.sock.send.return_value = len(data)

        transport = self.socket_transport()
        transport._buffer.append(data)
        self.loop._add_writer(7, transport._write_ready)
        transport._write_ready()
        self.assertTrue(self.sock.send.called)
        self.assertFalse(self.loop.writers)

    def test_write_ready_closing(self):
        data = memoryview(b'data')
        self.sock.send.return_value = len(data)

        transport = self.socket_transport()
        transport._closing = True
        transport._buffer.append(data)
        self.loop._add_writer(7, transport._write_ready)
        transport._write_ready()
        self.assertTrue(self.sock.send.called)
        self.assertFalse(self.loop.writers)
        self.sock.close.assert_called_with()
        self.protocol.connection_lost.assert_called_with(None)

    @unittest.skipIf(sys.flags.optimize, "Assertions are disabled in optimized mode")
    def test_write_ready_no_data(self):
        transport = self.socket_transport()
        
        self.assertRaises(AssertionError, transport._write_ready)

    def test_write_ready_partial(self):
        data = memoryview(b'data')
        self.sock.send.return_value = 2

        transport = self.socket_transport()
        transport._buffer.append(data)
        self.loop._add_writer(7, transport._write_ready)
        transport._write_ready()
        self.loop.assert_writer(7, transport._write_ready)
        self.assertEqual(list_to_buffer([b'ta']), transport._buffer)

    def test_write_ready_partial_none(self):
        data = b'data'
        self.sock.send.return_value = 0

        transport = self.socket_transport()
        transport._buffer.append(data)
        self.loop._add_writer(7, transport._write_ready)
        transport._write_ready()
        self.loop.assert_writer(7, transport._write_ready)
        self.assertEqual(list_to_buffer([b'data']), transport._buffer)

    def test_write_ready_tryagain(self):
        self.sock.send.side_effect = BlockingIOError

        transport = self.socket_transport()
        buffer = list_to_buffer([b'data1', b'data2'])
        transport._buffer = buffer
        self.loop._add_writer(7, transport._write_ready)
        transport._write_ready()

        self.loop.assert_writer(7, transport._write_ready)
        self.assertEqual(buffer, transport._buffer)

    def test_write_ready_exception(self):
        err = self.sock.send.side_effect = OSError()

        transport = self.socket_transport()
        transport._fatal_error = mock.Mock()
        transport._buffer.extend(b'data')
        transport._write_ready()
        transport._fatal_error.assert_called_with(
                                   err,
                                   'Fatal write error on socket transport')

    def test_write_eof(self):
        tr = self.socket_transport()
        self.assertTrue(tr.can_write_eof())
        tr.write_eof()
        self.sock.shutdown.assert_called_with(socket.SHUT_WR)
        tr.write_eof()
        self.assertEqual(self.sock.shutdown.call_count, 1)
        tr.close()

    def test_write_eof_buffer(self):
        tr = self.socket_transport()
        self.sock.send.side_effect = BlockingIOError
        tr.write(b'data')
        tr.write_eof()
        self.assertEqual(tr._buffer, list_to_buffer([b'data']))
        self.assertTrue(tr._eof)
        self.assertFalse(self.sock.shutdown.called)
        self.sock.send.side_effect = lambda _: 4
        tr._write_ready()
        self.assertTrue(self.sock.send.called)
        self.sock.shutdown.assert_called_with(socket.SHUT_WR)
        tr.close()

    def test_write_eof_after_close(self):
        tr = self.socket_transport()
        tr.close()
        self.loop.run_until_complete(asyncio.sleep(0))
        tr.write_eof()

    @mock.patch('asyncio.base_events.logger')
    def test_transport_close_remove_writer(self, m_log):
        remove_writer = self.loop._remove_writer = mock.Mock()

        transport = self.socket_transport()
        transport.close()
        remove_writer.assert_called_with(self.sock_fd)


class SelectorSocketTransportBufferedProtocolTests(test_utils.TestCase):

    def setUp(self):
        super().setUp()
        self.loop = self.new_test_loop()

        self.protocol = test_utils.make_test_protocol(asyncio.BufferedProtocol)
        self.buf = bytearray(1)
        self.protocol.get_buffer.side_effect = lambda hint: self.buf

        self.sock = mock.Mock(socket.socket)
        self.sock_fd = self.sock.fileno.return_value = 7

    def socket_transport(self, waiter=None):
        transport = _SelectorSocketTransport(self.loop, self.sock,
                                             self.protocol, waiter=waiter)
        self.addCleanup(close_transport, transport)
        return transport

    def test_ctor(self):
        waiter = self.loop.create_future()
        tr = self.socket_transport(waiter=waiter)
        self.loop.run_until_complete(waiter)

        self.loop.assert_reader(7, tr._read_ready)
        test_utils.run_briefly(self.loop)
        self.protocol.connection_made.assert_called_with(tr)

    def test_get_buffer_error(self):
        transport = self.socket_transport()
        transport._fatal_error = mock.Mock()

        self.loop.call_exception_handler = mock.Mock()
        self.protocol.get_buffer.side_effect = LookupError()

        transport._read_ready()

        self.assertTrue(transport._fatal_error.called)
        self.assertTrue(self.protocol.get_buffer.called)
        self.assertFalse(self.protocol.buffer_updated.called)

    def test_get_buffer_zerosized(self):
        transport = self.socket_transport()
        transport._fatal_error = mock.Mock()

        self.loop.call_exception_handler = mock.Mock()
        self.protocol.get_buffer.side_effect = lambda hint: bytearray(0)

        transport._read_ready()

        self.assertTrue(transport._fatal_error.called)
        self.assertTrue(self.protocol.get_buffer.called)
        self.assertFalse(self.protocol.buffer_updated.called)

    def test_proto_type_switch(self):
        self.protocol = test_utils.make_test_protocol(asyncio.Protocol)
        transport = self.socket_transport()

        self.sock.recv.return_value = b'data'
        transport._read_ready()

        self.protocol.data_received.assert_called_with(b'data')

        

        buf_proto = test_utils.make_test_protocol(asyncio.BufferedProtocol)
        buf = bytearray(4)
        buf_proto.get_buffer.side_effect = lambda hint: buf

        transport.set_protocol(buf_proto)

        self.sock.recv_into.return_value = 10
        transport._read_ready()

        buf_proto.get_buffer.assert_called_with(-1)
        buf_proto.buffer_updated.assert_called_with(10)

    def test_buffer_updated_error(self):
        transport = self.socket_transport()
        transport._fatal_error = mock.Mock()

        self.loop.call_exception_handler = mock.Mock()
        self.protocol.buffer_updated.side_effect = LookupError()

        self.sock.recv_into.return_value = 10
        transport._read_ready()

        self.assertTrue(transport._fatal_error.called)
        self.assertTrue(self.protocol.get_buffer.called)
        self.assertTrue(self.protocol.buffer_updated.called)

    def test_read_eof_received_error(self):
        transport = self.socket_transport()
        transport.close = mock.Mock()
        transport._fatal_error = mock.Mock()

        self.loop.call_exception_handler = mock.Mock()

        self.protocol.eof_received.side_effect = LookupError()

        self.sock.recv_into.return_value = 0
        transport._read_ready()

        self.protocol.eof_received.assert_called_with()
        self.assertTrue(transport._fatal_error.called)

    def test_read_ready(self):
        transport = self.socket_transport()

        self.sock.recv_into.return_value = 10
        transport._read_ready()

        self.protocol.get_buffer.assert_called_with(-1)
        self.protocol.buffer_updated.assert_called_with(10)

    def test_read_ready_eof(self):
        transport = self.socket_transport()
        transport.close = mock.Mock()

        self.sock.recv_into.return_value = 0
        transport._read_ready()

        self.protocol.eof_received.assert_called_with()
        transport.close.assert_called_with()

    def test_read_ready_eof_keep_open(self):
        transport = self.socket_transport()
        transport.close = mock.Mock()

        self.sock.recv_into.return_value = 0
        self.protocol.eof_received.return_value = True
        transport._read_ready()

        self.protocol.eof_received.assert_called_with()
        self.assertFalse(transport.close.called)

    @mock.patch('logging.exception')
    def test_read_ready_tryagain(self, m_exc):
        self.sock.recv_into.side_effect = BlockingIOError

        transport = self.socket_transport()
        transport._fatal_error = mock.Mock()
        transport._read_ready()

        self.assertFalse(transport._fatal_error.called)

    @mock.patch('logging.exception')
    def test_read_ready_tryagain_interrupted(self, m_exc):
        self.sock.recv_into.side_effect = InterruptedError

        transport = self.socket_transport()
        transport._fatal_error = mock.Mock()
        transport._read_ready()

        self.assertFalse(transport._fatal_error.called)

    @mock.patch('logging.exception')
    def test_read_ready_conn_reset(self, m_exc):
        err = self.sock.recv_into.side_effect = ConnectionResetError()

        transport = self.socket_transport()
        transport._force_close = mock.Mock()
        with test_utils.disable_logger():
            transport._read_ready()
        transport._force_close.assert_called_with(err)

    @mock.patch('logging.exception')
    def test_read_ready_err(self, m_exc):
        err = self.sock.recv_into.side_effect = OSError()

        transport = self.socket_transport()
        transport._fatal_error = mock.Mock()
        transport._read_ready()

        transport._fatal_error.assert_called_with(
                                   err,
                                   'Fatal read error on socket transport')


class SelectorDatagramTransportTests(test_utils.TestCase):

    def setUp(self):
        super().setUp()
        self.loop = self.new_test_loop()
        self.protocol = test_utils.make_test_protocol(asyncio.DatagramProtocol)
        self.sock = mock.Mock(spec_set=socket.socket)
        self.sock.fileno.return_value = 7

    def datagram_transport(self, address=None):
        self.sock.getpeername.side_effect = None if address else OSError
        transport = _SelectorDatagramTransport(self.loop, self.sock,
                                               self.protocol,
                                               address=address)
        self.addCleanup(close_transport, transport)
        return transport

    def test_read_ready(self):
        transport = self.datagram_transport()

        self.sock.recvfrom.return_value = (b'data', ('0.0.0.0', 1234))
        transport._read_ready()

        self.protocol.datagram_received.assert_called_with(
            b'data', ('0.0.0.0', 1234))

    def test_transport_inheritance(self):
        transport = self.datagram_transport()
        self.assertIsInstance(transport, asyncio.DatagramTransport)

    def test_read_ready_tryagain(self):
        transport = self.datagram_transport()

        self.sock.recvfrom.side_effect = BlockingIOError
        transport._fatal_error = mock.Mock()
        transport._read_ready()

        self.assertFalse(transport._fatal_error.called)

    def test_read_ready_err(self):
        transport = self.datagram_transport()

        err = self.sock.recvfrom.side_effect = RuntimeError()
        transport._fatal_error = mock.Mock()
        transport._read_ready()

        transport._fatal_error.assert_called_with(
                                   err,
                                   'Fatal read error on datagram transport')

    def test_read_ready_oserr(self):
        transport = self.datagram_transport()

        err = self.sock.recvfrom.side_effect = OSError()
        transport._fatal_error = mock.Mock()
        transport._read_ready()

        self.assertFalse(transport._fatal_error.called)
        self.protocol.error_received.assert_called_with(err)

    def test_sendto(self):
        data = b'data'
        transport = self.datagram_transport()
        transport.sendto(data, ('0.0.0.0', 1234))
        self.assertTrue(self.sock.sendto.called)
        self.assertEqual(
            self.sock.sendto.call_args[0], (data, ('0.0.0.0', 1234)))

    def test_sendto_bytearray(self):
        data = bytearray(b'data')
        transport = self.datagram_transport()
        transport.sendto(data, ('0.0.0.0', 1234))
        self.assertTrue(self.sock.sendto.called)
        self.assertEqual(
            self.sock.sendto.call_args[0], (data, ('0.0.0.0', 1234)))

    def test_sendto_memoryview(self):
        data = memoryview(b'data')
        transport = self.datagram_transport()
        transport.sendto(data, ('0.0.0.0', 1234))
        self.assertTrue(self.sock.sendto.called)
        self.assertEqual(
            self.sock.sendto.call_args[0], (data, ('0.0.0.0', 1234)))

    def test_sendto_no_data(self):
        transport = self.datagram_transport()
        transport._buffer.append((b'data', ('0.0.0.0', 12345)))
        transport.sendto(b'', ())
        self.assertFalse(self.sock.sendto.called)
        self.assertEqual(
            [(b'data', ('0.0.0.0', 12345))], list(transport._buffer))

    def test_sendto_buffer(self):
        transport = self.datagram_transport()
        transport._buffer.append((b'data1', ('0.0.0.0', 12345)))
        transport.sendto(b'data2', ('0.0.0.0', 12345))
        self.assertFalse(self.sock.sendto.called)
        self.assertEqual(
            [(b'data1', ('0.0.0.0', 12345)),
             (b'data2', ('0.0.0.0', 12345))],
            list(transport._buffer))

    def test_sendto_buffer_bytearray(self):
        data2 = bytearray(b'data2')
        transport = self.datagram_transport()
        transport._buffer.append((b'data1', ('0.0.0.0', 12345)))
        transport.sendto(data2, ('0.0.0.0', 12345))
        self.assertFalse(self.sock.sendto.called)
        self.assertEqual(
            [(b'data1', ('0.0.0.0', 12345)),
             (b'data2', ('0.0.0.0', 12345))],
            list(transport._buffer))
        self.assertIsInstance(transport._buffer[1][0], bytes)

    def test_sendto_buffer_memoryview(self):
        data2 = memoryview(b'data2')
        transport = self.datagram_transport()
        transport._buffer.append((b'data1', ('0.0.0.0', 12345)))
        transport.sendto(data2, ('0.0.0.0', 12345))
        self.assertFalse(self.sock.sendto.called)
        self.assertEqual(
            [(b'data1', ('0.0.0.0', 12345)),
             (b'data2', ('0.0.0.0', 12345))],
            list(transport._buffer))
        self.assertIsInstance(transport._buffer[1][0], bytes)

    def test_sendto_tryagain(self):
        data = b'data'

        self.sock.sendto.side_effect = BlockingIOError

        transport = self.datagram_transport()
        transport.sendto(data, ('0.0.0.0', 12345))

        self.loop.assert_writer(7, transport._sendto_ready)
        self.assertEqual(
            [(b'data', ('0.0.0.0', 12345))], list(transport._buffer))

    @mock.patch('asyncio.selector_events.logger')
    def test_sendto_exception(self, m_log):
        data = b'data'
        err = self.sock.sendto.side_effect = RuntimeError()

        transport = self.datagram_transport()
        transport._fatal_error = mock.Mock()
        transport.sendto(data, ())

        self.assertTrue(transport._fatal_error.called)
        transport._fatal_error.assert_called_with(
                                   err,
                                   'Fatal write error on datagram transport')
        transport._conn_lost = 1

        transport._address = ('123',)
        transport.sendto(data)
        transport.sendto(data)
        transport.sendto(data)
        transport.sendto(data)
        transport.sendto(data)
        m_log.warning.assert_called_with('socket.send() raised exception.')

    def test_sendto_error_received(self):
        data = b'data'

        self.sock.sendto.side_effect = ConnectionRefusedError

        transport = self.datagram_transport()
        transport._fatal_error = mock.Mock()
        transport.sendto(data, ())

        self.assertEqual(transport._conn_lost, 0)
        self.assertFalse(transport._fatal_error.called)

    def test_sendto_error_received_connected(self):
        data = b'data'

        self.sock.send.side_effect = ConnectionRefusedError

        transport = self.datagram_transport(address=('0.0.0.0', 1))
        transport._fatal_error = mock.Mock()
        transport.sendto(data)

        self.assertFalse(transport._fatal_error.called)
        self.assertTrue(self.protocol.error_received.called)

    def test_sendto_str(self):
        transport = self.datagram_transport()
        self.assertRaises(TypeError, transport.sendto, 'str', ())

    def test_sendto_connected_addr(self):
        transport = self.datagram_transport(address=('0.0.0.0', 1))
        self.assertRaises(
            ValueError, transport.sendto, b'str', ('0.0.0.0', 2))

    def test_sendto_closing(self):
        transport = self.datagram_transport(address=(1,))
        transport.close()
        self.assertEqual(transport._conn_lost, 1)
        transport.sendto(b'data', (1,))
        self.assertEqual(transport._conn_lost, 2)

    def test_sendto_ready(self):
        data = b'data'
        self.sock.sendto.return_value = len(data)

        transport = self.datagram_transport()
        transport._buffer.append((data, ('0.0.0.0', 12345)))
        self.loop._add_writer(7, transport._sendto_ready)
        transport._sendto_ready()
        self.assertTrue(self.sock.sendto.called)
        self.assertEqual(
            self.sock.sendto.call_args[0], (data, ('0.0.0.0', 12345)))
        self.assertFalse(self.loop.writers)

    def test_sendto_ready_closing(self):
        data = b'data'
        self.sock.send.return_value = len(data)

        transport = self.datagram_transport()
        transport._closing = True
        transport._buffer.append((data, ()))
        self.loop._add_writer(7, transport._sendto_ready)
        transport._sendto_ready()
        self.sock.sendto.assert_called_with(data, ())
        self.assertFalse(self.loop.writers)
        self.sock.close.assert_called_with()
        self.protocol.connection_lost.assert_called_with(None)

    def test_sendto_ready_no_data(self):
        transport = self.datagram_transport()
        self.loop._add_writer(7, transport._sendto_ready)
        transport._sendto_ready()
        self.assertFalse(self.sock.sendto.called)
        self.assertFalse(self.loop.writers)

    def test_sendto_ready_tryagain(self):
        self.sock.sendto.side_effect = BlockingIOError

        transport = self.datagram_transport()
        transport._buffer.extend([(b'data1', ()), (b'data2', ())])
        self.loop._add_writer(7, transport._sendto_ready)
        transport._sendto_ready()

        self.loop.assert_writer(7, transport._sendto_ready)
        self.assertEqual(
            [(b'data1', ()), (b'data2', ())],
            list(transport._buffer))

    def test_sendto_ready_exception(self):
        err = self.sock.sendto.side_effect = RuntimeError()

        transport = self.datagram_transport()
        transport._fatal_error = mock.Mock()
        transport._buffer.append((b'data', ()))
        transport._sendto_ready()

        transport._fatal_error.assert_called_with(
                                   err,
                                   'Fatal write error on datagram transport')

    def test_sendto_ready_error_received(self):
        self.sock.sendto.side_effect = ConnectionRefusedError

        transport = self.datagram_transport()
        transport._fatal_error = mock.Mock()
        transport._buffer.append((b'data', ()))
        transport._sendto_ready()

        self.assertFalse(transport._fatal_error.called)

    def test_sendto_ready_error_received_connection(self):
        self.sock.send.side_effect = ConnectionRefusedError

        transport = self.datagram_transport(address=('0.0.0.0', 1))
        transport._fatal_error = mock.Mock()
        transport._buffer.append((b'data', ()))
        transport._sendto_ready()

        self.assertFalse(transport._fatal_error.called)
        self.assertTrue(self.protocol.error_received.called)

    @mock.patch('asyncio.base_events.logger.error')
    def test_fatal_error_connected(self, m_exc):
        transport = self.datagram_transport(address=('0.0.0.0', 1))
        err = ConnectionRefusedError()
        transport._fatal_error(err)
        self.assertFalse(self.protocol.error_received.called)
        m_exc.assert_not_called()

    @mock.patch('asyncio.base_events.logger.error')
    def test_fatal_error_connected_custom_error(self, m_exc):
        class MyException(Exception):
            pass
        transport = self.datagram_transport(address=('0.0.0.0', 1))
        err = MyException()
        transport._fatal_error(err)
        self.assertFalse(self.protocol.error_received.called)
        m_exc.assert_called_with(
            test_utils.MockPattern(
                'Fatal error on transport\nprotocol:.*\ntransport:.*'),
            exc_info=(MyException, MOCK_ANY, MOCK_ANY))


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