Fix BlockIO issues

.github/workflows/build.yaml

@@ -9,9 +9,30 @@ on:
     types: [released]
 permissions: read-all
 jobs:
+  test-image:
+    name: Generate test ext4 image
+    runs-on: ubuntu-latest
+    steps:
+      - name: Generate test.ext4
+        shell: bash
+        run: |
+          set -e
+          mkdir test
+          echo "hello world" > test/test.txt
+          for i in {1..100};do
+            echo "hello world" >> test/test.txt
+          done
+          dd if=/dev/zero of=test.ext4 count=1024 bs=1024
+          mkfs.ext4 test.ext4 -d test
+      - uses: actions/upload-artifact@v4
+        with:
+          name: test.ext4
+          path: test.ext4
+          if-no-files-found: error
   test:
     name: Test on ${{ matrix.os }} python ${{ matrix.python }}
     runs-on: ${{ matrix.os }}
+    needs: [test-image]
     strategy:
       fail-fast: false
       matrix:
@@ -20,17 +41,22 @@ jobs:
           - windows-latest
           - macos-latest
         python:
-          - '3.11'
-          - '3.12'
-          - '3.13'
+          - "3.11"
+          - "3.12"
+          - "3.13"
     steps:
       - name: Checkout the Git repository
         uses: actions/checkout@v4
+      - name: Download test.ext4
+        uses: actions/download-artifact@v4
+        with:
+          name: test.ext4
+          path: .
       - name: Setup Python
         uses: actions/setup-python@v5
         with:
           python-version: ${{ matrix.python }}
-          cache: 'pip'
+          cache: "pip"
       - name: Run test
         shell: bash
         run: ./test.sh
@@ -44,8 +70,8 @@ jobs:
       - name: Setup Python
         uses: actions/setup-python@v5
         with:
-          python-version: '3.11'
-          cache: 'pip'
+          python-version: "3.11"
+          cache: "pip"
       - name: Install build tool
         run: pip install build
       - name: Building package
@@ -94,8 +120,7 @@ jobs:
           name: pip
           path: dist
       - name: Upload to release
-        run:
-          find . -type f | xargs -rI {} gh release upload "$TAG" {} --clobber
+        run: find . -type f | xargs -rI {} gh release upload "$TAG" {} --clobber
         env:
           GH_TOKEN: ${{ github.token }}
           TAG: ${{ github.event.release.tag_name }}

.gitignore

@@ -158,3 +158,5 @@ cython_debug/
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 #.idea/
+
+*.ext4

ext4/_compat.py

@@ -0,0 +1,12 @@
+try:
+    from typing import override
+
+except ImportError:
+    from typing import Callable
+    from typing import Any
+
+    def override(fn: Callable[..., Any]):
+        return fn
+
+
+__all__ = ["override"]

ext4/block.py

@@ -1,6 +1,8 @@
 import io
 import errno
 
+from ._compat import override
+
 
 class BlockIOBlocks(object):
     def __init__(self, blockio):
@@ -45,8 +47,8 @@ class BlockIO(io.RawIOBase):
     def __init__(self, inode):
         super().__init__()
         self.inode = inode
-        self.cursor = 0
-        self.blocks = BlockIOBlocks(self)
+        self.cursor: int = 0
+        self.blocks: BlockIOBlocks = BlockIOBlocks(self)
 
     def __len__(self):
         return self.inode.i_size
@@ -56,10 +58,19 @@ def extents(self):
         return self.inode.extents
 
     @property
-    def block_size(self):
+    def block_size(self) -> int:
         return self.inode.volume.block_size
 
-    def seek(self, offset, mode=io.SEEK_SET):
+    @override
+    def readable(self) -> bool:
+        return True
+
+    @override
+    def seekable(self) -> bool:
+        return True
+
+    @override
+    def seek(self, offset: int, mode: int = io.SEEK_SET) -> int:
         if mode == io.SEEK_CUR:
             offset += self.cursor
 
@@ -73,11 +84,14 @@ def seek(self, offset, mode=io.SEEK_SET):
             raise OSError(errno.EINVAL, "Invalid argument")
 
         self.cursor = offset
+        return offset
 
-    def tell(self):
+    @override
+    def tell(self) -> int:
         return self.cursor
 
-    def read(self, size=-1):
+    @override
+    def read(self, size: int = -1) -> bytes:
         if size < 0:
             size = len(self) - self.cursor
 
@@ -88,7 +102,7 @@ def read(self, size=-1):
 
         return data
 
-    def peek(self, size=0):
+    def peek(self, size: int = 0) -> bytes:
         if self.cursor >= len(self):
             return b""
 

ext4/inode.py

@@ -7,6 +7,8 @@
 from ctypes import c_uint16
 from ctypes import sizeof
 
+from ._compat import override
+
 from .struct import Ext4Struct
 from .struct import crc32c
 from .struct import MagicError
@@ -264,7 +266,7 @@ def validate(self):
             self.tree.validate()
 
     @property
-    def is_inline(self):
+    def is_inline(self) -> bool:
         return (self.i_flags & EXT4_FL.EXTENTS) == 0
 
     @property
@@ -279,7 +281,7 @@ def headers(self):
     def indices(self):
         return self.tree.indices
 
-    def _open(self, mode="rb", encoding=None, newline=None):
+    def _open(self, mode: str = "rb", encoding: None = None, newline: None = None):
         if mode != "rb" or encoding is not None or newline is not None:
             raise NotImplementedError()
 
@@ -339,7 +341,10 @@ class Socket(Inode):
 
 
 class File(Inode):
-    def open(self, mode="rb", encoding=None, newline=None):
+    @override
+    def open(
+        self, mode: str = "rb", encoding: None = None, newline: None = None
+    ) -> io.BytesIO | BlockIO:
         return self._open(mode, encoding, newline)
 
 

ext4/volume.py

@@ -172,14 +172,14 @@ def block_read(self, index, count=1):
         return self.read(count * block_size)
 
     @staticmethod
-    def path_tuple(path):
+    def path_tuple(path: str | bytes) -> tuple[bytes, ...]:
         if isinstance(path, bytes):
             path = path.decode("utf-8")
 
         return tuple(x.encode("utf-8") for x in PurePosixPath(path).parts[1:])
 
     @cached(cache=LRUCache(maxsize=32))
-    def inode_at(self, path):
+    def inode_at(self, path: str | bytes) -> Inode:
         paths = list(self.path_tuple(path))
         cwd = self.root
         if not paths:

test.py

@@ -1,10 +1,12 @@
 import sys
 import ext4
 
+from typing import cast
+
 FAILED = False
 
 
-def test_path_tuple(path, expected):
+def test_path_tuple(path: str | bytes, expected: tuple[bytes, ...]):
     global FAILED
     print(f"check Volume.path_tuple({path}): ", end="")
     try:
@@ -21,12 +23,36 @@ def test_path_tuple(path, expected):
         print(e)
 
 
+def _assert(source: str):
+    global FAILED
+    print(f"check {source}: ", end="")
+    if eval(source):
+        print("pass")
+        return
+
+    FAILED = True
+    print("fail")
+
+
 test_path_tuple("/", tuple())
 test_path_tuple(b"/", tuple())
 test_path_tuple("/test", (b"test",))
 test_path_tuple(b"/test", (b"test",))
 test_path_tuple("/test/test", (b"test", b"test"))
 test_path_tuple(b"/test/test", (b"test", b"test"))
 
+with open("test.ext4", "rb") as f:
+    # Extract specific file
+    volume = ext4.Volume(f, offset=0)
+    inode = cast(ext4.File, volume.inode_at("/test.txt"))
+    _assert("isinstance(inode, ext4.File)")
+    b = inode.open()
+    _assert("isinstance(b, ext4.BlockIO)")
+    _assert("b.readable()")
+    _assert("b.seekable()")
+    _assert("b.seek(1) == 1")
+    _assert("b.seek(0) == 0")
+    _assert("b.seek(10) == 10")
+
 if FAILED:
     sys.exit(1)

test.sh

@@ -13,4 +13,15 @@ python -m pip install wheel
 python -m pip install \
   --extra-index-url=https://wheels.eeems.codes/ \
   -r requirements.txt
+if ! [ -f test.ext4 ];then
+  tmp_dir=$(mktemp -d)
+  trap "rm -r \"$tmp_dir\"" EXIT
+  echo "hello world" > "$tmp_dir"/test.txt
+  for i in {1..100};do
+    echo "hello world" >> "$tmp_dir"/test.txt
+  done
+  dd if=/dev/zero of=test.ext4 count=1024 bs=1024
+  trap "rm test.ext4" EXIT
+  mkfs.ext4 test.ext4 -d "$tmp_dir"
+fi
 python test.py