Verified Commit 70bdbd76 authored by David Runge's avatar David Runge
Browse files

Add method to read db file

repo_management/files.py:
Add `_read_db_file()` to open a db file and return it as
tarfile.Tarfile. The method assumes gzip-compressed databases by
default, but can be used with other compression algorithms.

tests/test_file.py:
Add fixtures and tests for `_read_db_file()`.
parent 84e32326
Pipeline #5539 passed with stage
in 48 seconds
import tarfile
from pathlib import Path
def _read_db_file(db_path: Path, compression: str = "gz") -> tarfile.TarFile:
"""Read a repository database file
Parameters
----------
db_path: Path
A pathlib.Path instance, representing the location of the database file
compression: str
The compression used for the database file (defaults to 'gz')
Raises
------
ValueError
If the file represented by db_path does not exist
tarfile.ReadError
If the file could not be opened
tarfile.CompressionError
If the provided compression does not match the compression of the file or if the compression type is unknown
Returns
-------
tarfile.Tarfile
An instance of Tarfile
"""
return tarfile.open(name=db_path, mode=f"r:{compression}")
import os
import shutil
import tarfile
import tempfile
from contextlib import nullcontext as does_not_raise
from pathlib import Path
from typing import Iterator
from pytest import fixture, raises
from repo_management import files
def create_db_file(compression: str = "gz", remove_db: bool = False) -> Path:
(file_number, db_file) = tempfile.mkstemp(suffix=".db")
temp_dirs = [
tempfile.mkdtemp(),
tempfile.mkdtemp(),
tempfile.mkdtemp(),
tempfile.mkdtemp(),
]
with tarfile.open(db_file, f"w:{compression}") as db_tar:
for name in temp_dirs:
db_tar.add(name)
for name in temp_dirs:
shutil.rmtree(name)
if remove_db:
os.remove(db_file)
return Path(db_file)
@fixture(scope="function")
def create_gz_db_file() -> Iterator[Path]:
db_file = create_db_file()
yield db_file
os.remove(db_file)
@fixture(scope="function")
def create_bzip_db_file() -> Iterator[Path]:
db_file = create_db_file(compression="bz2")
yield db_file
os.remove(db_file)
@fixture(scope="function")
def create_null_db_file() -> Iterator[Path]:
yield create_db_file(remove_db=True)
def test__read_db_file(create_gz_db_file: Path) -> None:
with does_not_raise():
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")
def test__read_db_file_does_not_exist(create_null_db_file: Path) -> None:
with raises(FileNotFoundError):
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):
files._read_db_file(create_bzip_db_file)
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