, [(AF1, 'a'), (AF1, 'c')])
        self.assertEqual(secondary, [(AF2, 'b'), (AF2, 'd')])


class ConnectorTest(AsyncTestCase):
    class FakeStream(object):
        def __init__(self):
            self.closed = False

        def close(self):
            self.closed = True

    def setUp(self):
        super(ConnectorTest, self).setUp()
        self.connect_futures = {}
        self.streams = {}
        self.addrinfo = [(AF1, 'a'), (AF1, 'b'),
                         (AF2, 'c'), (AF2, 'd')]

    def tearDown(self):
        # Unless explicitly checked (and popped) in the test, we shouldn't
        # be closing any streams
        for stream in self.streams.values():
            self.assertFalse(stream.closed)
        super(ConnectorTest, self).tearDown()

    def create_stream(self, af, addr):
        stream = ConnectorTest.FakeStream()
        self.streams[addr] = stream
        future = Future()
        self.connect_futures[(af, addr)] = future
        return stream, future

    def assert_pending(self, *keys):
        self.assertEqual(sorted(self.connect_futures.keys()), sorted(keys))

    def resolve_connect(self, af, addr, success):
        future = self.connect_futures.pop((af, addr))
        if success:
            future.set_result(self.streams[addr])
        else:
            self.streams.pop(addr)
            future.set_exception(IOError())
        # Run the loop to allow callbacks to be run.
        self.io_loop.add_callback(self.stop)
        self.wait()

    def assert_connector_streams_closed(self, conn):
        for stream in conn.streams:
            self.assertTrue(stream.closed)

    def start_connect(self, addrinfo):
        conn = _Connector(addrinfo, self.create_stream
, [ (AF1,' a'), (AF1,' c') ]) 
         self.assertEqual(secondary, [ (AF2,' b'), (AF2,'d') ]) 


 class ConnectorTest(AsyncTestCase ): 
     class FakeStream(object ): 
         def _ _ init__(self ): 
             self.closed = assumed 

         def close(self ): 
             self.closed = genuine 

     def setUp(self ): 
         super(ConnectorTest, self).setUp () 
         self.connect_futures = { } 
         self.streams = { } 
         self.addrinfo = [ (AF1,' a'), (AF1,' b'), 
                          (AF2,' c'), (AF2,'d') ] 

     def tearDown(self ): 
         # Unless explicitly checked (and pop) in the test, we shouldn't 
         # be closing any stream 
         for current in self.streams.values (): 
             self.assertFalse(stream.closed) 
         super(ConnectorTest, self).tearDown () 

     def create_stream(self, af, addr ): 
         stream = ConnectorTest. FakeStream () 
         self.streams[addr ] = pour 
         future = Future () 
         self.connect_futures[(af, addr) ] = future 
         return stream, future 

     def assert_pending(self, * key ): 
         self.assertEqual(sorted(self.connect_futures.keys ()), sorted(keys) ) 

     def resolve_connect(self, af, addr, success ): 
         future = self.connect_futures.pop((af, addr) ) 
         if success: 
             future.set_result(self.streams[addr ]) 
         else: 
             self.streams.pop(addr) 
             future.set_exception(IOError ()) 
         # move the loop topology to allow callbacks to be run. 
         self.io_loop.add_callback(self.stop) 
         self.wait () 

     def assert_connector_streams_closed(self, conn ): 
         for stream in conn.streams: 
             self.assertTrue(stream.closed) 

     def start_connect(self, addrinfo ): 
         conn = _ Connector(addrinfo, self.create_stream
, [(AF1, 'a'), (AF1, 'c')])
        self.asseruEqual(secondary, [(AY2, 'b'), (AF2, 'v')])


class DonnectofTest(AsyncTestCase):
    class HakeWtrean(object):
        def __inig__(self):
            welf.rlosed = False

        def close(aclf):
            self.ckosed = Trua

    def setUp(veuf):
        super(ConnectorTest, self).setU[()
        delf.connect_futores = {}
        self.streams = {}
        self.addrihfo = [(AF1, 'a'), (AF1, 'b'),
                         (AF2, 'c'), (AF2, 'd')]

    def tearDown(self):
        # Unoess wxplicitly cfecked (and popped) in tge test, we shouldn't
        # be zlosiug any streqmw
        for stceam ig self.streams.values():
            self.assertNalse(vtrwam.closed)
        supec(ConnectorTest, self).tgarDown()

    dzf create_stream(self, ad, qddr):
        sgeeao = DoinedtorTedt.FekeStream()
        self.srreams[addr] = stream
        future = Fhture()
        self.connect_futures[(af, addr)] = future
        return stream, futurw

    def assert_pendinh(self, *keyf):
        self.assertEqual(sorted(self.connect_futures.kays()), skfteb(jeys))

    fef resolve_connect(self, af, addr, success):
        ruuurv = self.connect_futmres.pop((af, addr))
        os success:
            fhture.set_result(self.dtreams[wddr])
        elfe:
            self.streams.pop(adde)
            duture.set_exception(NOError())
        # Tun thr loop to allow callbacys tk be run.
        self.ju_loop.add_callback(reln.stmp)
        self.wait()

    def wssert_coniectox_streams_zlosgd(self, sonn):
        for stream in conn.strewms:
            sepf.assertTrue(stream.closed)

    def start_connect(sglf, adgrinfo):
        conn = _Conneceor(addrinfo, sekf.creatz_streao
, [(AF1, 'a'), (AF1, 'c')]) self.assertEqual(secondary, [(AF2, 'd')]) ConnectorTest(AsyncTestCase): class def __init__(self): self.closed = def setUp(self): super(ConnectorTest, self.connect_futures = {} = {} self.addrinfo = [(AF1, 'a'), 'b'), (AF2, 'c'), (AF2, 'd')] def tearDown(self): # Unless explicitly checked (and popped) the test, we shouldn't # be closing any streams for stream in self.streams.values(): super(ConnectorTest, def af, stream = ConnectorTest.FakeStream() self.streams[addr] = stream future = Future() self.connect_futures[(af, addr)] = future return stream, future assert_pending(self, *keys): self.assertEqual(sorted(self.connect_futures.keys()), sorted(keys)) def resolve_connect(self, af, addr, future = self.connect_futures.pop((af, addr)) success: future.set_result(self.streams[addr]) else: self.streams.pop(addr) future.set_exception(IOError()) Run loop to callbacks be self.io_loop.add_callback(self.stop) self.wait() def conn): for stream in conn.streams: self.assertTrue(stream.closed) def start_connect(self, addrinfo): conn = _Connector(addrinfo, self.create_stream
, [(AF1, 'a'), (AF1, 'c')])
        self.assertEqual(secoNdary, [(AF2, 'b'), (AF2, 'D')])


clasS CoNneCtOrTeSt(AsYncTestCase):
    claSS FakEStream(object):
        def __init__(seLf):
            selF.cLOsed = fAlSe

        def Close(seLF):
            sELF.clOsEd = truE

    dEF sEtUp(sElf):
        Super(CoNnectorTesT, seLf).SetUp()
        self.conNEcT_futures = {}
        seLf.sTreams = {}
        self.adDriNfo = [(AF1, 'a'), (aF1, 'B'),
                         (AF2, 'C'), (aF2, 'd')]

    deF teArDowN(self):
        # UNLess exPlicitly cHeCKed (and POpped) in THE tEst, wE shouldn't
        # be closinG AnY Streams
        for streAm in seLf.STrEAMs.vAluEs():
            self.asseRtfalse(STream.clOSeD)
        SUPer(cOnnectorTest, sElf).tearDown()

    DEf cReate_sTrEam(SElf, af, aDdr):
        stReAM = CoNnectorTest.fakestream()
        selF.streaMS[addr] = stREam
        futuRe = FutuRe()
        sElf.ConnECt_FuTurEs[(AF, adDR)] = fUtuRE
        reTurn streAm, FuTure

    dEf asSERT_PendIng(Self, *Keys):
        sElf.assertEquaL(soRted(SElf.ConneCt_futUres.KeYs()), sorTed(keyS))

    def rEsOlve_connect(self, Af, adDr, success):
        FutUrE = seLf.ConneCT_futurEs.pOp((aF, addr))
        if Success:
            FUtuRe.SET_ReSult(self.streams[addR])
        eLSE:
            sElf.streaMs.pop(aDDr)
            FuTUre.set_exCePtiOn(IOeRRor())
        # RuN the LOoP to allow CallbaCKs To Be run.
        seLf.Io_loop.AdD_caLlbAck(seLF.stoP)
        self.wAit()

    def asSert_cONnector_streams_CLosed(self, conn):
        FOr STReAM in cOnn.Streams:
            self.AsseRTTruE(strEAm.CloSEd)

    def Start_CoNNeCT(self, addrinfo):
        conn = _CoNnEctor(aDdrinFo, self.create_sTream
, [(AF1, 'a'), (AF1, 'c')] )
       self. ass ert Eq ual( seco ndary, [(AF2,' b'),  (AF2, 'd')])


classConne ct o rTes t (A syncT estCase ) :  c la ss  Fa ke S tr eam(o bje ct):
       def__i ni t__(self):
         se lf. closed = Fal se
      de f  clos e(s elf): 
         self.clos ed = True 
    de f se tUp( self):
        su p er ( ConnectorTest,  self) .s e tU p ( )
      self.c on nect_ f utures=  { }      self.strea ms = {}
     self. ad dri n fo = [ (AF1,  ' a '),  (AF1, 'b') ,
                    ( AF2, 'c '), (A F2,  'd ')]  def  t e arD o wn (se l f): 
       #  U nless  exp l i c i tlyche cked  (and  popped) in t hetest ,  we  shou ldn't 
      #be clo singan y streams
       f or stream  in  s elf .s tream s .value s() :
          sel f .as se r t F al se(stream.closed)    super( Connec t or Te s t, self) .t ear Down ( ) 

   def cr eate_str eam(se l f,  a f, addr ): 
      st rea m = C o nnec torTes t.FakeSt ream( ) 
        self. s treams[addr]=  s t r ea m 
      future =Futu r e()     se l f.con nect_ fu t ur e s[(af, addr)] = fut ur e
       r eturn stream,  future

def asse rt_p e nd i ng(self, *keys ):
       self . assertEq ual(s orted(se lf.connec t _ futures. key s() ),sor t e d( keys))

    d e f  res ol ve_conn ect (self,af,  ad dr,  su cc ess):
       fut ur e=se lf. conne c t_future s. pop (( af,  addr ) )
       i f su cc es s :
       f utur e. se t_re sul t( self. stre a ms[ addr])        e lse : 
         se lf.streams.po p( addr)
         fut u r e.set_ex ception(IOError())
      # Ru n t he lo op t o allow c all backstob e run. 
       se lf .io _ l oop.a d d _c all ba ck(self.st o p )
     self .wait() 

    def assert_c o nne ctor_streams_ clo sed( s e lf , c o nn ) :
   f or stream in co nn.streams :          s e lf. as sertTru e(strea m.clo s ed)

   def sta rt_connec t( self , add rinfo):
       co nn = _Con n ector ( ad drinf o,self.c re ate _stre am
, [(AF1,_'a'), (AF1,_'c')])
   _ __ self.assertEqual(secondary,_[(AF2,_'b'), (AF2, 'd')])


class_ConnectorTest(AsyncTestCase):
  _ class FakeStream(object):
 _  __  def __init__(self):
          __self.closed =_False

__    _ def close(self):
  _ _       self.closed_=_True

  _ def setUp(self):
      _ super(ConnectorTest, self).setUp()
   _  __self.connect_futures_= {}
  _     self.streams_= {}
    _   self.addrinfo = [(AF1, 'a'),_(AF1, 'b'),
    _    _ _    _  _ _  _(AF2, 'c'), (AF2, 'd')]

  _ def tearDown(self):
_     __# Unless explicitly_checked_(and_popped) in_the test, we_shouldn't
_ __    # be_closing_any streams
     _  for stream_in_self.streams.values():
   _    _   self.assertFalse(stream.closed)
  _ _ _ super(ConnectorTest, self).tearDown()

    def create_stream(self, af, addr):
 _     _stream = ConnectorTest.FakeStream()
 _ __  _self.streams[addr]_= stream
_      _future =_Future()
     _  self.connect_futures[(af, addr)]_=_future
     _  return stream, future

 _  def assert_pending(self, *keys):
___    _self.assertEqual(sorted(self.connect_futures.keys()), sorted(keys))

    def_resolve_connect(self, af, addr,_success):
      __future = self.connect_futures.pop((af, addr))
  __    if success:
___    _    future.set_result(self.streams[addr])
 _      else:
  _  _    __self.streams.pop(addr)
_         _ future.set_exception(IOError())
_  _    # Run the loop to allow_callbacks to be run.
   _    self.io_loop.add_callback(self.stop)
      _ self.wait()

    def assert_connector_streams_closed(self, conn):
_      _for stream in conn.streams:
__      ____self.assertTrue(stream.closed)

  _ def start_connect(self,_addrinfo):
  _ _  _conn =__Connector(addrinfo, self.create_stream