def testWritePrio(self):
     reactor = Reactor()
     local0, remote0 = socketpair()
     local1, remote1 = socketpair()
     local1.setblocking(0)
     def remoteHandler0():
         remote0.send('ape')
     def remoteHandler1():
         remote1.send('nut')
     reactor.addWriter(remote0, remoteHandler0, 0)
     reactor.addWriter(remote1, remoteHandler1, 2)
     reactor.step() #0
     self.assertEquals('ape', local0.recv(999))
     try:
         local1.recv(999)
         self.fail('must be no data on the socket yet')
     except error:
         pass
     reactor.step() #1
     try:
         local1.recv(999)
         self.fail('must be no data on the socket yet')
     except error:
         pass
     reactor.step() #2
     self.assertEquals('nut', local1.recv(999))

 def do_send_with_send_error(self, raise_on_send, send_exception,
                             expect_answer=True,
                             expect_send_exception=None):
     """
     Sets up two connected sockets, wraps the sender socket into a BadSocket
     class, then performs a do_send() test.
     Parameters:
     raise_on_send: the byte at which send_exception should be raised
                    (see BadSocket)
     send_exception: the exception to raise (see BadSocket)
     expect_answer: whether the send is expected to complete (and hence
                    the read socket should get the message)
     expect_send_exception: the exception msgq is expected to raise when
                            send_exception is raised by BadSocket.
     """
     (write, read) = socket.socketpair(socket.AF_UNIX, socket.SOCK_STREAM)
     (control_write, control_read) = socket.socketpair(socket.AF_UNIX,
                                                       socket.SOCK_STREAM)
     badwrite = BadSocket(write, raise_on_send, send_exception)
     self.do_send(badwrite, read, control_write, control_read,
                  expect_answer, expect_send_exception)
     write.close()
     read.close()
     control_write.close()
     control_read.close()

	def __init__ (self, ircprox, localid, remotid):
		"""Start a new PrivateChannel.  The server can be
		   reached over ircprox.send_to_server(), which will
		   ensure proper locking.  The communication is
		   between the given localid and remotid (both in
		   nick form, see nick2nai(), and anything passing
		   over the PrivateChannel will be sent as a PRIVMSG
		   targeted at remotid, followed by TLS and one line
		   of base64-encoded TLS binary data.  Large TLS data
		   portions may be split over multiple lines, but
		   each is separately base64-encoded, meaning, each
		   line has its own "=" sign trailer.
		"""
		assert (remotid [:1] != '#')
		self.ircprox = ircprox
		self.localid = localid
		self.remotid = remotid
		self.localid_nai = nick2nai (localid)
		self.remotid_nai = nick2nai (remotid)
		self.localhs = None
		self.remoths = None
		self.plaindownbuf = ''
		self.poolcrypt, self.chancrypt = socket.socketpair ()
		self.poolplain, self.chanplain = socket.socketpair ()
		print 'PrivateChannel crypt @pool =', self.poolcrypt, '@chan =', self.chancrypt
		print 'PrivateChannel plain @pool =', self.poolplain, '@chan =', self.chanplain
		self.insecure = 0
		#TODO# Following should not return before done
		# self.start ()
		self.initiate_starttls_handshake ()

 def __init__(self, pcapfile, connection, ssfile):
   self.ssfile = ssfile
   self.pcapfile = pcapfile
   self.connection = connection
   self.inbound=dict()
   self.outbound=dict()
   self.in_rsocket,self.in_wsocket = socket.socketpair()
   self.out_rsocket,self.out_wsocket = socket.socketpair()

    def setUp(self):
        (self.c, self.s) = socket.socketpair(socket.AF_UNIX, socket.SOCK_STREAM)

        self.client = TransPort(self.c)

        self.server = threading.Thread(target=_server, args=(self.s,))
        self.server.start()

def create_sock_pair(port=0):
    '''Create socket pair. Works also on windows by using an ephemeral TCP port.'''
    if hasattr(socket, 'socketpair'):
        client_sock, srv_sock = socket.socketpair()
        return client_sock, srv_sock

    # Create a non-blocking temporary server socket
    temp_srv_sock = socket.socket()
    temp_srv_sock.setblocking(False)
    temp_srv_sock.bind(('localhost', port))
    port = temp_srv_sock.getsockname()[1]
    temp_srv_sock.listen(1)
    with closing(temp_srv_sock):
        # Create non-blocking client socket
        client_sock = socket.socket()
        client_sock.setblocking(False)
        while True:
            try:
                client_sock.connect(('localhost', port))
            except socket.error as err:
                # EWOULDBLOCK is not an error, as the socket is non-blocking
                if err.errno not in socket_errors_nonblocking:
                    raise

        # Use select to wait for connect() to succeed.
        timeout = 1
        readable = select.select([temp_srv_sock], [], [], timeout)[0]
        if temp_srv_sock not in readable:
            raise Exception('Client socket not connected in {} second(s)'.format(timeout))
        srv_sock = temp_srv_sock.accept()[0]
        client_sock.setblocking(True)

    return client_sock, srv_sock

    def rpc_run(self, pcode, user, visitor):
        uid = 61018

        # localtime = time.asctime(time.localtime(time.time()))
        # userdir = '/tmp/' + (localtime[8]+localtime[17]+localtime[18]+localtime[14]+localtime[15])
        userdir = "/tmp" + "/" + user.replace("/", "9")
        if not os.path.exists(userdir):
            os.mkdir(userdir)
            os.chmod(userdir, 0770)
        # print "Directory created " + userdir

        db = zoodb.cred_setup()
        person = db.query(zoodb.Cred).get(user)
        if not person:
            return None
        token = person.token

        (sa, sb) = socket.socketpair(socket.AF_UNIX, socket.SOCK_STREAM, 0)
        pid = os.fork()
        if pid == 0:
            if os.fork() <= 0:
                sa.close()
                ProfileAPIServer(user, visitor, token).run_sock(sb)
                sys.exit(0)
            else:
                sys.exit(0)
        sb.close()
        os.waitpid(pid, 0)

        sandbox = sandboxlib.Sandbox(userdir, uid, "/profilesvc/lockfile")
        with rpclib.RpcClient(sa) as profile_api_client:
            return sandbox.run(lambda: run_profile(pcode, profile_api_client))

 def test_echo(self, concurrency):
     msg = b"ping"
     sockets = []
     try:
         for i in range(concurrency):
             sock1, sock2 = socket.socketpair()
             self.loop.create_dispatcher(Echo, sock2)
             sockets.append(sock1)
         t = concurrent.thread(self.loop.run_forever)
         t.start()
         try:
             start = time.time()
             for sock in sockets:
                 osutils.uninterruptible(sock.send, msg)
             for sock in sockets:
                 data = osutils.uninterruptible(sock.recv, len(msg))
                 self.assertEqual(data, msg)
             elapsed = time.time() - start
             print("%7d echos: %f seconds" % (concurrency, elapsed))
         finally:
             self.loop.call_soon_threadsafe(self.loop.stop)
             t.join()
     finally:
         for sock in sockets:
             sock.close()

    def __init__(self, fd):
        (self._rsock, wsock) = socket.socketpair(socket.AF_UNIX, socket.SOCK_DGRAM)
        child = os.fork()
        if child != 0:
            wsock.close()
            self._child = child
            return

        self._rsock.close()

        # This is the read loop in the forked process and it won't quit until either the process
        # gets killed or there is a read error.
        try:
            while True:
                packet = handleEAgain(os.read, fd, MAX_PACKET_SIZE)
                handleEAgain(wsock.send, pickle.dumps((0, packet)))
                if len(packet) == 0:
                    break
        except KeyboardInterrupt:
            pass
        except EnvironmentError as e:
            handleEAgain(wsock.send, pickle.dumps((e.errno, '')))
        finally:
            os.close(fd)
            wsock.close()
            os._exit(os.EX_OK)

    def test_pack_ping(self):
        config = kademlia.utils.load_config("config.json")

        loop = asyncio.get_event_loop()
        loop.set_debug(config['debug']['asyncio']['enabled'])

        service = kademlia.Service(config, loop)
        echo = kademlia.utils.get_echo_bytes()

        rsock, wsock = socketpair()

        reader, writer = loop.run_until_complete(
            asyncio.open_connection(sock=rsock, loop=loop)
        )

        wsock.send(
            service.tcpService.rpc.pack_ping(service.tcpService.node, echo)
        )

        _command, _echo, _remoteNode, _data = loop.run_until_complete(
            asyncio.ensure_future(
                service.tcpService.rpc.read_command(reader)
            )
        )
        writer.close()
        wsock.close()

        self.assertEqual(_command, kademlia.const.kad.command.PING)
        self.assertEqual(echo, _echo)

    def __init__(self, bootstrap, timeout=60, debug=True, token=None):
        self.debug = debug
        self.timeout = timeout 
        self.domainname = None
        self.token = token
        self.socket = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
        self.socket.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)
        self.socket.setsockopt(socket.IPPROTO_TCP,socket.TCP_NODELAY,1)

        self.bootstraplist = self.discoverbootstrap(bootstrap)
        if len(self.bootstraplist) == 0:
            raise ConnectionError("No bootstrap found")
        if not self.connecttobootstrap():
            raise ConnectionError("Cannot connect to any bootstrap")
        myaddr = findOwnIP()
        myport = self.socket.getsockname()[1]
        self.me = Peer(myaddr,myport,NODE_CLIENT)
        self.commandnumber = random.randint(1, sys.maxint)

        # synchronization
        self.lock = RLock()
        self.ctrlsockets, self.ctrlsocketr = socket.socketpair()
        self.reqlist = []     # requests we have received from client threads
        self.pendingops = {}  # pending requests indexed by command number

        # spawn thread, invoke comm_loop
        comm_thread = Thread(target=self.comm_loop, name='CommunicationThread')
        comm_thread.start()

def SocketPair(family=None, type=socket.SOCK_STREAM, proto=0):
    """ Creats a connected pair of sockets.  Works on win32 by creating
    a listening soket on localhost and then connecting to it.  The 
    family arg defaults to AF_INET on win32 and AF_UNIX elsewhere. """
    
    if sys.platform != 'win32':
        if family is None:
            family = socket.AF_UNIX
        return socket.socketpair(family, type, proto)

    if family is None:
        family = socket.AF_INET
        
    accepted = None
    conn = None
    listener = socket.socket(family=family, type=type, proto=proto)
    try:
        listener.bind(('127.0.0.1', 0))
        listener.listen(16)
        conn = socket.socket(family=family, type=type, proto=proto)
        conn.connect(listener.getsockname())
        while True:
            accepted, addr = listener.accept()
            if addr == conn.getsockname():
                return accepted, conn
    except Exception:
        if accepted:
            accepted.close()
        if conn:
            conn.close()
        raise
    finally:
        listener.close()

def _start_child(nonetns):
    # Create socket pair to communicate
    (s0, s1) = socket.socketpair(socket.AF_UNIX, socket.SOCK_STREAM, 0)
    # Spawn a child that will run in a loop
    pid = os.fork()
    if pid:
        s1.close()
        return (s0, pid)

    # FIXME: clean up signal handers, atexit functions, etc.
    try: # pragma: no cover
        # coverage doesn't seem to understand fork
        s0.close()
        srv = nemu.protocol.Server(s1, s1)
        if not nonetns:
            # create new name space
            unshare.unshare(unshare.CLONE_NEWNET)
            # Enable packet forwarding
            execute([SYSCTL_PATH, '-w', 'net.ipv4.ip_forward=1'])
            execute([SYSCTL_PATH, '-w', 'net.ipv6.conf.default.forwarding=1'])
        srv.run()
    except BaseException, e:
        s = "Slave node aborting: %s\n" % str(e)
        sep = "=" * 70 + "\n"
        sys.stderr.write(s + sep)
        traceback.print_exc(file=sys.stdout)
        sys.stderr.write(sep)
        try:
            # try to pass the error to parent, if possible
            s1.send("500 " + s)
        except:
            pass
        os._exit(1)

def test_parse_from_real_socket():
    # would fail on python2.6 before the recv_into hack
    sock, sink = socket.socketpair()
    sink.send(complete_request)
    reader = SocketReader(sock)
    stream = HttpStream(reader)
    assert stream.headers()

 def _pair(self):
     a, b = socket.socketpair()
     a.setblocking(False)
     b.setblocking(False)
     c, d = ScheduledFile.fromSocket(a), ScheduledFile.fromSocket(b)
     c.autoflush = d.autoflush = True
     return c, d

    def test_create_unix_connection_5(self):
        s1, s2 = socket.socketpair(socket.AF_UNIX)

        excs = []

        class Proto(asyncio.Protocol):
            def connection_lost(self, exc):
                excs.append(exc)

        proto = Proto()

        async def client():
            t, _ = await self.loop.create_unix_connection(
                lambda: proto,
                None,
                sock=s2)

            t.write(b'AAAAA')
            s1.close()
            t.write(b'AAAAA')
            await asyncio.sleep(0.1, loop=self.loop)

        self.loop.run_until_complete(client())

        self.assertEqual(len(excs), 1)
        self.assertIn(excs[0].__class__,
            (BrokenPipeError, ConnectionResetError))

    def _test_recv(self, recv_func):
        rd, wr = socket.socketpair()
        self.addCleanup(rd.close)
        # wr closed explicitly by parent

        # single-byte payload guard us against partial recv
        datas = [b"x", b"y", b"z"]

        code = '\n'.join((
            'import os, socket, sys, time',
            '',
            'fd = int(sys.argv[1])',
            'family = %s' % int(wr.family),
            'sock_type = %s' % int(wr.type),
            'datas = %r' % datas,
            'sleep_time = %r' % self.sleep_time,
            '',
            'wr = socket.fromfd(fd, family, sock_type)',
            'os.close(fd)',
            '',
            'with wr:',
            '    for data in datas:',
            '        # let the parent block on recv()',
            '        time.sleep(sleep_time)',
            '        wr.sendall(data)',
        ))

        fd = wr.fileno()
        proc = self.subprocess(code, str(fd), pass_fds=[fd])
        with kill_on_error(proc):
            wr.close()
            for data in datas:
                self.assertEqual(data, recv_func(rd, len(data)))
            self.assertEqual(proc.wait(), 0)

    def test_buildProtocolReturnsNone(self):
        """
        {IReactorSocket.adoptStreamConnection} returns None if the given
        factory's buildProtocol returns None.
        """

        # Build reactor before anything else: allow self.buildReactor()
        # to skip the test if any of the self.requiredInterfaces isn't
        # provided by the reactor (example: Windows), preventing later
        # failures unrelated to the test itself.
        reactor = self.buildReactor()

        from socket import socketpair

        class NoneFactory(ServerFactory):
            def buildProtocol(self, address):
                return None

        s1, s2 = socketpair(AF_UNIX, SOCK_STREAM)
        s1.setblocking(False)
        self.addCleanup(s1.close)
        self.addCleanup(s2.close)

        s1FD = s1.fileno()
        factory = NoneFactory()
        result = reactor.adoptStreamConnection(s1FD, AF_UNIX, factory)
        self.assertIsNone(result)

    def test_handle_error_failures(self):

        class EvilDispatcher(Echo):

            def handle_read(self):
                Echo.handle_read(self)
                raise Exception("Expected error")

            def handle_error(self):
                # This is a very big anti-pattern for dispatchers,
                # asyncore.poll2 will raise errors raised from handle_error.
                raise Exception("Evil error")

        def pinger(sock):
            msg = b"ping"
            osutils.uninterruptible(sock.send, msg)
            osutils.uninterruptible(sock.recv, len(msg))
            sock.close()
            self.loop.call_soon_threadsafe(self.loop.stop)

        sock1, sock2 = socket.socketpair()
        # The dispatcher and pinger owns the sockets
        self.loop.create_dispatcher(EvilDispatcher, sock2)
        t = concurrent.thread(pinger, args=(sock1,))
        t.start()
        try:
            # Correct error handling willl allow this test to complete without
            # errors. This used to abort the event loop with the error raised
            # in handle_error.
            self.loop.run_forever()
        finally:
            t.join()

    def test_remove_handler_from_handler(self):
        # Create two sockets with simultaneous read events.
        client, server = socket.socketpair()
        try:
            client.send(b'abc')
            server.send(b'abc')

            # After reading from one fd, remove the other from the IOLoop.
            chunks = []

            def handle_read(fd, events):
                chunks.append(fd.recv(1024))
                if fd is client:
                    self.io_loop.remove_handler(server)
                else:
                    self.io_loop.remove_handler(client)
            self.io_loop.add_handler(client, handle_read, self.io_loop.READ)
            self.io_loop.add_handler(server, handle_read, self.io_loop.READ)
            self.io_loop.call_later(0.1, self.stop)
            self.wait()

            # Only one fd was read; the other was cleanly removed.
            self.assertEqual(chunks, [b'abc'])
        finally:
            client.close()
            server.close()