Module rsyscall.tests.test_fd

Expand source code Browse git 
from rsyscall import local_thread
from rsyscall.fcntl import O
from rsyscall.sched import CLONE
from rsyscall.tests.trio_test_case import TrioTestCase
import gc

class TestFS(TrioTestCase):
    async def asyncSetUp(self) -> None:
        self.thr = local_thread

    async def test_fd_gc(self) -> None:
        "Verify that file descriptors actually get GC'd."
        gc.collect()
        await self.thr.task.run_fd_table_gc()
        devnull = await self.thr.ptr("/dev/null")
        first = int(await self.thr.task.open(devnull, O.RDONLY))
        for _ in range(5):
            child = await self.thr.clone(CLONE.FILES)
            for _ in range(50):
                await child.task.open(devnull, O.RDONLY)
        gc.collect()
        await self.thr.task.run_fd_table_gc()
        last = int(await self.thr.task.open(devnull, O.RDONLY))
        self.assertEqual(first, last)

Classes

class TestFS (methodName='runTest')

A trio-enabled variant of unittest.TestCase

Create an instance of the class that will use the named test method when executed. Raises a ValueError if the instance does not have a method with the specified name.

Expand source code Browse git 
class TestFS(TrioTestCase):
    async def asyncSetUp(self) -> None:
        self.thr = local_thread

    async def test_fd_gc(self) -> None:
        "Verify that file descriptors actually get GC'd."
        gc.collect()
        await self.thr.task.run_fd_table_gc()
        devnull = await self.thr.ptr("/dev/null")
        first = int(await self.thr.task.open(devnull, O.RDONLY))
        for _ in range(5):
            child = await self.thr.clone(CLONE.FILES)
            for _ in range(50):
                await child.task.open(devnull, O.RDONLY)
        gc.collect()
        await self.thr.task.run_fd_table_gc()
        last = int(await self.thr.task.open(devnull, O.RDONLY))
        self.assertEqual(first, last)

Ancestors

Class variables

var nursery : trio.Nursery

Methods

async def test_fd_gc(self) ‑> NoneType

Verify that file descriptors actually get GC'd.

Expand source code Browse git 
async def test_fd_gc(self) -> None:
    "Verify that file descriptors actually get GC'd."
    gc.collect()
    await self.thr.task.run_fd_table_gc()
    devnull = await self.thr.ptr("/dev/null")
    first = int(await self.thr.task.open(devnull, O.RDONLY))
    for _ in range(5):
        child = await self.thr.clone(CLONE.FILES)
        for _ in range(50):
            await child.task.open(devnull, O.RDONLY)
    gc.collect()
    await self.thr.task.run_fd_table_gc()
    last = int(await self.thr.task.open(devnull, O.RDONLY))
    self.assertEqual(first, last)

Inherited members