test_import_audio_file.py 11.8 KB
Newer Older
1
import os
Eliot Berriot's avatar
Eliot Berriot committed
2

Eliot Berriot's avatar
Eliot Berriot committed
3
import pytest
4
5
from django.core.management import call_command
from django.core.management.base import CommandError
6

7
8
from funkwhale_api.common import utils as common_utils
from funkwhale_api.music.management.commands import import_files
9

10
DATA_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "files")
11
12


13
def test_management_command_requires_a_valid_library_id(factories):
Eliot Berriot's avatar
Eliot Berriot committed
14
    path = os.path.join(DATA_DIR, "dummy_file.ogg")
15

Eliot Berriot's avatar
Eliot Berriot committed
16
    with pytest.raises(CommandError, match=r".*Invalid library id.*"):
17
        call_command("import_files", "wrong_id", path, interactive=False)
18
19


20
def test_in_place_import_only_from_music_dir(factories, settings):
21
    library = factories["music.Library"](actor__local=True)
Eliot Berriot's avatar
Eliot Berriot committed
22
23
    settings.MUSIC_DIRECTORY_PATH = "/nope"
    path = os.path.join(DATA_DIR, "dummy_file.ogg")
Eliot Berriot's avatar
Eliot Berriot committed
24
25
26
    with pytest.raises(
        CommandError, match=r".*Importing in-place only works if importing.*"
    ):
27
        call_command(
28
            "import_files", str(library.uuid), path, in_place=True, interactive=False
29
30
31
        )


32
def test_import_with_multiple_argument(factories, mocker):
33
    library = factories["music.Library"](actor__local=True)
34
35
36
    path1 = os.path.join(DATA_DIR, "dummy_file.ogg")
    path2 = os.path.join(DATA_DIR, "utf8-éà◌.ogg")
    mocked_filter = mocker.patch(
Eliot Berriot's avatar
Eliot Berriot committed
37
        "funkwhale_api.music.management.commands.import_files.Command.filter_matching",
38
39
        return_value=({"new": [], "skipped": []}),
    )
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
    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 == "cli-{}".format(now.isoformat())
    assert upload.import_status == "pending"
    assert upload.source == "file://{}".format(path)
Eliot Berriot's avatar
Eliot Berriot committed
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
    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
91
92

    mocked_process.assert_called_once_with(upload_id=upload.pk)
93
94


RenonDis's avatar
RenonDis committed
95
def test_import_with_replace_flag(factories, mocker):
96
    library = factories["music.Library"](actor__local=True)
RenonDis's avatar
RenonDis committed
97
    path = os.path.join(DATA_DIR, "dummy_file.ogg")
98
99
100
    mocked_process = mocker.patch("funkwhale_api.music.tasks.process_upload")
    call_command(
        "import_files", str(library.uuid), path, replace=True, interactive=False
RenonDis's avatar
RenonDis committed
101
    )
102
    upload = library.uploads.last()
RenonDis's avatar
RenonDis committed
103

Eliot Berriot's avatar
Eliot Berriot committed
104
    assert upload.import_metadata["funkwhale"]["config"]["replace"] is True
RenonDis's avatar
RenonDis committed
105

106
    mocked_process.assert_called_once_with(upload_id=upload.pk)
107
108


109
110
111
112
113
114
115
116
117
118
119
120
121
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()
122

123
    assert upload.import_reference == "test"
124

125
    mocked_process.assert_called_once_with(upload_id=upload.pk)
126
127


128
def test_import_files_skip_if_path_already_imported(factories, mocker):
129
    library = factories["music.Library"](actor__local=True)
Eliot Berriot's avatar
Eliot Berriot committed
130
    path = os.path.join(DATA_DIR, "dummy_file.ogg")
131

132
133
134
135
    # existing one with same source
    factories["music.Upload"](
        library=library, import_status="finished", source="file://{}".format(path)
    )
136

137
    call_command(
Eliot Berriot's avatar
Eliot Berriot committed
138
        "import_files", str(library.uuid), path, async_=False, interactive=False
139
140
    )
    assert library.uploads.count() == 1
141
142


143
144
def test_import_files_in_place(factories, mocker, settings):
    settings.MUSIC_DIRECTORY_PATH = DATA_DIR
145
146
    mocked_process = mocker.patch("funkwhale_api.music.tasks.process_upload")
    library = factories["music.Library"](actor__local=True)
Eliot Berriot's avatar
Eliot Berriot committed
147
    path = os.path.join(DATA_DIR, "utf8-éà◌.ogg")
148
    call_command(
Eliot Berriot's avatar
Eliot Berriot committed
149
        "import_files",
150
        str(library.uuid),
151
        path,
152
        async_=False,
153
        in_place=True,
Eliot Berriot's avatar
Eliot Berriot committed
154
155
        interactive=False,
    )
156
157
158
    upload = library.uploads.last()
    assert bool(upload.audio_file) is False
    mocked_process.assert_called_once_with(upload_id=upload.pk)
159
160


161
def test_storage_rename_utf_8_files(factories):
Eliot Berriot's avatar
Eliot Berriot committed
162
163
    upload = factories["music.Upload"](audio_file__filename="été.ogg")
    assert upload.audio_file.name.endswith("ete.ogg")
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354


@pytest.mark.parametrize("name", ["modified", "moved", "created", "deleted"])
def test_handle_event(name, mocker):
    handler = mocker.patch.object(import_files, "handle_{}".format(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()
355
356
357
358
359
360
361
362
363
364
365
366
367
368


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 "Successfully imported {} new tracks".format(imported) in captured.out
    assert "For details, please refer to import reference" in captured.out