diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/test_atlocation.py | 7 | ||||
| -rw-r--r-- | tests/test_simple.py | 47 |
2 files changed, 39 insertions, 15 deletions
diff --git a/tests/test_atlocation.py b/tests/test_atlocation.py index 5d7286a..b107975 100644 --- a/tests/test_atlocation.py +++ b/tests/test_atlocation.py @@ -95,13 +95,8 @@ class AtLocationTest(unittest.TestCase): if filetype == "symlink" and loctype != "emptypath": follow_symlinks_values.append(False) for follow_symlinks in follow_symlinks_values: - # Mypy fails to see that loctype and filetype really are - # literals rather than arbitrary strings. atlocctx = self.create( - loctype, # type: ignore[arg-type] - filetype, # type: ignore[arg-type] - "X", - follow_symlinks, + loctype, filetype, "X", follow_symlinks ) yield (filetype, atlocctx) diff --git a/tests/test_simple.py b/tests/test_simple.py index eb03384..114b922 100644 --- a/tests/test_simple.py +++ b/tests/test_simple.py @@ -5,6 +5,7 @@ import asyncio import errno import os import pathlib +import signal import socket import unittest @@ -92,6 +93,26 @@ class AsnycioTest(unittest.IsolatedAsyncioTestCase): efd.write() self.assertEqual(await fut, 1) + async def test_signalfd(self) -> None: + testsig = signal.SIGUSR1 + sfd = linuxnamespaces.SignalFD( + [testsig], linuxnamespaces.SignalFDFlags.NONBLOCK + ) + self.addCleanup(sfd.close) + oldmask = signal.pthread_sigmask(signal.SIG_SETMASK, [testsig]) + self.addCleanup(signal.pthread_sigmask, signal.SIG_SETMASK, oldmask) + fut = asyncio.ensure_future(sfd.aread()) + await asyncio.sleep(0.000001) # Let the loop run + self.assertFalse(fut.done()) + sigval = 123456789 + mypid = os.getpid() + linuxnamespaces.sigqueue(mypid, testsig, sigval) + siginfo = await fut + self.assertEqual(siginfo.ssi_signo, testsig) + self.assertEqual(siginfo.ssi_pid, mypid) + self.assertEqual(siginfo.ssi_uid, os.getuid()) + self.assertEqual(siginfo.ssi_int, sigval) + async def test_run_in_fork(self) -> None: with linuxnamespaces.EventFD( 0, linuxnamespaces.EventFDFlags.NONBLOCK @@ -105,9 +126,13 @@ class AsnycioTest(unittest.IsolatedAsyncioTestCase): async def test_copyfd_file_sock(self) -> None: sock1, sock2 = socket.socketpair() - with sock1, sock2, linuxnamespaces.FileDescriptor( - os.open("/etc/passwd", os.O_RDONLY) - ) as rfd: + with ( + sock1, + sock2, + linuxnamespaces.FileDescriptor( + os.open("/etc/passwd", os.O_RDONLY) + ) as rfd, + ): fut = asyncio.ensure_future( linuxnamespaces.async_copyfd(rfd, sock1.fileno(), 999) ) @@ -118,9 +143,13 @@ class AsnycioTest(unittest.IsolatedAsyncioTestCase): async def test_copyfd_file_pipe(self) -> None: rfdp, wfdp = linuxnamespaces.FileDescriptor.pipe(blocking=False) - with rfdp, wfdp, linuxnamespaces.FileDescriptor( - os.open("/etc/passwd", os.O_RDONLY) - ) as rfd: + with ( + rfdp, + wfdp, + linuxnamespaces.FileDescriptor( + os.open("/etc/passwd", os.O_RDONLY) + ) as rfd, + ): fut = asyncio.ensure_future( linuxnamespaces.async_copyfd(rfd, wfdp, 999) ) @@ -185,7 +214,7 @@ class UnshareTest(unittest.TestCase): | linuxnamespaces.CloneFlags.NEWPID ) linuxnamespaces.newuidmap(-1, [idmap], False) - @linuxnamespaces.run_in_fork + @linuxnamespaces.run_in_fork.now def setup() -> None: self.assertEqual(os.getpid(), 1) linuxnamespaces.mount("proc", "/proc", "proc") @@ -210,7 +239,7 @@ class UnshareTest(unittest.TestCase): ) linuxnamespaces.mount("tmpfs", "/mnt", "tmpfs", data="mode=0755") os.mkdir("/mnt/dev") - linuxnamespaces.populate_dev("/", "/mnt", pidns=False) + linuxnamespaces.populate_dev("/", "/mnt", pts="host") self.assertTrue(os.access("/mnt/dev/null", os.W_OK)) pathlib.Path("/mnt/dev/null").write_text("") @@ -278,7 +307,7 @@ class UnshareIdmapTest(unittest.TestCase): os.setregid(0, 0) linuxnamespaces.mount("tmpfs", "/mnt", "tmpfs") os.mkdir("/mnt/dev") - @linuxnamespaces.run_in_fork + @linuxnamespaces.run_in_fork.now def test() -> None: linuxnamespaces.populate_dev("/", "/mnt") test() |