kopia lustrzana https://dev.funkwhale.audio/funkwhale/funkwhale
379 wiersze
12 KiB
Python
379 wiersze
12 KiB
Python
import os
|
|
|
|
import pytest
|
|
from django.core.management import call_command
|
|
from django.core.management.base import CommandError
|
|
|
|
from funkwhale_api.common import utils as common_utils
|
|
from funkwhale_api.music.management.commands import import_files
|
|
|
|
DATA_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "files")
|
|
|
|
|
|
def test_management_command_requires_a_valid_library_id(factories):
|
|
path = os.path.join(DATA_DIR, "dummy_file.ogg")
|
|
|
|
with pytest.raises(CommandError, match=r".*Invalid library id.*"):
|
|
call_command("import_files", "wrong_id", path, interactive=False)
|
|
|
|
|
|
def test_in_place_import_only_from_music_dir(factories, settings):
|
|
library = factories["music.Library"](actor__local=True)
|
|
settings.MUSIC_DIRECTORY_PATH = "/nope"
|
|
path = os.path.join(DATA_DIR, "dummy_file.ogg")
|
|
with pytest.raises(
|
|
CommandError, match=r".*Importing in-place only works if importing.*"
|
|
):
|
|
call_command(
|
|
"import_files", str(library.uuid), path, in_place=True, interactive=False
|
|
)
|
|
|
|
|
|
def test_import_with_multiple_argument(factories, mocker):
|
|
library = factories["music.Library"](actor__local=True)
|
|
path1 = os.path.join(DATA_DIR, "dummy_file.ogg")
|
|
path2 = os.path.join(DATA_DIR, "utf8-éà◌.ogg")
|
|
mocked_filter = mocker.patch(
|
|
"funkwhale_api.music.management.commands.import_files.Command.filter_matching",
|
|
return_value=({"new": [], "skipped": []}),
|
|
)
|
|
call_command("import_files", str(library.uuid), path1, path2, interactive=False)
|
|
mocked_filter.assert_called_once_with([path1, path2], library)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"path",
|
|
[os.path.join(DATA_DIR, "dummy_file.ogg"), os.path.join(DATA_DIR, "utf8-éà◌.ogg")],
|
|
)
|
|
def test_import_files_stores_proper_data(factories, mocker, now, path):
|
|
mocked_process = mocker.patch("funkwhale_api.music.tasks.process_upload")
|
|
library = factories["music.Library"](actor__local=True)
|
|
call_command(
|
|
"import_files", str(library.uuid), path, async_=False, interactive=False
|
|
)
|
|
upload = library.uploads.last()
|
|
assert upload.import_reference == f"cli-{now.isoformat()}"
|
|
assert upload.import_status == "pending"
|
|
assert upload.source == f"file://{path}"
|
|
assert upload.import_metadata == {
|
|
"funkwhale": {
|
|
"config": {"replace": False, "dispatch_outbox": False, "broadcast": False}
|
|
}
|
|
}
|
|
|
|
mocked_process.assert_called_once_with(upload_id=upload.pk)
|
|
|
|
|
|
def test_import_with_outbox_flag(factories, mocker):
|
|
library = factories["music.Library"](actor__local=True)
|
|
path = os.path.join(DATA_DIR, "dummy_file.ogg")
|
|
mocked_process = mocker.patch("funkwhale_api.music.tasks.process_upload")
|
|
call_command(
|
|
"import_files", str(library.uuid), path, outbox=True, interactive=False
|
|
)
|
|
upload = library.uploads.last()
|
|
|
|
assert upload.import_metadata["funkwhale"]["config"]["dispatch_outbox"] is True
|
|
|
|
mocked_process.assert_called_once_with(upload_id=upload.pk)
|
|
|
|
|
|
def test_import_with_broadcast_flag(factories, mocker):
|
|
library = factories["music.Library"](actor__local=True)
|
|
path = os.path.join(DATA_DIR, "dummy_file.ogg")
|
|
mocked_process = mocker.patch("funkwhale_api.music.tasks.process_upload")
|
|
call_command(
|
|
"import_files", str(library.uuid), path, broadcast=True, interactive=False
|
|
)
|
|
upload = library.uploads.last()
|
|
|
|
assert upload.import_metadata["funkwhale"]["config"]["broadcast"] is True
|
|
|
|
mocked_process.assert_called_once_with(upload_id=upload.pk)
|
|
|
|
|
|
def test_import_with_replace_flag(factories, mocker):
|
|
library = factories["music.Library"](actor__local=True)
|
|
path = os.path.join(DATA_DIR, "dummy_file.ogg")
|
|
mocked_process = mocker.patch("funkwhale_api.music.tasks.process_upload")
|
|
call_command(
|
|
"import_files", str(library.uuid), path, replace=True, interactive=False
|
|
)
|
|
upload = library.uploads.last()
|
|
|
|
assert upload.import_metadata["funkwhale"]["config"]["replace"] is True
|
|
|
|
mocked_process.assert_called_once_with(upload_id=upload.pk)
|
|
|
|
|
|
def test_import_with_custom_reference(factories, mocker):
|
|
library = factories["music.Library"](actor__local=True)
|
|
path = os.path.join(DATA_DIR, "dummy_file.ogg")
|
|
mocked_process = mocker.patch("funkwhale_api.music.tasks.process_upload")
|
|
call_command(
|
|
"import_files",
|
|
str(library.uuid),
|
|
path,
|
|
reference="test",
|
|
replace=True,
|
|
interactive=False,
|
|
)
|
|
upload = library.uploads.last()
|
|
|
|
assert upload.import_reference == "test"
|
|
|
|
mocked_process.assert_called_once_with(upload_id=upload.pk)
|
|
|
|
|
|
def test_import_files_skip_if_path_already_imported(factories, mocker):
|
|
library = factories["music.Library"](actor__local=True)
|
|
path = os.path.join(DATA_DIR, "dummy_file.ogg")
|
|
|
|
# existing one with same source
|
|
factories["music.Upload"](
|
|
library=library, import_status="finished", source=f"file://{path}"
|
|
)
|
|
|
|
call_command(
|
|
"import_files", str(library.uuid), path, async_=False, interactive=False
|
|
)
|
|
assert library.uploads.count() == 1
|
|
|
|
|
|
def test_import_files_in_place(factories, mocker, settings):
|
|
settings.MUSIC_DIRECTORY_PATH = DATA_DIR
|
|
mocked_process = mocker.patch("funkwhale_api.music.tasks.process_upload")
|
|
library = factories["music.Library"](actor__local=True)
|
|
path = os.path.join(DATA_DIR, "utf8-éà◌.ogg")
|
|
call_command(
|
|
"import_files",
|
|
str(library.uuid),
|
|
path,
|
|
async_=False,
|
|
in_place=True,
|
|
interactive=False,
|
|
)
|
|
upload = library.uploads.last()
|
|
assert not upload.audio_file
|
|
mocked_process.assert_called_once_with(upload_id=upload.pk)
|
|
|
|
|
|
def test_storage_rename_utf_8_files(factories):
|
|
upload = factories["music.Upload"](audio_file__filename="été.ogg")
|
|
assert upload.audio_file.name.endswith("ete.ogg")
|
|
|
|
|
|
@pytest.mark.parametrize("name", ["modified", "moved", "created", "deleted"])
|
|
def test_handle_event(name, mocker):
|
|
handler = mocker.patch.object(import_files, f"handle_{name}")
|
|
|
|
event = {"type": name}
|
|
stdout = mocker.Mock()
|
|
kwargs = {"hello": "world"}
|
|
import_files.handle_event(event, stdout, **kwargs)
|
|
|
|
handler.assert_called_once_with(event=event, stdout=stdout, **kwargs)
|
|
|
|
|
|
def test_handle_created(mocker):
|
|
handle_modified = mocker.patch.object(import_files, "handle_modified")
|
|
|
|
event = mocker.Mock()
|
|
stdout = mocker.Mock()
|
|
kwargs = {"hello": "world"}
|
|
import_files.handle_created(event, stdout, **kwargs)
|
|
|
|
handle_modified.assert_called_once_with(event, stdout, **kwargs)
|
|
|
|
|
|
def test_handle_deleted(factories, mocker):
|
|
stdout = mocker.Mock()
|
|
event = {
|
|
"path": "/path.mp3",
|
|
}
|
|
library = factories["music.Library"]()
|
|
deleted = factories["music.Upload"](
|
|
library=library,
|
|
source="file://{}".format(event["path"]),
|
|
import_status="finished",
|
|
audio_file=None,
|
|
)
|
|
kept = [
|
|
factories["music.Upload"](
|
|
library=library,
|
|
source="file://{}".format(event["path"]),
|
|
import_status="finished",
|
|
),
|
|
factories["music.Upload"](
|
|
source="file://{}".format(event["path"]),
|
|
import_status="finished",
|
|
audio_file=None,
|
|
),
|
|
]
|
|
|
|
import_files.handle_deleted(
|
|
event=event, stdout=stdout, library=library, in_place=True
|
|
)
|
|
|
|
with pytest.raises(deleted.DoesNotExist):
|
|
deleted.refresh_from_db()
|
|
|
|
for upload in kept:
|
|
upload.refresh_from_db()
|
|
|
|
|
|
def test_handle_moved(factories, mocker):
|
|
stdout = mocker.Mock()
|
|
event = {
|
|
"src_path": "/path.mp3",
|
|
"dest_path": "/new_path.mp3",
|
|
}
|
|
library = factories["music.Library"]()
|
|
updated = factories["music.Upload"](
|
|
library=library,
|
|
source="file://{}".format(event["src_path"]),
|
|
import_status="finished",
|
|
audio_file=None,
|
|
)
|
|
untouched = [
|
|
factories["music.Upload"](
|
|
library=library,
|
|
source="file://{}".format(event["src_path"]),
|
|
import_status="finished",
|
|
),
|
|
factories["music.Upload"](
|
|
source="file://{}".format(event["src_path"]),
|
|
import_status="finished",
|
|
audio_file=None,
|
|
),
|
|
]
|
|
|
|
import_files.handle_moved(
|
|
event=event, stdout=stdout, library=library, in_place=True
|
|
)
|
|
|
|
updated.refresh_from_db()
|
|
assert updated.source == "file://{}".format(event["dest_path"])
|
|
for upload in untouched:
|
|
source = upload.source
|
|
upload.refresh_from_db()
|
|
assert source == upload.source
|
|
|
|
|
|
def test_handle_modified_creates_upload(tmpfile, factories, mocker):
|
|
stdout = mocker.Mock()
|
|
event = {
|
|
"path": tmpfile.name,
|
|
}
|
|
process_upload = mocker.patch("funkwhale_api.music.tasks.process_upload")
|
|
library = factories["music.Library"]()
|
|
import_files.handle_modified(
|
|
event=event,
|
|
stdout=stdout,
|
|
library=library,
|
|
in_place=True,
|
|
reference="hello",
|
|
replace=False,
|
|
dispatch_outbox=False,
|
|
broadcast=False,
|
|
)
|
|
upload = library.uploads.latest("id")
|
|
assert upload.source == "file://{}".format(event["path"])
|
|
|
|
process_upload.assert_called_once_with(upload_id=upload.pk)
|
|
|
|
|
|
def test_handle_modified_skips_existing_checksum(tmpfile, factories, mocker):
|
|
stdout = mocker.Mock()
|
|
event = {
|
|
"path": tmpfile.name,
|
|
}
|
|
tmpfile.write(b"hello")
|
|
|
|
library = factories["music.Library"]()
|
|
factories["music.Upload"](
|
|
checksum=common_utils.get_file_hash(tmpfile),
|
|
library=library,
|
|
import_status="finished",
|
|
)
|
|
import_files.handle_modified(
|
|
event=event,
|
|
stdout=stdout,
|
|
library=library,
|
|
in_place=True,
|
|
)
|
|
assert library.uploads.count() == 1
|
|
|
|
|
|
def test_handle_modified_update_existing_path_if_found(tmpfile, factories, mocker):
|
|
stdout = mocker.Mock()
|
|
event = {
|
|
"path": tmpfile.name,
|
|
}
|
|
update_track_metadata = mocker.patch(
|
|
"funkwhale_api.music.tasks.update_track_metadata"
|
|
)
|
|
get_metadata = mocker.patch("funkwhale_api.music.models.Upload.get_metadata")
|
|
library = factories["music.Library"]()
|
|
track = factories["music.Track"](attributed_to=library.actor)
|
|
upload = factories["music.Upload"](
|
|
source="file://{}".format(event["path"]),
|
|
track=track,
|
|
checksum="old",
|
|
library=library,
|
|
import_status="finished",
|
|
audio_file=None,
|
|
)
|
|
import_files.handle_modified(
|
|
event=event,
|
|
stdout=stdout,
|
|
library=library,
|
|
in_place=True,
|
|
)
|
|
update_track_metadata.assert_called_once_with(
|
|
get_metadata.return_value,
|
|
upload.track,
|
|
)
|
|
|
|
|
|
def test_handle_modified_update_existing_path_if_found_and_attributed_to(
|
|
tmpfile, factories, mocker
|
|
):
|
|
stdout = mocker.Mock()
|
|
event = {
|
|
"path": tmpfile.name,
|
|
}
|
|
update_track_metadata = mocker.patch(
|
|
"funkwhale_api.music.tasks.update_track_metadata"
|
|
)
|
|
library = factories["music.Library"]()
|
|
factories["music.Upload"](
|
|
source="file://{}".format(event["path"]),
|
|
checksum="old",
|
|
library=library,
|
|
track__attributed_to=factories["federation.Actor"](),
|
|
import_status="finished",
|
|
audio_file=None,
|
|
)
|
|
import_files.handle_modified(
|
|
event=event,
|
|
stdout=stdout,
|
|
library=library,
|
|
in_place=True,
|
|
)
|
|
update_track_metadata.assert_not_called()
|
|
|
|
|
|
def test_import_files(factories, capsys):
|
|
# smoke test to ensure the command run properly
|
|
library = factories["music.Library"](actor__local=True)
|
|
call_command(
|
|
"import_files", str(library.uuid), DATA_DIR, interactive=False, recursive=True
|
|
)
|
|
captured = capsys.readouterr()
|
|
|
|
imported = library.uploads.filter(import_status="finished").count()
|
|
assert imported > 0
|
|
assert f"Successfully imported {imported} new tracks" in captured.out
|
|
assert "For details, please refer to import reference" in captured.out
|