feat(analyzer): rework organise_file using pathlib
- use uuids instead of datetime - massively using pathlib to manipulate paths
This commit is contained in:
parent
dac09869f3
commit
d4ffaf9a89
2 changed files with 66 additions and 190 deletions
|
@ -1,7 +1,5 @@
|
|||
import os
|
||||
import shutil
|
||||
import time
|
||||
from unittest import mock
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
|
@ -10,106 +8,49 @@ from libretime_analyzer.pipeline.organise_file import organise_file
|
|||
from ..conftest import AUDIO_FILENAME
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params,exception",
|
||||
[
|
||||
((42, "", "", dict()), TypeError),
|
||||
(("", 23, "", dict()), TypeError),
|
||||
(("", "", 5, dict()), TypeError),
|
||||
(("", "", "", 12345), TypeError),
|
||||
],
|
||||
)
|
||||
def test_organise_file_wrong_params(params, exception):
|
||||
with pytest.raises(exception):
|
||||
organise_file(*params)
|
||||
|
||||
|
||||
def test_organise_file(src_dir, dest_dir):
|
||||
organise_file(
|
||||
os.path.join(src_dir, AUDIO_FILENAME),
|
||||
dest_dir,
|
||||
def organise_file_args_factory(filepath: Path, dest_dir: Path):
|
||||
return (
|
||||
str(filepath),
|
||||
str(dest_dir),
|
||||
AUDIO_FILENAME,
|
||||
dict(),
|
||||
)
|
||||
assert os.path.exists(os.path.join(dest_dir, AUDIO_FILENAME))
|
||||
|
||||
|
||||
def test_organise_file_samefile(src_dir):
|
||||
organise_file(
|
||||
os.path.join(src_dir, AUDIO_FILENAME),
|
||||
src_dir,
|
||||
AUDIO_FILENAME,
|
||||
dict(),
|
||||
)
|
||||
assert os.path.exists(os.path.join(src_dir, AUDIO_FILENAME))
|
||||
|
||||
|
||||
def import_and_restore(src_dir, dest_dir) -> dict:
|
||||
"""
|
||||
Small helper to test the organise_file function.
|
||||
Move the file and restore it back to it's origine.
|
||||
"""
|
||||
# Import the file
|
||||
metadata = organise_file(
|
||||
os.path.join(src_dir, AUDIO_FILENAME),
|
||||
dest_dir,
|
||||
AUDIO_FILENAME,
|
||||
dict(),
|
||||
{},
|
||||
)
|
||||
|
||||
# Copy it back to the original location
|
||||
shutil.copy(
|
||||
os.path.join(dest_dir, AUDIO_FILENAME),
|
||||
os.path.join(src_dir, AUDIO_FILENAME),
|
||||
)
|
||||
|
||||
return metadata
|
||||
def test_organise_file(src_dir: Path, dest_dir: Path):
|
||||
organise_file(*organise_file_args_factory(src_dir / AUDIO_FILENAME, dest_dir))
|
||||
assert (dest_dir / AUDIO_FILENAME).exists()
|
||||
|
||||
|
||||
def test_organise_file_duplicate_file(src_dir, dest_dir):
|
||||
# Import the file once
|
||||
import_and_restore(src_dir, dest_dir)
|
||||
|
||||
# Import it again. It shouldn't overwrite the old file and instead create a new
|
||||
metadata = import_and_restore(src_dir, dest_dir)
|
||||
|
||||
assert metadata["full_path"] != os.path.join(dest_dir, AUDIO_FILENAME)
|
||||
assert os.path.exists(metadata["full_path"])
|
||||
assert os.path.exists(os.path.join(dest_dir, AUDIO_FILENAME))
|
||||
def test_organise_file_samefile(src_dir: Path):
|
||||
organise_file(*organise_file_args_factory(src_dir / AUDIO_FILENAME, src_dir))
|
||||
assert (src_dir / AUDIO_FILENAME).exists()
|
||||
|
||||
|
||||
def test_organise_file_triplicate_file(src_dir, dest_dir):
|
||||
# Here we use mock to patch out the time.localtime() function so that it
|
||||
# always returns the same value. This allows us to consistently simulate this test cases
|
||||
# where the last two of the three files are imported at the same time as the timestamp.
|
||||
with mock.patch("libretime_analyzer.pipeline.organise_file.time") as mock_time:
|
||||
mock_time.localtime.return_value = time.localtime() # date(2010, 10, 8)
|
||||
mock_time.side_effect = time.time
|
||||
def test_organise_file_duplicate_file(src_dir: Path, dest_dir: Path):
|
||||
for i in range(1, 4):
|
||||
# Make a copy so we can reuse the file
|
||||
filename = f"{i}_{AUDIO_FILENAME}"
|
||||
shutil.copy(src_dir / AUDIO_FILENAME, src_dir / filename)
|
||||
|
||||
# Import the file once
|
||||
import_and_restore(src_dir, dest_dir)
|
||||
# Import it again. It shouldn't overwrite the old file and instead create a new
|
||||
metadata1 = import_and_restore(src_dir, dest_dir)
|
||||
metadata = organise_file(
|
||||
*organise_file_args_factory(src_dir / filename, dest_dir)
|
||||
)
|
||||
|
||||
# Reimport for the third time, which should have the same timestamp as the second one
|
||||
# thanks to us mocking out time.localtime()
|
||||
metadata2 = import_and_restore(src_dir, dest_dir)
|
||||
|
||||
# Check if file exists and if filename is <original>_<date>.<ext>
|
||||
assert os.path.exists(metadata1["full_path"])
|
||||
assert len(os.path.basename(metadata1["full_path"]).split("_")) == 2
|
||||
|
||||
# Check if file exists and if filename is <original>_<date>_<uuid>.<ext>
|
||||
assert os.path.exists(metadata2["full_path"])
|
||||
assert len(os.path.basename(metadata2["full_path"]).split("_")) == 3
|
||||
full_path = Path(metadata["full_path"])
|
||||
assert full_path.exists()
|
||||
if i == 1:
|
||||
assert full_path.name == AUDIO_FILENAME
|
||||
else:
|
||||
assert len(full_path.name) == len(AUDIO_FILENAME) + 1 + 36 # _ + UUID size
|
||||
|
||||
|
||||
def test_organise_file_bad_permissions_dest_dir(src_dir):
|
||||
def test_organise_file_bad_permissions_dest_dir(src_dir: Path):
|
||||
with pytest.raises(OSError):
|
||||
# /sys is using sysfs on Linux, which is unwritable
|
||||
organise_file(
|
||||
os.path.join(src_dir, AUDIO_FILENAME),
|
||||
"/sys/foobar",
|
||||
AUDIO_FILENAME,
|
||||
dict(),
|
||||
*organise_file_args_factory(
|
||||
src_dir / AUDIO_FILENAME,
|
||||
Path("/sys/foobar"),
|
||||
)
|
||||
)
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue