1. 模块列表
  2. 函数列表
  3. asyncore.file_wrapper()

Python asyncore 模块，file_wrapper() 实例源码

我们从Python开源项目中，提取了以下26个代码示例，用于说明如何使用asyncore.file_wrapper()。

项目：zippy    作者：securesystemslab    | 项目源码 | 文件源码 
def test_recv(self):
        fd = os.open(TESTFN, os.O_RDONLY)
        w = asyncore.file_wrapper(fd)
        os.close(fd)

        self.assertNotEqual(w.fd, fd)
        self.assertNotEqual(w.fileno(), fd)
        self.assertEqual(w.recv(13), b"It's not dead")
        self.assertEqual(w.read(6), b", it's")
        w.close()
        self.assertRaises(OSError, w.read, 1)
项目：zippy    作者：securesystemslab    | 项目源码 | 文件源码 
def test_send(self):
        d1 = b"Come again?"
        d2 = b"I want to buy some cheese."
        fd = os.open(TESTFN, os.O_WRONLY | os.O_APPEND)
        w = asyncore.file_wrapper(fd)
        os.close(fd)

        w.write(d1)
        w.send(d2)
        w.close()
        with open(TESTFN, 'rb') as file:
            self.assertEqual(file.read(), self.d + d1 + d2)
项目：oil    作者：oilshell    | 项目源码 | 文件源码 
def test_recv(self):
        fd = os.open(TESTFN, os.O_RDONLY)
        w = asyncore.file_wrapper(fd)
        os.close(fd)

        self.assertNotEqual(w.fd, fd)
        self.assertNotEqual(w.fileno(), fd)
        self.assertEqual(w.recv(13), "It's not dead")
        self.assertEqual(w.read(6), ", it's")
        w.close()
        self.assertRaises(OSError, w.read, 1)
项目：oil    作者：oilshell    | 项目源码 | 文件源码 
def test_send(self):
        d1 = "Come again?"
        d2 = "I want to buy some cheese."
        fd = os.open(TESTFN, os.O_WRONLY | os.O_APPEND)
        w = asyncore.file_wrapper(fd)
        os.close(fd)

        w.write(d1)
        w.send(d2)
        w.close()
        self.assertEqual(file(TESTFN).read(), self.d + d1 + d2)
项目：python2-tracer    作者：extremecoders-re    | 项目源码 | 文件源码 
def test_recv(self):
        fd = os.open(TESTFN, os.O_RDONLY)
        w = asyncore.file_wrapper(fd)
        os.close(fd)

        self.assertNotEqual(w.fd, fd)
        self.assertNotEqual(w.fileno(), fd)
        self.assertEqual(w.recv(13), "It's not dead")
        self.assertEqual(w.read(6), ", it's")
        w.close()
        self.assertRaises(OSError, w.read, 1)
项目：python2-tracer    作者：extremecoders-re    | 项目源码 | 文件源码 
def test_send(self):
        d1 = "Come again?"
        d2 = "I want to buy some cheese."
        fd = os.open(TESTFN, os.O_WRONLY | os.O_APPEND)
        w = asyncore.file_wrapper(fd)
        os.close(fd)

        w.write(d1)
        w.send(d2)
        w.close()
        self.assertEqual(file(TESTFN).read(), self.d + d1 + d2)
项目：web_ctp    作者：molebot    | 项目源码 | 文件源码 
def test_recv(self):
        fd = os.open(TESTFN, os.O_RDONLY)
        w = asyncore.file_wrapper(fd)
        os.close(fd)

        self.assertNotEqual(w.fd, fd)
        self.assertNotEqual(w.fileno(), fd)
        self.assertEqual(w.recv(13), b"It's not dead")
        self.assertEqual(w.read(6), b", it's")
        w.close()
        self.assertRaises(OSError, w.read, 1)
项目：web_ctp    作者：molebot    | 项目源码 | 文件源码 
def test_send(self):
        d1 = b"Come again?"
        d2 = b"I want to buy some cheese."
        fd = os.open(TESTFN, os.O_WRONLY | os.O_APPEND)
        w = asyncore.file_wrapper(fd)
        os.close(fd)

        w.write(d1)
        w.send(d2)
        w.close()
        with open(TESTFN, 'rb') as file:
            self.assertEqual(file.read(), self.d + d1 + d2)
项目：pefile.pypy    作者：cloudtracer    | 项目源码 | 文件源码 
def test_recv(self):
        fd = os.open(TESTFN, os.O_RDONLY)
        w = asyncore.file_wrapper(fd)
        os.close(fd)

        self.assertNotEqual(w.fd, fd)
        self.assertNotEqual(w.fileno(), fd)
        self.assertEqual(w.recv(13), "It's not dead")
        self.assertEqual(w.read(6), ", it's")
        w.close()
        self.assertRaises(OSError, w.read, 1)
项目：pefile.pypy    作者：cloudtracer    | 项目源码 | 文件源码 
def test_send(self):
        d1 = "Come again?"
        d2 = "I want to buy some cheese."
        fd = os.open(TESTFN, os.O_WRONLY | os.O_APPEND)
        w = asyncore.file_wrapper(fd)
        os.close(fd)

        w.write(d1)
        w.send(d2)
        w.close()
        self.assertEqual(file(TESTFN).read(), self.d + d1 + d2)
项目：ouroboros    作者：pybee    | 项目源码 | 文件源码 
def test_recv(self):
        fd = os.open(support.TESTFN, os.O_RDONLY)
        w = asyncore.file_wrapper(fd)
        os.close(fd)

        self.assertNotEqual(w.fd, fd)
        self.assertNotEqual(w.fileno(), fd)
        self.assertEqual(w.recv(13), b"It's not dead")
        self.assertEqual(w.read(6), b", it's")
        w.close()
        self.assertRaises(OSError, w.read, 1)
项目：ouroboros    作者：pybee    | 项目源码 | 文件源码 
def test_send(self):
        d1 = b"Come again?"
        d2 = b"I want to buy some cheese."
        fd = os.open(support.TESTFN, os.O_WRONLY | os.O_APPEND)
        w = asyncore.file_wrapper(fd)
        os.close(fd)

        w.write(d1)
        w.send(d2)
        w.close()
        with open(support.TESTFN, 'rb') as file:
            self.assertEqual(file.read(), self.d + d1 + d2)
项目：ouroboros    作者：pybee    | 项目源码 | 文件源码 
def test_resource_warning(self):
        # Issue #11453
        fd = os.open(support.TESTFN, os.O_RDONLY)
        f = asyncore.file_wrapper(fd)

        os.close(fd)
        with support.check_warnings(('', ResourceWarning)):
            f = None
            support.gc_collect()
项目：ouroboros    作者：pybee    | 项目源码 | 文件源码 
def test_close_twice(self):
        fd = os.open(support.TESTFN, os.O_RDONLY)
        f = asyncore.file_wrapper(fd)
        os.close(fd)

        f.close()
        self.assertEqual(f.fd, -1)
        # calling close twice should not fail
        f.close()
项目：ndk-python    作者：gittor    | 项目源码 | 文件源码 
def test_recv(self):
        fd = os.open(TESTFN, os.O_RDONLY)
        w = asyncore.file_wrapper(fd)
        os.close(fd)

        self.assertNotEqual(w.fd, fd)
        self.assertNotEqual(w.fileno(), fd)
        self.assertEqual(w.recv(13), "It's not dead")
        self.assertEqual(w.read(6), ", it's")
        w.close()
        self.assertRaises(OSError, w.read, 1)
项目：ndk-python    作者：gittor    | 项目源码 | 文件源码 
def test_send(self):
        d1 = "Come again?"
        d2 = "I want to buy some cheese."
        fd = os.open(TESTFN, os.O_WRONLY | os.O_APPEND)
        w = asyncore.file_wrapper(fd)
        os.close(fd)

        w.write(d1)
        w.send(d2)
        w.close()
        self.assertEqual(file(TESTFN).read(), self.d + d1 + d2)
项目：RPKI-toolkit    作者：pavel-odintsov    | 项目源码 | 文件源码 
def init_file_dispatcher(self, fd):
    """
    Kludge to plug asyncore.file_dispatcher into asynchat.  Call from
    subclass's __init__() method, after calling
    PDUChannel.__init__(), and don't read this on a full stomach.
    """

    self.connected = True
    self._fileno = fd
    self.socket = asyncore.file_wrapper(fd)
    self.add_channel()
    flags = fcntl.fcntl(fd, fcntl.F_GETFL, 0)
    flags = flags | os.O_NONBLOCK
    fcntl.fcntl(fd, fcntl.F_SETFL, flags)
项目：RPKI-toolkit    作者：pavel-odintsov    | 项目源码 | 文件源码 
def init_file_dispatcher(self, fd):
    """
    Kludge to plug asyncore.file_dispatcher into asynchat.  Call from
    subclass's __init__() method, after calling
    PDUChannel.__init__(), and don't read this on a full stomach.
    """

    self.connected = True
    self._fileno = fd
    self.socket = asyncore.file_wrapper(fd)
    self.add_channel()
    flags = fcntl.fcntl(fd, fcntl.F_GETFL, 0)
    flags = flags | os.O_NONBLOCK
    fcntl.fcntl(fd, fcntl.F_SETFL, flags)
项目：RPKI-toolkit    作者：pavel-odintsov    | 项目源码 | 文件源码 
def init_file_dispatcher(self, fd):
    """
    Kludge to plug asyncore.file_dispatcher into asynchat.  Call from
    subclass's __init__() method, after calling
    PDUChannel.__init__(), and don't read this on a full stomach.
    """

    self.connected = True
    self._fileno = fd
    self.socket = asyncore.file_wrapper(fd)
    self.add_channel()
    flags = fcntl.fcntl(fd, fcntl.F_GETFL, 0)
    flags = flags | os.O_NONBLOCK
    fcntl.fcntl(fd, fcntl.F_SETFL, flags)
项目：RPKI-toolkit    作者：pavel-odintsov    | 项目源码 | 文件源码 
def init_file_dispatcher(self, fd):
    """
    Kludge to plug asyncore.file_dispatcher into asynchat.  Call from
    subclass's __init__() method, after calling
    PDUChannel.__init__(), and don't read this on a full stomach.
    """

    self.connected = True
    self._fileno = fd
    self.socket = asyncore.file_wrapper(fd)
    self.add_channel()
    flags = fcntl.fcntl(fd, fcntl.F_GETFL, 0)
    flags = flags | os.O_NONBLOCK
    fcntl.fcntl(fd, fcntl.F_SETFL, flags)
项目：RPKI-toolkit    作者：pavel-odintsov    | 项目源码 | 文件源码 
def init_file_dispatcher(self, fd):
    """
    Kludge to plug asyncore.file_dispatcher into asynchat.  Call from
    subclass's __init__() method, after calling
    PDUChannel.__init__(), and don't read this on a full stomach.
    """

    self.connected = True
    self._fileno = fd
    self.socket = asyncore.file_wrapper(fd)
    self.add_channel()
    flags = fcntl.fcntl(fd, fcntl.F_GETFL, 0)
    flags = flags | os.O_NONBLOCK
    fcntl.fcntl(fd, fcntl.F_SETFL, flags)
项目：RPKI-toolkit    作者：pavel-odintsov    | 项目源码 | 文件源码 
def init_file_dispatcher(self, fd):
    """
    Kludge to plug asyncore.file_dispatcher into asynchat.  Call from
    subclass's __init__() method, after calling
    PDUChannel.__init__(), and don't read this on a full stomach.
    """

    self.connected = True
    self._fileno = fd
    self.socket = asyncore.file_wrapper(fd)
    self.add_channel()
    flags = fcntl.fcntl(fd, fcntl.F_GETFL, 0)
    flags = flags | os.O_NONBLOCK
    fcntl.fcntl(fd, fcntl.F_SETFL, flags)
项目：kbe_server    作者：xiaohaoppy    | 项目源码 | 文件源码 
def test_recv(self):
        fd = os.open(support.TESTFN, os.O_RDONLY)
        w = asyncore.file_wrapper(fd)
        os.close(fd)

        self.assertNotEqual(w.fd, fd)
        self.assertNotEqual(w.fileno(), fd)
        self.assertEqual(w.recv(13), b"It's not dead")
        self.assertEqual(w.read(6), b", it's")
        w.close()
        self.assertRaises(OSError, w.read, 1)
项目：kbe_server    作者：xiaohaoppy    | 项目源码 | 文件源码 
def test_send(self):
        d1 = b"Come again?"
        d2 = b"I want to buy some cheese."
        fd = os.open(support.TESTFN, os.O_WRONLY | os.O_APPEND)
        w = asyncore.file_wrapper(fd)
        os.close(fd)

        w.write(d1)
        w.send(d2)
        w.close()
        with open(support.TESTFN, 'rb') as file:
            self.assertEqual(file.read(), self.d + d1 + d2)
项目：kbe_server    作者：xiaohaoppy    | 项目源码 | 文件源码 
def test_resource_warning(self):
        # Issue #11453
        fd = os.open(support.TESTFN, os.O_RDONLY)
        f = asyncore.file_wrapper(fd)

        os.close(fd)
        with support.check_warnings(('', ResourceWarning)):
            f = None
            support.gc_collect()
项目：kbe_server    作者：xiaohaoppy    | 项目源码 | 文件源码 
def test_close_twice(self):
        fd = os.open(support.TESTFN, os.O_RDONLY)
        f = asyncore.file_wrapper(fd)
        os.close(fd)

        f.close()
        self.assertEqual(f.fd, -1)
        # calling close twice should not fail
        f.close()