diff options
| -rw-r--r-- | tests/test_simple.py | 40 |
1 files changed, 40 insertions, 0 deletions
diff --git a/tests/test_simple.py b/tests/test_simple.py index 62facf5..b3331a3 100644 --- a/tests/test_simple.py +++ b/tests/test_simple.py @@ -2,6 +2,7 @@ # SPDX-License-Identifier: GPL-3 import asyncio +import errno import os import pathlib import socket @@ -74,6 +75,45 @@ class AsnycioTest(unittest.IsolatedAsyncioTestCase): await set_ready() await asyncio.wait_for(fut, 10) + async def test_copyfd(self) -> None: + rfd1, wfd1 = os.pipe2(os.O_NONBLOCK) + rfd2, wfd2 = os.pipe2(os.O_NONBLOCK) + fut = asyncio.ensure_future(linuxnamespaces.async_copyfd(rfd1, wfd2)) + os.write(wfd1, b"hello") + await asyncio.sleep(0.000001) # Let the loop run + os.write(wfd1, b"world") + loop = asyncio.get_running_loop() + fut2 = loop.create_future() + def callback() -> None: + loop.remove_reader(rfd2) + fut2.set_result(None) + loop.add_reader(rfd2, callback) + await fut2 + self.assertEqual(os.read(rfd2, 11), b"helloworld") + self.assertFalse(fut.done()) + os.close(wfd1) + await asyncio.sleep(0.000001) # Let the loop run + self.assertTrue(fut.done()) + os.close(rfd1) + os.close(rfd2) + os.close(wfd2) + self.assertEqual(await fut, 10) + + async def test_copyfd_epipe(self) -> None: + rfd1, wfd1 = os.pipe2(os.O_NONBLOCK) + rfd2, wfd2 = os.pipe2(os.O_NONBLOCK) + fut = asyncio.ensure_future(linuxnamespaces.async_copyfd(rfd1, wfd2)) + os.close(rfd2) + os.write(wfd1, b"hello") + await asyncio.sleep(0.000001) # Let the loop run + self.assertTrue(fut.done()) + os.close(rfd1) + os.close(wfd1) + os.close(wfd2) + exc = fut.exception() + self.assertIsInstance(exc, OSError) + self.assertEqual(exc.errno, errno.EPIPE) + class UnshareTest(unittest.TestCase): @pytest.mark.forked |