Unverified Commit 2660ca6c authored by David Runge's avatar David Runge
Browse files

Add iterator for TarFile members

Add iterator method `_read_db_file_member()` to yield the members of a
provided tarfile.TarFile.
The members are filtered by name using a regular expression, which can
be overridden.

Change `create_db_file()` to also create 'desc' and 'files' files below
the created temporary directories.
Assert, that all tested methods return a value.
Add test for `_read_db_file_member()`.
parent 70bdbd76
import re
import tarfile
from pathlib import Path
from typing import Iterator
def _read_db_file(db_path: Path, compression: str = "gz") -> tarfile.TarFile:
......@@ -28,3 +30,20 @@ def _read_db_file(db_path: Path, compression: str = "gz") -> tarfile.TarFile:
return tarfile.open(name=db_path, mode=f"r:{compression}")
def _read_db_file_member(db_file: tarfile.TarFile, regex: str = "(/desc|/files)$") -> Iterator[tarfile.TarInfo]:
"""Read the members of a database file, represented by an instance of tarfile.TarFile and yield the members as
instances of tarfile.TarInfo
An instance of TarFile representing a repository database
regex: str
A regular expression used to filter the names of the members contained in db_file (defaults to
for name in [name for name in db_file.getnames() if re.search(regex, name)]:
yield db_file.getmember(name)
......@@ -4,7 +4,7 @@ import tarfile
import tempfile
from contextlib import nullcontext as does_not_raise
from pathlib import Path
from typing import Iterator
from typing import Iterator, List
from pytest import fixture, raises
......@@ -19,9 +19,17 @@ def create_db_file(compression: str = "gz", remove_db: bool = False) -> Path:
files: List[str] = []
for tmp_dir in temp_dirs:
for file_name in ["desc", "files"]:
with open(f"{tmp_dir}/{file_name}", "x"):
files += [f"{tmp_dir}/{file_name}"]
with tarfile.open(db_file, f"w:{compression}") as db_tar:
for name in temp_dirs:
for name in temp_dirs + files:
for name in temp_dirs:
if remove_db:
......@@ -51,19 +59,24 @@ def create_null_db_file() -> Iterator[Path]:
def test__read_db_file(create_gz_db_file: Path) -> None:
with does_not_raise():
assert files._read_db_file(create_gz_db_file)
def test__read_db_file_wrong_compression(create_gz_db_file: Path) -> None:
with raises(tarfile.CompressionError):
files._read_db_file(create_gz_db_file, compression="foo")
assert files._read_db_file(create_gz_db_file, compression="foo")
def test__read_db_file_does_not_exist(create_null_db_file: Path) -> None:
with raises(FileNotFoundError):
assert files._read_db_file(create_null_db_file)
def test__read_db_file_wrong_db_compression(create_bzip_db_file: Path) -> None:
with raises(tarfile.ReadError):
assert files._read_db_file(create_bzip_db_file)
def test__read_db_file_member(create_gz_db_file: Path) -> None:
for member in files._read_db_file_member(db_file=files._read_db_file(create_gz_db_file)):
assert member
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment