Improve benchmarks and add more (#77)

Author: Schamper

pyproject.toml

@@ -46,7 +46,7 @@ dump-nskeyedarchiver = "dissect.util.tools.dump_nskeyedarchiver:main"
 
 [tool.ruff]
 line-length = 120
-required-version = ">=0.9.0"
+required-version = ">=0.11.0"
 
 [tool.ruff.format]
 docstring-code-format = true

tests/conftest.py

@@ -0,0 +1,17 @@
+import importlib.util
+
+import pytest
+
+HAS_BENCHMARK = importlib.util.find_spec("pytest_benchmark") is not None
+
+
+def pytest_configure(config: pytest.Config) -> None:
+    if not HAS_BENCHMARK:
+        # If we don't have pytest-benchmark (or pytest-codspeed) installed, register the benchmark marker ourselves
+        # to avoid pytest warnings
+        config.addinivalue_line("markers", "benchmark: mark test for benchmarking (requires pytest-benchmark)")
+
+
+def pytest_runtest_setup(item: pytest.Item) -> None:
+    if not HAS_BENCHMARK and item.get_closest_marker("benchmark") is not None:
+        pytest.skip("pytest-benchmark is not installed")

tests/test_compression.py

@@ -1,11 +1,17 @@
+from __future__ import annotations
+
 import hashlib
 import lzma
 from io import BytesIO
+from typing import TYPE_CHECKING
 
 import pytest
 
 from dissect.util.compression import lz4, lznt1, lzo, lzxpress, lzxpress_huffman, sevenbit, xz
 
+if TYPE_CHECKING:
+    from pytest_benchmark.fixture import BenchmarkFixture
+
 
 def test_lz4_decompress() -> None:
     assert (
@@ -26,6 +32,19 @@ def test_lznt1_decompress() -> None:
     )
 
 
+@pytest.mark.benchmark
+def test_benchmark_lznt1_decompress(benchmark: BenchmarkFixture) -> None:
+    buf = bytes.fromhex(
+        "38b08846232000204720410010a24701a045204400084501507900c045200524"
+        "138805b4024a44ef0358028c091601484500be009e000401189000"
+    )
+    assert benchmark(lznt1.decompress, buf) == (
+        b"F# F# G A A G F# E D D E F# F# E E F# F# G A A "
+        b"G F# E D D E F# E D D E E F# D E F# G F# D E F# "
+        b"G F# E D E A F# F# G A A G F# E D D E F# E D D\x00"
+    )
+
+
 def test_lzo_decompress() -> None:
     assert (
         lzo.decompress(bytes.fromhex("0361626361626320f314000f616263616263616263616263616263616263110000"), False)
@@ -231,28 +250,68 @@ def test_lzxpress_huffman_decompress() -> None:
     )
 
 
+@pytest.mark.benchmark
+def test_benchmark_lzxpress_huffman_decompress(benchmark: BenchmarkFixture) -> None:
+    buf = bytes.fromhex(
+        "0000000000000000000000000000000000000000000000000000000000000000"
+        "0000000000000000000000000000000030230000000000000000000000000000"
+        "0000000000000000000000000000000000000000000000000000000000000000"
+        "0000000000000000000000000000000000000000000000000000000000000000"
+        "0200000000000000000000000000002000000000000000000000000000000000"
+        "0000000000000000000000000000000000000000000000000000000000000000"
+        "0000000000000000000000000000000000000000000000000000000000000000"
+        "0000000000000000000000000000000000000000000000000000000000000000"
+        "a8dc0000ff2601"
+    )
+    assert benchmark(lzxpress_huffman.decompress, buf) == b"abc" * 100
+
+
 def test_lzxpress_decompress() -> None:
     assert lzxpress.decompress(bytes.fromhex("ffff ff1f 6162 6317 000f ff26 01")) == b"abc" * 100
 
 
+@pytest.mark.benchmark
+def test_benchmark_lzxpress_decompress(benchmark: BenchmarkFixture) -> None:
+    buf = bytes.fromhex("ffff ff1f 6162 6317 000f ff26 01")
+    assert benchmark(lzxpress.decompress, buf) == b"abc" * 100
+
+
 def test_sevenbit_compress() -> None:
     result = sevenbit.compress(b"7-bit compression test string")
     target = bytes.fromhex("b796384d078ddf6db8bc3c9fa7df6e10bd3ca783e67479da7d06")
     assert result == target
 
 
+@pytest.mark.benchmark
+def test_benchmark_sevenbit_compress(benchmark: BenchmarkFixture) -> None:
+    buf = b"7-bit compression test string"
+    assert benchmark(sevenbit.compress, buf) == bytes.fromhex("b796384d078ddf6db8bc3c9fa7df6e10bd3ca783e67479da7d06")
+
+
 def test_sevenbit_decompress() -> None:
     result = sevenbit.decompress(bytes.fromhex("b796384d078ddf6db8bc3c9fa7df6e10bd3ca783e67479da7d06"))
     target = b"7-bit compression test string"
     assert result == target
 
 
+@pytest.mark.benchmark
+def test_benchmark_sevenbit_decompress(benchmark: BenchmarkFixture) -> None:
+    buf = bytes.fromhex("b796384d078ddf6db8bc3c9fa7df6e10bd3ca783e67479da7d06")
+    assert benchmark(sevenbit.decompress, buf) == b"7-bit compression test string"
+
+
 def test_sevenbit_decompress_wide() -> None:
     result = sevenbit.decompress(bytes.fromhex("b796384d078ddf6db8bc3c9fa7df6e10bd3ca783e67479da7d06"), wide=True)
     target = "7-bit compression test string".encode("utf-16-le")
     assert result == target
 
 
+@pytest.mark.benchmark
+def test_benchmark_sevenbit_decompress_wide(benchmark: BenchmarkFixture) -> None:
+    buf = bytes.fromhex("b796384d078ddf6db8bc3c9fa7df6e10bd3ca783e67479da7d06")
+    assert benchmark(sevenbit.decompress, buf, wide=True) == "7-bit compression test string".encode("utf-16-le")
+
+
 def test_xz_repair_checksum() -> None:
     buf = BytesIO(
         bytes.fromhex(

tests/test_cpio.py

@@ -1,11 +1,16 @@
+from __future__ import annotations
+
 import gzip
 from pathlib import Path
-from tarfile import TarFile
+from typing import TYPE_CHECKING
 
 import pytest
 
 from dissect.util import cpio
 
+if TYPE_CHECKING:
+    from tarfile import TarFile
+
 
 def absolute_path(filename: str) -> Path:
     return Path(__file__).parent / filename

tests/test_crc32c.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import importlib.util
 from typing import TYPE_CHECKING
 
 import pytest
@@ -10,8 +9,6 @@
 if TYPE_CHECKING:
     from pytest_benchmark.fixture import BenchmarkFixture
 
-HAS_BENCHMARK = importlib.util.find_spec("pytest_benchmark") is not None
-
 
 @pytest.mark.parametrize(
     ("data", "value", "expected"),
@@ -36,6 +33,6 @@ def test_crc32c(data: bytes, value: int, expected: int) -> None:
     assert crc32c.crc32c(data, value) == expected
 
 
-@pytest.mark.skipif(not HAS_BENCHMARK, reason="pytest-benchmark not installed")
+@pytest.mark.benchmark
 def test_crc32c_benchmark(benchmark: BenchmarkFixture) -> None:
     benchmark(crc32c.crc32c, b"hello, world!", 0)

tests/test_feature.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import pytest
 
 from dissect.util.feature import Feature, FeatureError, feature, feature_enabled

tests/test_plist.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import datetime
 import sys
 import uuid

tests/test_sid.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import importlib.util
 import io
 from typing import TYPE_CHECKING, BinaryIO
 
@@ -11,8 +10,6 @@
 if TYPE_CHECKING:
     from pytest_benchmark.fixture import BenchmarkFixture
 
-HAS_BENCHMARK = importlib.util.find_spec("pytest_benchmark") is not None
-
 
 def id_fn(val: bytes | str) -> str:
     if isinstance(val, io.BytesIO):
@@ -91,7 +88,7 @@ def test_read_sid(binary_sid: bytes | BinaryIO, endian: str, swap_last: bool, re
     assert readable_sid == sid.read_sid(binary_sid, endian, swap_last)
 
 
-@pytest.mark.skipif(not HAS_BENCHMARK, reason="pytest-benchmark not installed")
+@pytest.mark.benchmark
 @pytest.mark.parametrize(
     ("binary_sid", "swap_last"),
     [

tests/test_stream.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import importlib.util
 import io
 import zlib
 from typing import TYPE_CHECKING
@@ -13,8 +12,6 @@
 if TYPE_CHECKING:
     from pytest_benchmark.fixture import BenchmarkFixture
 
-HAS_BENCHMARK = importlib.util.find_spec("pytest_benchmark") is not None
-
 
 def test_range_stream() -> None:
     buf = io.BytesIO(b"\x01" * 10 + b"\x02" * 10 + b"\x03" * 10)
@@ -236,13 +233,25 @@ def _read(self, offset: int, length: int) -> bytes:
         return b"\x00" * length
 
 
-@pytest.mark.skipif(not HAS_BENCHMARK, reason="pytest-benchmark not installed")
+@pytest.mark.benchmark
 def test_aligned_stream_read_fixed_size_benchmark(benchmark: BenchmarkFixture) -> None:
     fh = NullStream(1024 * 1024)
-    benchmark(fh.read, 1234)
+
+    @benchmark
+    def run() -> None:
+        fh.seek(0)
+        fh.read(1234)
+
+    assert run is None
 
 
-@pytest.mark.skipif(not HAS_BENCHMARK, reason="pytest-benchmark not installed")
+@pytest.mark.benchmark
 def test_aligned_stream_read_none_size_benchmark(benchmark: BenchmarkFixture) -> None:
     fh = NullStream(None)
-    benchmark(fh.read, 1234)
+
+    @benchmark
+    def run() -> None:
+        fh.seek(0)
+        fh.read(1234)
+
+    assert run is None

tests/test_ts.py

@@ -1,11 +1,16 @@
+from __future__ import annotations
+
 import platform
 from datetime import datetime, timedelta, timezone
 from importlib import reload
-from types import ModuleType
+from typing import TYPE_CHECKING
 from unittest.mock import patch
 
 import pytest
 
+if TYPE_CHECKING:
+    from types import ModuleType
+
 
 @pytest.fixture(params=["windows", "emscripten", "linux"])
 def imported_ts(request: pytest.FixtureRequest) -> ModuleType: