test: init exit

Author: YoSTEALTH

pyproject.toml

@@ -37,6 +37,9 @@ Issues = "https://github.com/YoSTEALTH/Liburing/issues"
 [project.optional-dependencies]
 test = ["pytest"]
 
+[tool.ruff]
+line-length = 120
+
 [tool.pyoz]
 # Path to your Zig source file
 module-path = "src/liburing/root.zig"

src/liburing/class.zig

@@ -157,10 +157,8 @@ pub const Timespec = extern struct {
 pub const Ring = extern struct {
     _io_uring: ?*c.io_uring,
 
-    pub fn __new__() ?Ring {
-        const io_uring: *c.io_uring = std.heap.c_allocator.create(c.io_uring) catch {
-            return oz.raiseMemoryError("`Ring()` - Out of Memory!");
-        };
+    pub fn __new__() !Ring {
+        const io_uring: *c.io_uring = try std.heap.c_allocator.create(c.io_uring);
         io_uring.* = std.mem.zeroes(c.io_uring); // set default value to `0`
         return .{ ._io_uring = io_uring };
     }
@@ -261,6 +259,8 @@ pub const Cqe = extern struct {
     }
 };
 
+// TODO: Merge `CqeIter` into `Cqe`
+
 ///Example
 ///    >>> for i in CqeIter(ring, cqe):
 ///    >>>     cqe[0].user_data == i  # do stuff

test/conftest.py

@@ -77,11 +77,7 @@ def skip_by_platform(request):
     """
     if r := request.node.get_closest_marker("skip_linux"):
         if liburing.linux_version_check(version := r.args[0]):
-            msg = (
-                r.args[1]
-                if len(r.args) > 1
-                else f"Kernel `{LINUX_VERSION} < {version}`"
-            )
+            msg = r.args[1] if len(r.args) > 1 else f"Kernel `{LINUX_VERSION} < {version}`"
             pytest.skip(msg)
 
 

test/queue/init_exit_test.py

@@ -0,0 +1,85 @@
+import pytest
+import liburing
+
+
+def test_io_uring_init_exit():
+    ring = liburing.Ring()
+    assert ring.flags == 0
+    assert ring.ring_fd == 0
+    assert ring.features == 0
+    assert ring.enter_ring_fd == 0
+    assert ring.int_flags == 0
+
+    with pytest.raises(TypeError):
+        liburing.io_uring_queue_init(-1, ring, 0)
+
+    with pytest.raises(TypeError):
+        liburing.io_uring_queue_init(8, ring, -1)
+
+    assert liburing.io_uring_queue_init(1, ring, 0) == 0
+    assert ring.ring_fd > -1
+    assert ring.enter_ring_fd > -1
+    with pytest.raises(RuntimeError):  # Trying to init again
+        liburing.io_uring_queue_init(4, ring, 0)
+    assert liburing.io_uring_queue_exit(ring) is None
+    # Trying to exit again
+    with pytest.raises(RuntimeError):
+        liburing.io_uring_queue_exit(ring) is None  # will not call internally but
+
+    ring = liburing.Ring()
+    with pytest.raises(RuntimeError):
+        liburing.io_uring_queue_exit(ring)
+
+
+def test_setup_iopoll_sqpoll():
+    for i, setup in enumerate((liburing.IORING_SETUP_IOPOLL, liburing.IORING_SETUP_SQPOLL), 1):
+        ring = liburing.Ring()
+        assert liburing.io_uring_queue_init(1, ring, setup) == 0
+        assert liburing.io_uring_queue_exit(ring) is None
+
+
+def test_max_entries_plus():
+    max_plus = 32768 + 1
+    ring = liburing.Ring()
+    with pytest.raises(OSError):  # OSError: [Errno 22] Invalid argument
+        liburing.io_uring_queue_init(max_plus, ring)
+    with pytest.raises(RuntimeError):
+        liburing.io_uring_queue_exit(ring)
+
+
+def test_init_max():
+    ring = liburing.Ring()
+    cqe = liburing.Cqe()
+    maximum = 32768
+    liburing.io_uring_queue_init(maximum, ring)
+
+    for i in range(maximum - 1):
+        sqe = liburing.io_uring_get_sqe(ring)
+        liburing.io_uring_prep_nop(sqe)
+        sqe.flags = liburing.IOSQE_IO_LINK | liburing.IOSQE_ASYNC
+        sqe.user_data = i + 1  # `sqe.user_data` can not be `0`
+    else:  # last
+        sqe = liburing.io_uring_get_sqe(ring)
+        liburing.io_uring_prep_nop(sqe)
+        sqe.flags = liburing.IOSQE_ASYNC
+        sqe.user_data = i + 1 + 1
+
+    assert liburing.io_uring_submit_and_wait_timeout(ring, cqe, maximum) == maximum
+
+    for i in range(maximum):
+        assert cqe[i].user_data == i + 1
+
+    liburing.io_uring_cq_advance(ring, maximum)
+    with pytest.raises(BlockingIOError):
+        liburing.io_uring_peek_cqe(ring, cqe)
+    liburing.io_uring_queue_exit(ring)
+
+
+def test_io_uring_submit_and_wait(ring, cqe):
+    sqe = liburing.io_uring_get_sqe(ring)
+    sqe.user_data = 123
+    assert liburing.io_uring_sq_ready(ring) == 1
+    assert liburing.io_uring_submit_and_wait(ring, 1) == 1
+    assert liburing.io_uring_sq_ready(ring) == 0
+    liburing.io_uring_peek_cqe(ring, cqe)
+    assert cqe[0].user_data == 123