Generate htree image for testing (#21)

* Generate htree image for testing

* Add more htree testing and better typing

* Get tests working

* Add inode_at to directories as well

* Add inode_at to directories as well

* Fix tests

* Fix lint

* Review feedback

* Review feedback

Author: Eeems

.github/workflows/build.yaml

@@ -47,6 +47,7 @@ jobs:
             test32.ext4.tmp
             test64.ext4
             test64.ext4.tmp
+            test_htree.ext4
           if-no-files-found: error
   test:
     name: Test on ${{ matrix.os }} python ${{ matrix.python }}

.gitignore

@@ -160,3 +160,4 @@ cython_debug/
 #.idea/
 
 *.ext4
+*.ext4.tmp

_test_image.sh

@@ -1,23 +1,60 @@
 #!/bin/bash
 set -e
 
+if ! command -v chronic &>/dev/null; then
+  chronic() {
+    "$@"
+  }
+fi
+
+mkimage() {
+  name="$1"
+  shift
+  dir="$1"
+  shift
+  size="$1"
+  shift
+
+  echo "[test] Making image $name < $dir..."
+  chronic dd if=/dev/zero of="$name".ext4.tmp count="$size" bs=1048576
+  chronic mkfs.ext4 -g 1024 "$@" "$name".ext4.tmp -d "$dir"
+  echo -n F >"$name".ext4
+  cat "$name".ext4.tmp >>"$name".ext4
+}
+
 tmp_dir=$(mktemp -d)
+# shellcheck disable=SC2064
 trap "rm -r \"$tmp_dir\"" EXIT
-echo "hello world" > "$tmp_dir"/test.txt
-for i in {1..1000};do
-  echo "hello world" >> "$tmp_dir"/test.txt
-done
-for i in {1..100};do
-  echo "hello world$i" > "$tmp_dir"/test$i.txt
-  for j in {1..20};do
-    setfattr -n user.name$j -v value${i}_$j "$tmp_dir"/test$i.txt
+echo "[test] Using temporary directory: $tmp_dir"
+echo "[test] Generating files..."
+echo "hello world" >"$tmp_dir"/test.txt
+for i in {1..1000}; do
+  echo "echo "hello world" >>'$tmp_dir/test.txt'"
+done | xargs -P "$(nproc)" -I {} bash -c '{}'
+for i in {1..100}; do
+  echo "echo 'hello world$i' >'$tmp_dir/test$i.txt'"
+done | xargs -P "$(nproc)" -I {} bash -c '{}'
+for i in {1..100}; do
+  for j in {1..20}; do
+    echo "setfattr -n 'user.name$j' -v 'value${i}_$j' '$tmp_dir/test$i.txt'"
   done
-done
-dd if=/dev/zero of=test32.ext4.tmp count=20 bs=1048576
-dd if=/dev/zero of=test64.ext4.tmp count=20 bs=1048576
-mkfs.ext4 -g 1024 -O 64bit test64.ext4.tmp -d "$tmp_dir"
-mkfs.ext4 -g 1024 -O ^64bit test32.ext4.tmp -d "$tmp_dir"
-echo -n F > test32.ext4
-cat test32.ext4.tmp >> test32.ext4
-echo -n F > test64.ext4
-cat test64.ext4.tmp >> test64.ext4
+done | xargs -P "$(nproc)" -I {} bash -c '{}'
+
+mkimage test32 "$tmp_dir" 20 -O ^64bit
+mkimage test64 "$tmp_dir" 20 -O 64bit
+
+rm -f "$tmp_dir"/test*.txt
+echo "[test] Generating files..."
+
+echo "[test] Making image test_htree..."
+chronic dd if=/dev/zero of=test_htree.ext4 count=20 bs=1048576
+chronic mkfs.ext4 -g 1024 -b 1024 -O 64bit,dir_index test_htree.ext4
+sudo mount -t ext4 test_htree.ext4 "$tmp_dir"
+# shellcheck disable=SC2064
+trap "sudo umount \"$tmp_dir\";rmdir \"$tmp_dir\"" EXIT
+sudo mkdir "$tmp_dir"/empty
+printf '%s\n' "$tmp_dir"/{1..200} | xargs sudo touch
+sudo umount "$tmp_dir"
+# shellcheck disable=SC2064
+trap "rmdir \"$tmp_dir\"" EXIT
+chronic e2fsck -Dy test_htree.ext4

ext4/__init__.py

@@ -69,6 +69,9 @@
 from .xattr import ExtendedAttributeEntry
 
 from .htree import DXRoot
+from .htree import DotDirectoryEntry2
+from .htree import DXEntry
+from .htree import DXRootInfo
 
 __all__ = [
     "DX_HASH",
@@ -132,4 +135,7 @@
     "ExtendedAttributeHeader",
     "ExtendedAttributeEntry",
     "DXRoot",
+    "DotDirectoryEntry2",
+    "DXEntry",
+    "DXRootInfo",
 ]

ext4/htree.py

@@ -11,7 +11,6 @@
 from ctypes import LittleEndianStructure
 
 from typing import final
-from typing import cast
 from typing import TYPE_CHECKING
 
 from collections.abc import Generator
@@ -28,7 +27,18 @@
 
 
 class LittleEndianStructureWithVolume(LittleEndianStructure):
-    volume: "Volume | None" = None
+    def __init__(self):
+        super().__init__()
+        self._volume: "Volume | None" = None
+
+    @property
+    def volume(self) -> "Volume":
+        assert self._volume is not None
+        return self._volume
+
+    @volume.setter
+    def volume(self, volume: "Volume") -> None:
+        self._volume = volume
 
 
 @final
@@ -45,7 +55,7 @@ class DotDirectoryEntry2(LittleEndianStructureWithVolume):
 
     def verify(self) -> None:
         name = assert_cast(self.name, bytes)  # pyright: ignore[reportAny]
-        if name in (b".\0\0\0", b"..\0\0"):
+        if name in (b".", b".."):
             return
 
         message = f"{self} dot or dotdot entry name invalid! actual={name}"
@@ -141,8 +151,6 @@ class DXRoot(DXEntriesBase):
 
     def __init__(self, inode: "Directory"):
         super().__init__(inode, 0)
-        cast(DotDirectoryEntry2, self.dot).volume = inode.volume
-        cast(DotDirectoryEntry2, self.dotdot).volume = inode.volume
 
 
 @final

ext4/inode.py

@@ -2,6 +2,8 @@
 from __future__ import annotations
 
 import io
+import os
+import errno
 import warnings
 
 from ctypes import LittleEndianStructure
@@ -12,8 +14,12 @@
 
 from typing import cast
 from typing import final
+from typing import Any
 from typing import TYPE_CHECKING
 
+from cachetools import cachedmethod
+from cachetools import LRUCache
+
 from ._compat import override
 from ._compat import ReadableStream
 from ._compat import assert_cast
@@ -437,9 +443,11 @@ def readlink(self):
 class Directory(Inode):
     def __init__(self, volume: "Volume", offset: int, i_no: int):
         super().__init__(volume, offset, i_no)
+        self._inode_at_cache: LRUCache[str | bytes, Inode] = LRUCache(maxsize=32)
         self._dirents: None | list[DirectoryEntry | DirectoryEntry2] = None
+        self.htree: DXRoot | None = None
         if self.is_htree:
-            self.htree: DXRoot | None = DXRoot(self)
+            self.htree = DXRoot(self)
 
     @override
     def verify(self):
@@ -471,7 +479,9 @@ def is_encrypted(self) -> bool:
     def hash_in_dirent(self) -> bool:
         return self.is_casefolded and self.is_encrypted
 
-    def _opendir(self):
+    def _opendir(
+        self,
+    ) -> Generator[DirectoryEntry | DirectoryEntry2, None, None]:
         if self._dirents is not None:
             for dirent in self._dirents:
                 yield dirent
@@ -554,7 +564,9 @@ def _get_file_type(self, dirent: DirectoryEntry | DirectoryEntry2) -> EXT4_FT:
             f"Unexpected file type {file_type} for inode {dirent_inode}"
         )
 
-    def opendir(self):
+    def opendir(
+        self,
+    ) -> Generator[tuple[DirectoryEntry | DirectoryEntry2, EXT4_FT], Any, None]:  # pyright: ignore[reportExplicitAny]
         for dirent in self._opendir():
             if isinstance(dirent, DirectoryEntry2):
                 file_type = assert_cast(dirent.file_type, EXT4_FT)  # pyright: ignore[reportAny]
@@ -568,3 +580,37 @@ def opendir(self):
                 file_type = self._get_file_type(dirent)
 
             yield dirent, file_type
+
+    @cachedmethod(lambda self: self._inode_at_cache)  # pyright: ignore[reportAny]
+    def inode_at(self, path: str | bytes) -> Inode:
+        if (isinstance(path, str) and path.startswith("/")) or (
+            isinstance(path, bytes) and path.startswith(b"/")
+        ):
+            return self.volume.inode_at(path)
+
+        if isinstance(path, bytes):
+            path = path.decode("utf-8")
+
+        paths = list(self.volume.path_tuple(f"/{path}"))
+        cwd = self
+        if not paths:
+            return cwd
+
+        while paths:
+            if not isinstance(cwd, Directory):
+                raise OSError(errno.ENOTDIR, os.strerror(errno.ENOTDIR))
+
+            name = paths.pop(0)
+            inode = None
+            for dirent, _ in cwd.opendir():
+                if dirent.name_bytes == name:
+                    dirent_inode = assert_cast(dirent.inode, int)  # pyright: ignore[reportAny]
+                    inode = self.volume.inodes[dirent_inode]
+                    break
+
+            if inode is None:
+                raise FileNotFoundError(path)
+
+            cwd = inode
+
+        return cwd

ext4/volume.py

@@ -119,8 +119,8 @@ def bad_blocks(self):
         return self.inodes[EXT4_INO.BAD]
 
     @property
-    def root(self):
-        return self.inodes[EXT4_INO.ROOT]
+    def root(self) -> Directory:
+        return assert_cast(self.inodes[EXT4_INO.ROOT], Directory)
 
     @property
     def user_quota(self):