Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions tests/handlers/compression/test_gzip.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,9 @@
from pathlib import Path

from unblob.handlers.compression.gzip import MultiVolumeGzipHandler


def test_multivolume_is_valid_gzip_empty_file(tmp_path: Path):
empty = tmp_path / "empty"
empty.touch()
assert not MultiVolumeGzipHandler().is_valid_gzip(empty)
4 changes: 2 additions & 2 deletions tests/test_file_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -83,13 +83,13 @@ def fake_file() -> File:

class TestFile:
def test_file_from_empty_bytes(self):
with pytest.raises(InvalidInputFormat):
with pytest.raises(ValueError): # noqa: PT011
File.from_bytes(b"")

def test_file_from_empty_file(self, tmp_path):
file_path = tmp_path / "file"
file_path.touch()
with pytest.raises(InvalidInputFormat):
with pytest.raises(ValueError, match="cannot mmap an empty file"):
File.from_path(file_path)


Expand Down
20 changes: 11 additions & 9 deletions unblob/file_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,17 +40,13 @@ class SeekError(ValueError):
"""Specific ValueError for File.seek."""


class InvalidInputFormat(Exception):
pass


class File(mmap.mmap):
access: int

@classmethod
def from_bytes(cls, content: bytes):
if not content:
raise InvalidInputFormat("Can't create File from empty bytes.")
raise ValueError("Can't create File from empty bytes.")
m = cls(-1, len(content))
m.write(content)
m.seek(0)
Expand All @@ -59,12 +55,14 @@ def from_bytes(cls, content: bytes):

@classmethod
def from_path(cls, path: Path, access=mmap.ACCESS_READ):
"""Create File.

Needs a valid non-empty file,
raises ValueError on empty files.
"""
mode = "r+b" if access == mmap.ACCESS_WRITE else "rb"
with path.open(mode) as base_file:
try:
m = cls(base_file.fileno(), 0, access=access)
except ValueError as exc:
raise InvalidInputFormat from exc
m = cls(base_file.fileno(), 0, access=access)
m.access = access
return m

Expand Down Expand Up @@ -124,6 +122,10 @@ def tell(self):
return self._file.tell() - self._offset


class InvalidInputFormat(Exception):
pass


class Endian(enum.Enum):
LITTLE = "<"
BIG = ">"
Expand Down
6 changes: 1 addition & 5 deletions unblob/handlers/archive/sevenzip.py
Original file line number Diff line number Diff line change
Expand Up @@ -107,11 +107,7 @@ class MultiVolumeSevenZipHandler(DirectoryHandler):

def calculate_multifile(self, file: Path) -> Optional[MultiFile]:
paths = sorted(
[
p
for p in file.parent.glob(f"{file.stem}.*")
if p.resolve().exists() and p.stat().st_size > 0
]
[p for p in file.parent.glob(f"{file.stem}.*") if p.resolve().exists()]
)
if not paths:
return None
Expand Down
13 changes: 7 additions & 6 deletions unblob/handlers/compression/gzip.py
Original file line number Diff line number Diff line change
Expand Up @@ -156,7 +156,12 @@ class MultiVolumeGzipHandler(DirectoryHandler):
PATTERN = Glob("*.gz.*")

def is_valid_gzip(self, path: Path) -> bool:
with File.from_path(path) as f:
try:
file = File.from_path(path)
except ValueError:
return False

with file as f:
try:
fp = SingleMemberGzipReader(f)
if not fp.read_header():
Expand All @@ -167,11 +172,7 @@ def is_valid_gzip(self, path: Path) -> bool:

def calculate_multifile(self, file: Path) -> Optional[MultiFile]:
paths = sorted(
[
p
for p in file.parent.glob(f"{file.stem}.*")
if p.resolve().exists() and p.stat().st_size > 0
]
[p for p in file.parent.glob(f"{file.stem}.*") if p.resolve().exists()]
)

# we 'discard' paths that are not the first in the ordered list,
Expand Down