Module rsyscall.tests.test_inotify
Expand source code Browse git
from rsyscall.tests.trio_test_case import TrioTestCase
from rsyscall import local_thread
from rsyscall.sys.inotify import *
from rsyscall.inotify_watch import Inotify
from rsyscall.fcntl import O
from rsyscall.stdlib import mkdtemp
class TestInotify(TrioTestCase):
async def asyncSetUp(self) -> None:
self.thr = local_thread
self.tmpdir = await mkdtemp(self.thr)
self.path = self.tmpdir.path
self.ify = await Inotify.make(self.thr)
async def asyncTearDown(self) -> None:
await self.tmpdir.cleanup()
async def test_create(self) -> None:
watch = await self.ify.add(self.path, IN.CREATE)
name = "foo"
fd = await self.thr.task.open(await self.thr.ram.ptr(self.path/name), O.CREAT|O.EXCL)
event = await watch.wait_until_event(IN.CREATE, name)
self.assertEqual(event.name, name)
self.assertEqual(event.mask, IN.CREATE)
Classes
class TestInotify (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 TestInotify(TrioTestCase): async def asyncSetUp(self) -> None: self.thr = local_thread self.tmpdir = await mkdtemp(self.thr) self.path = self.tmpdir.path self.ify = await Inotify.make(self.thr) async def asyncTearDown(self) -> None: await self.tmpdir.cleanup() async def test_create(self) -> None: watch = await self.ify.add(self.path, IN.CREATE) name = "foo" fd = await self.thr.task.open(await self.thr.ram.ptr(self.path/name), O.CREAT|O.EXCL) event = await watch.wait_until_event(IN.CREATE, name) self.assertEqual(event.name, name) self.assertEqual(event.mask, IN.CREATE)
Ancestors
- TrioTestCase
- unittest.case.TestCase
Class variables
var nursery : trio.Nursery
Methods
async def test_create(self) ‑> NoneType
-
Expand source code Browse git
async def test_create(self) -> None: watch = await self.ify.add(self.path, IN.CREATE) name = "foo" fd = await self.thr.task.open(await self.thr.ram.ptr(self.path/name), O.CREAT|O.EXCL) event = await watch.wait_until_event(IN.CREATE, name) self.assertEqual(event.name, name) self.assertEqual(event.mask, IN.CREATE)
Inherited members