novaburst/funquail
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Audio federation

Browse source
This commit is contained in:
Eliot Berriot 2018-09-22 12:29:30 +00:00
commit e49a460203
85 changed files with 2591 additions and 1197 deletions
Download patch file Download diff file Expand all files Collapse all files

10
api/tests/music/test_activity.py
View file

@ -14,14 +14,14 @@ def test_get_track_activity_url_no_mbid(settings, factories):
assert track.get_activity_url() == expected
def test_track_file_import_status_updated_broadcast(factories, mocker):
def test_upload_import_status_updated_broadcast(factories, mocker):
group_send = mocker.patch("funkwhale_api.common.channels.group_send")
user = factories["users.User"]()
tf = factories["music.TrackFile"](
upload = factories["music.Upload"](
import_status="finished", library__actor__user=user
)
signals.track_file_import_status_updated.send(
sender=None, track_file=tf, old_status="pending", new_status="finished"
signals.upload_import_status_updated.send(
sender=None, upload=upload, old_status="pending", new_status="finished"
)
group_send.assert_called_once_with(
"user.{}.imports".format(user.pk),
@ -32,7 +32,7 @@ def test_track_file_import_status_updated_broadcast(factories, mocker):
"type": "import.status_updated",
"old_status": "pending",
"new_status": "finished",
"track_file": serializers.TrackFileForOwnerSerializer(tf).data,
"upload": serializers.UploadForOwnerSerializer(upload).data,
},
},
)

18
api/tests/music/test_api.py
View file

@ -25,26 +25,26 @@ def test_can_restrict_api_views_to_authenticated_users(
assert response.status_code == 401
def test_track_file_url_is_restricted_to_authenticated_users(
def test_upload_url_is_restricted_to_authenticated_users(
api_client, factories, preferences
):
preferences["common__api_authentication_required"] = True
tf = factories["music.TrackFile"](library__privacy_level="instance")
assert tf.audio_file is not None
url = tf.track.listen_url
upload = factories["music.Upload"](library__privacy_level="instance")
assert upload.audio_file is not None
url = upload.track.listen_url
response = api_client.get(url)
assert response.status_code == 401
def test_track_file_url_is_accessible_to_authenticated_users(
def test_upload_url_is_accessible_to_authenticated_users(
logged_in_api_client, factories, preferences
):
actor = logged_in_api_client.user.create_actor()
preferences["common__api_authentication_required"] = True
tf = factories["music.TrackFile"](library__actor=actor)
assert tf.audio_file is not None
url = tf.track.listen_url
upload = factories["music.Upload"](library__actor=actor)
assert upload.audio_file is not None
url = upload.track.listen_url
response = logged_in_api_client.get(url)
assert response.status_code == 200
assert response["X-Accel-Redirect"] == "/_protected{}".format(tf.audio_file.url)
assert response["X-Accel-Redirect"] == "/_protected{}".format(upload.audio_file.url)

60
api/tests/music/test_commands.py
View file

@ -1,14 +1,14 @@
import os
from funkwhale_api.music.management.commands import fix_track_files
from funkwhale_api.music.management.commands import fix_uploads
DATA_DIR = os.path.dirname(os.path.abspath(__file__))
def test_fix_track_files_bitrate_length(factories, mocker):
tf1 = factories["music.TrackFile"](bitrate=1, duration=2)
tf2 = factories["music.TrackFile"](bitrate=None, duration=None)
c = fix_track_files.Command()
def test_fix_uploads_bitrate_length(factories, mocker):
upload1 = factories["music.Upload"](bitrate=1, duration=2)
upload2 = factories["music.Upload"](bitrate=None, duration=None)
c = fix_uploads.Command()
mocker.patch(
"funkwhale_api.music.utils.get_audio_file_data",
@ -17,59 +17,59 @@ def test_fix_track_files_bitrate_length(factories, mocker):
c.fix_file_data(dry_run=False)
tf1.refresh_from_db()
tf2.refresh_from_db()
upload1.refresh_from_db()
upload2.refresh_from_db()
# not updated
assert tf1.bitrate == 1
assert tf1.duration == 2
assert upload1.bitrate == 1
assert upload1.duration == 2
# updated
assert tf2.bitrate == 42
assert tf2.duration == 43
assert upload2.bitrate == 42
assert upload2.duration == 43
def test_fix_track_files_size(factories, mocker):
tf1 = factories["music.TrackFile"]()
tf2 = factories["music.TrackFile"]()
tf1.__class__.objects.filter(pk=tf1.pk).update(size=1)
tf2.__class__.objects.filter(pk=tf2.pk).update(size=None)
c = fix_track_files.Command()
def test_fix_uploads_size(factories, mocker):
upload1 = factories["music.Upload"]()
upload2 = factories["music.Upload"]()
upload1.__class__.objects.filter(pk=upload1.pk).update(size=1)
upload2.__class__.objects.filter(pk=upload2.pk).update(size=None)
c = fix_uploads.Command()
mocker.patch("funkwhale_api.music.models.TrackFile.get_file_size", return_value=2)
mocker.patch("funkwhale_api.music.models.Upload.get_file_size", return_value=2)
c.fix_file_size(dry_run=False)
tf1.refresh_from_db()
tf2.refresh_from_db()
upload1.refresh_from_db()
upload2.refresh_from_db()
# not updated
assert tf1.size == 1
assert upload1.size == 1
# updated
assert tf2.size == 2
assert upload2.size == 2
def test_fix_track_files_mimetype(factories, mocker):
def test_fix_uploads_mimetype(factories, mocker):
mp3_path = os.path.join(DATA_DIR, "test.mp3")
ogg_path = os.path.join(DATA_DIR, "test.ogg")
tf1 = factories["music.TrackFile"](
upload1 = factories["music.Upload"](
audio_file__from_path=mp3_path,
source="file://{}".format(mp3_path),
mimetype="application/x-empty",
)
# this one already has a mimetype set, to it should not be updated
tf2 = factories["music.TrackFile"](
upload2 = factories["music.Upload"](
audio_file__from_path=ogg_path,
source="file://{}".format(ogg_path),
mimetype="audio/something",
)
c = fix_track_files.Command()
c = fix_uploads.Command()
c.fix_mimetypes(dry_run=False)
tf1.refresh_from_db()
tf2.refresh_from_db()
upload1.refresh_from_db()
upload2.refresh_from_db()
assert tf1.mimetype == "audio/mpeg"
assert tf2.mimetype == "audio/something"
assert upload1.mimetype == "audio/mpeg"
assert upload2.mimetype == "audio/something"

163
api/tests/music/test_models.py
View file

@ -3,8 +3,10 @@ import os
import pytest
from django.utils import timezone
from django.urls import reverse
from funkwhale_api.music import importers, models, tasks
from funkwhale_api.federation import utils as federation_utils
DATA_DIR = os.path.dirname(os.path.abspath(__file__))
@ -157,33 +159,33 @@ def test_audio_track_mime_type(extention, mimetype, factories):
name = ".".join(["test", extention])
path = os.path.join(DATA_DIR, name)
tf = factories["music.TrackFile"](audio_file__from_path=path, mimetype=None)
upload = factories["music.Upload"](audio_file__from_path=path, mimetype=None)
assert tf.mimetype == mimetype
assert upload.mimetype == mimetype
def test_track_file_file_name(factories):
def test_upload_file_name(factories):
name = "test.mp3"
path = os.path.join(DATA_DIR, name)
tf = factories["music.TrackFile"](audio_file__from_path=path)
upload = factories["music.Upload"](audio_file__from_path=path)
assert tf.filename == tf.track.full_name + ".mp3"
assert upload.filename == upload.track.full_name + ".mp3"
def test_track_get_file_size(factories):
name = "test.mp3"
path = os.path.join(DATA_DIR, name)
tf = factories["music.TrackFile"](audio_file__from_path=path)
upload = factories["music.Upload"](audio_file__from_path=path)
assert tf.get_file_size() == 297745
assert upload.get_file_size() == 297745
def test_track_get_file_size_in_place(factories):
name = "test.mp3"
path = os.path.join(DATA_DIR, name)
tf = factories["music.TrackFile"](in_place=True, source="file://{}".format(path))
upload = factories["music.Upload"](in_place=True, source="file://{}".format(path))
assert tf.get_file_size() == 297745
assert upload.get_file_size() == 297745
def test_album_get_image_content(factories):
@ -202,7 +204,7 @@ def test_library(factories):
)
assert library.creation_date >= now
assert library.files.count() == 0
assert library.uploads.count() == 0
assert library.uuid is not None
@ -210,9 +212,9 @@ def test_library(factories):
"privacy_level,expected", [("me", True), ("instance", True), ("everyone", True)]
)
def test_playable_by_correct_actor(privacy_level, expected, factories):
tf = factories["music.TrackFile"](library__privacy_level=privacy_level)
queryset = tf.library.files.playable_by(tf.library.actor)
match = tf in list(queryset)
upload = factories["music.Upload"](library__privacy_level=privacy_level)
queryset = upload.library.uploads.playable_by(upload.library.actor)
match = upload in list(queryset)
assert match is expected
@ -220,10 +222,10 @@ def test_playable_by_correct_actor(privacy_level, expected, factories):
"privacy_level,expected", [("me", False), ("instance", True), ("everyone", True)]
)
def test_playable_by_instance_actor(privacy_level, expected, factories):
tf = factories["music.TrackFile"](library__privacy_level=privacy_level)
instance_actor = factories["federation.Actor"](domain=tf.library.actor.domain)
queryset = tf.library.files.playable_by(instance_actor)
match = tf in list(queryset)
upload = factories["music.Upload"](library__privacy_level=privacy_level)
instance_actor = factories["federation.Actor"](domain=upload.library.actor.domain)
queryset = upload.library.uploads.playable_by(instance_actor)
match = upload in list(queryset)
assert match is expected
@ -231,9 +233,22 @@ def test_playable_by_instance_actor(privacy_level, expected, factories):
"privacy_level,expected", [("me", False), ("instance", False), ("everyone", True)]
)
def test_playable_by_anonymous(privacy_level, expected, factories):
tf = factories["music.TrackFile"](library__privacy_level=privacy_level)
queryset = tf.library.files.playable_by(None)
match = tf in list(queryset)
upload = factories["music.Upload"](library__privacy_level=privacy_level)
queryset = upload.library.uploads.playable_by(None)
match = upload in list(queryset)
assert match is expected
@pytest.mark.parametrize("approved", [True, False])
def test_playable_by_follower(approved, factories):
upload = factories["music.Upload"](library__privacy_level="me")
actor = factories["federation.Actor"](local=True)
factories["federation.LibraryFollow"](
target=upload.library, actor=actor, approved=approved
)
queryset = upload.library.uploads.playable_by(actor)
match = upload in list(queryset)
expected = approved
assert match is expected
@ -241,11 +256,11 @@ def test_playable_by_anonymous(privacy_level, expected, factories):
"privacy_level,expected", [("me", True), ("instance", True), ("everyone", True)]
)
def test_track_playable_by_correct_actor(privacy_level, expected, factories):
tf = factories["music.TrackFile"]()
upload = factories["music.Upload"]()
queryset = models.Track.objects.playable_by(
tf.library.actor
).annotate_playable_by_actor(tf.library.actor)
match = tf.track in list(queryset)
upload.library.actor
).annotate_playable_by_actor(upload.library.actor)
match = upload.track in list(queryset)
assert match is expected
if expected:
assert bool(queryset.first().is_playable_by_actor) is expected
@ -255,12 +270,12 @@ def test_track_playable_by_correct_actor(privacy_level, expected, factories):
"privacy_level,expected", [("me", False), ("instance", True), ("everyone", True)]
)
def test_track_playable_by_instance_actor(privacy_level, expected, factories):
tf = factories["music.TrackFile"](library__privacy_level=privacy_level)
instance_actor = factories["federation.Actor"](domain=tf.library.actor.domain)
upload = factories["music.Upload"](library__privacy_level=privacy_level)
instance_actor = factories["federation.Actor"](domain=upload.library.actor.domain)
queryset = models.Track.objects.playable_by(
instance_actor
).annotate_playable_by_actor(instance_actor)
match = tf.track in list(queryset)
match = upload.track in list(queryset)
assert match is expected
if expected:
assert bool(queryset.first().is_playable_by_actor) is expected
@ -270,9 +285,9 @@ def test_track_playable_by_instance_actor(privacy_level, expected, factories):
"privacy_level,expected", [("me", False), ("instance", False), ("everyone", True)]
)
def test_track_playable_by_anonymous(privacy_level, expected, factories):
tf = factories["music.TrackFile"](library__privacy_level=privacy_level)
upload = factories["music.Upload"](library__privacy_level=privacy_level)
queryset = models.Track.objects.playable_by(None).annotate_playable_by_actor(None)
match = tf.track in list(queryset)
match = upload.track in list(queryset)
assert match is expected
if expected:
assert bool(queryset.first().is_playable_by_actor) is expected
@ -282,12 +297,12 @@ def test_track_playable_by_anonymous(privacy_level, expected, factories):
"privacy_level,expected", [("me", True), ("instance", True), ("everyone", True)]
)
def test_album_playable_by_correct_actor(privacy_level, expected, factories):
tf = factories["music.TrackFile"]()
upload = factories["music.Upload"]()
queryset = models.Album.objects.playable_by(
tf.library.actor
).annotate_playable_by_actor(tf.library.actor)
match = tf.track.album in list(queryset)
upload.library.actor
).annotate_playable_by_actor(upload.library.actor)
match = upload.track.album in list(queryset)
assert match is expected
if expected:
assert bool(queryset.first().is_playable_by_actor) is expected
@ -297,12 +312,12 @@ def test_album_playable_by_correct_actor(privacy_level, expected, factories):
"privacy_level,expected", [("me", False), ("instance", True), ("everyone", True)]
)
def test_album_playable_by_instance_actor(privacy_level, expected, factories):
tf = factories["music.TrackFile"](library__privacy_level=privacy_level)
instance_actor = factories["federation.Actor"](domain=tf.library.actor.domain)
upload = factories["music.Upload"](library__privacy_level=privacy_level)
instance_actor = factories["federation.Actor"](domain=upload.library.actor.domain)
queryset = models.Album.objects.playable_by(
instance_actor
).annotate_playable_by_actor(instance_actor)
match = tf.track.album in list(queryset)
match = upload.track.album in list(queryset)
assert match is expected
if expected:
assert bool(queryset.first().is_playable_by_actor) is expected
@ -312,9 +327,9 @@ def test_album_playable_by_instance_actor(privacy_level, expected, factories):
"privacy_level,expected", [("me", False), ("instance", False), ("everyone", True)]
)
def test_album_playable_by_anonymous(privacy_level, expected, factories):
tf = factories["music.TrackFile"](library__privacy_level=privacy_level)
upload = factories["music.Upload"](library__privacy_level=privacy_level)
queryset = models.Album.objects.playable_by(None).annotate_playable_by_actor(None)
match = tf.track.album in list(queryset)
match = upload.track.album in list(queryset)
assert match is expected
if expected:
assert bool(queryset.first().is_playable_by_actor) is expected
@ -324,12 +339,12 @@ def test_album_playable_by_anonymous(privacy_level, expected, factories):
"privacy_level,expected", [("me", True), ("instance", True), ("everyone", True)]
)
def test_artist_playable_by_correct_actor(privacy_level, expected, factories):
tf = factories["music.TrackFile"]()
upload = factories["music.Upload"]()
queryset = models.Artist.objects.playable_by(
tf.library.actor
).annotate_playable_by_actor(tf.library.actor)
match = tf.track.artist in list(queryset)
upload.library.actor
).annotate_playable_by_actor(upload.library.actor)
match = upload.track.artist in list(queryset)
assert match is expected
if expected:
assert bool(queryset.first().is_playable_by_actor) is expected
@ -339,12 +354,12 @@ def test_artist_playable_by_correct_actor(privacy_level, expected, factories):
"privacy_level,expected", [("me", False), ("instance", True), ("everyone", True)]
)
def test_artist_playable_by_instance_actor(privacy_level, expected, factories):
tf = factories["music.TrackFile"](library__privacy_level=privacy_level)
instance_actor = factories["federation.Actor"](domain=tf.library.actor.domain)
upload = factories["music.Upload"](library__privacy_level=privacy_level)
instance_actor = factories["federation.Actor"](domain=upload.library.actor.domain)
queryset = models.Artist.objects.playable_by(
instance_actor
).annotate_playable_by_actor(instance_actor)
match = tf.track.artist in list(queryset)
match = upload.track.artist in list(queryset)
assert match is expected
if expected:
assert bool(queryset.first().is_playable_by_actor) is expected
@ -354,24 +369,24 @@ def test_artist_playable_by_instance_actor(privacy_level, expected, factories):
"privacy_level,expected", [("me", False), ("instance", False), ("everyone", True)]
)
def test_artist_playable_by_anonymous(privacy_level, expected, factories):
tf = factories["music.TrackFile"](library__privacy_level=privacy_level)
upload = factories["music.Upload"](library__privacy_level=privacy_level)
queryset = models.Artist.objects.playable_by(None).annotate_playable_by_actor(None)
match = tf.track.artist in list(queryset)
match = upload.track.artist in list(queryset)
assert match is expected
if expected:
assert bool(queryset.first().is_playable_by_actor) is expected
def test_track_file_listen_url(factories):
tf = factories["music.TrackFile"]()
expected = tf.track.listen_url + "?file={}".format(tf.uuid)
def test_upload_listen_url(factories):
upload = factories["music.Upload"]()
expected = upload.track.listen_url + "?upload={}".format(upload.uuid)
assert tf.listen_url == expected
assert upload.listen_url == expected
def test_library_schedule_scan(factories, now, mocker):
on_commit = mocker.patch("funkwhale_api.common.utils.on_commit")
library = factories["music.Library"](files_count=5)
library = factories["music.Library"](uploads_count=5)
scan = library.schedule_scan()
@ -397,9 +412,9 @@ def test_library_schedule_scan_too_recent(factories, now):
def test_get_audio_data(factories):
tf = factories["music.TrackFile"]()
upload = factories["music.Upload"]()
result = tf.get_audio_data()
result = upload.get_audio_data()
assert result == {"duration": 229, "bitrate": 128000, "size": 3459481}
@ -419,3 +434,43 @@ def test_library_queryset_with_follows(factories):
l2 = list(qs)[1]
assert l1._follows == []
assert l2._follows == [follow]
def test_annotate_duration(factories):
tf = factories["music.Upload"](duration=32)
track = models.Track.objects.annotate_duration().get(pk=tf.track.pk)
assert track.duration == 32
def test_annotate_file_data(factories):
tf = factories["music.Upload"](size=42, bitrate=55, mimetype="audio/ogg")
track = models.Track.objects.annotate_file_data().get(pk=tf.track.pk)
assert track.size == 42
assert track.bitrate == 55
assert track.mimetype == "audio/ogg"
@pytest.mark.parametrize(
"model,factory_args,namespace",
[
(
"music.Upload",
{"library__actor__local": True},
"federation:music:uploads-detail",
),
("music.Library", {"actor__local": True}, "federation:music:libraries-detail"),
("music.Artist", {}, "federation:music:artists-detail"),
("music.Album", {}, "federation:music:albums-detail"),
("music.Track", {}, "federation:music:tracks-detail"),
],
)
def test_fid_is_populated(factories, model, factory_args, namespace):
instance = factories[model](**factory_args, fid=None)
assert instance.fid == federation_utils.full_url(
reverse(namespace, kwargs={"uuid": instance.uuid})
)

10
api/tests/music/test_music.py
View file

@ -2,6 +2,7 @@ import datetime
import pytest
from funkwhale_api.federation import utils as federation_utils
from funkwhale_api.music import models
@ -17,6 +18,9 @@ def test_can_create_artist_from_api(artists, mocker, db):
assert data["id"], "62c3befb-6366-4585-b256-809472333801"
assert artist.mbid, data["id"]
assert artist.name, "Adhesive Wombat"
assert artist.fid == federation_utils.full_url(
"/federation/music/artists/{}".format(artist.uuid)
)
def test_can_create_album_from_api(artists, albums, mocker, db):
@ -41,6 +45,9 @@ def test_can_create_album_from_api(artists, albums, mocker, db):
assert album.release_date, datetime.date(2005, 1, 1)
assert album.artist.name, "System of a Down"
assert album.artist.mbid, data["artist-credit"][0]["artist"]["id"]
assert album.fid == federation_utils.full_url(
"/federation/music/albums/{}".format(album.uuid)
)
def test_can_create_track_from_api(artists, albums, tracks, mocker, db):
@ -66,6 +73,9 @@ def test_can_create_track_from_api(artists, albums, tracks, mocker, db):
assert track.artist.name == "Adhesive Wombat"
assert str(track.album.mbid) == "a50d2a81-2a50-484d-9cb4-b9f6833f583e"
assert track.album.title == "Marsupial Madness"
assert track.fid == federation_utils.full_url(
"/federation/music/tracks/{}".format(track.uuid)
)
def test_can_create_track_from_api_with_corresponding_tags(

109
api/tests/music/test_serializers.py
View file

@ -46,8 +46,8 @@ def test_artist_with_albums_serializer(factories, to_api_date):
def test_album_track_serializer(factories, to_api_date):
tf = factories["music.TrackFile"]()
track = tf.track
upload = factories["music.Upload"]()
track = upload.track
expected = {
"id": track.id,
@ -59,33 +59,34 @@ def test_album_track_serializer(factories, to_api_date):
"is_playable": None,
"creation_date": to_api_date(track.creation_date),
"listen_url": track.listen_url,
"duration": None,
}
serializer = serializers.AlbumTrackSerializer(track)
assert serializer.data == expected
def test_track_file_serializer(factories, to_api_date):
tf = factories["music.TrackFile"]()
def test_upload_serializer(factories, to_api_date):
upload = factories["music.Upload"]()
expected = {
"uuid": str(tf.uuid),
"filename": tf.filename,
"track": serializers.TrackSerializer(tf.track).data,
"duration": tf.duration,
"mimetype": tf.mimetype,
"bitrate": tf.bitrate,
"size": tf.size,
"library": serializers.LibraryForOwnerSerializer(tf.library).data,
"creation_date": tf.creation_date.isoformat().split("+")[0] + "Z",
"uuid": str(upload.uuid),
"filename": upload.filename,
"track": serializers.TrackSerializer(upload.track).data,
"duration": upload.duration,
"mimetype": upload.mimetype,
"bitrate": upload.bitrate,
"size": upload.size,
"library": serializers.LibraryForOwnerSerializer(upload.library).data,
"creation_date": upload.creation_date.isoformat().split("+")[0] + "Z",
"import_date": None,
"import_status": "pending",
}
serializer = serializers.TrackFileSerializer(tf)
serializer = serializers.UploadSerializer(upload)
assert serializer.data == expected
def test_track_file_owner_serializer(factories, to_api_date):
tf = factories["music.TrackFile"](
def test_upload_owner_serializer(factories, to_api_date):
upload = factories["music.Upload"](
import_status="success",
import_details={"hello": "world"},
import_metadata={"import": "metadata"},
@ -95,15 +96,15 @@ def test_track_file_owner_serializer(factories, to_api_date):
)
expected = {
"uuid": str(tf.uuid),
"filename": tf.filename,
"track": serializers.TrackSerializer(tf.track).data,
"duration": tf.duration,
"mimetype": tf.mimetype,
"bitrate": tf.bitrate,
"size": tf.size,
"library": serializers.LibraryForOwnerSerializer(tf.library).data,
"creation_date": tf.creation_date.isoformat().split("+")[0] + "Z",
"uuid": str(upload.uuid),
"filename": upload.filename,
"track": serializers.TrackSerializer(upload.track).data,
"duration": upload.duration,
"mimetype": upload.mimetype,
"bitrate": upload.bitrate,
"size": upload.size,
"library": serializers.LibraryForOwnerSerializer(upload.library).data,
"creation_date": upload.creation_date.isoformat().split("+")[0] + "Z",
"metadata": {"test": "metadata"},
"import_metadata": {"import": "metadata"},
"import_date": None,
@ -112,7 +113,7 @@ def test_track_file_owner_serializer(factories, to_api_date):
"source": "upload://test",
"import_reference": "ref",
}
serializer = serializers.TrackFileForOwnerSerializer(tf)
serializer = serializers.UploadForOwnerSerializer(upload)
assert serializer.data == expected
@ -142,8 +143,8 @@ def test_album_serializer(factories, to_api_date):
def test_track_serializer(factories, to_api_date):
tf = factories["music.TrackFile"]()
track = tf.track
upload = factories["music.Upload"]()
track = upload.track
expected = {
"id": track.id,
@ -156,6 +157,10 @@ def test_track_serializer(factories, to_api_date):
"creation_date": to_api_date(track.creation_date),
"lyrics": track.get_lyrics_url(),
"listen_url": track.listen_url,
"duration": None,
"size": None,
"bitrate": None,
"mimetype": None,
}
serializer = serializers.TrackSerializer(track)
assert serializer.data == expected
@ -165,7 +170,7 @@ def test_user_cannot_bind_file_to_a_not_owned_library(factories):
user = factories["users.User"]()
library = factories["music.Library"]()
s = serializers.TrackFileForOwnerSerializer(
s = serializers.UploadForOwnerSerializer(
data={"library": library.uuid, "source": "upload://test"},
context={"user": user},
)
@ -176,7 +181,7 @@ def test_user_cannot_bind_file_to_a_not_owned_library(factories):
def test_user_can_create_file_in_own_library(factories, uploaded_audio_file):
user = factories["users.User"]()
library = factories["music.Library"](actor__user=user)
s = serializers.TrackFileForOwnerSerializer(
s = serializers.UploadForOwnerSerializer(
data={
"library": library.uuid,
"source": "upload://test",
@ -185,9 +190,9 @@ def test_user_can_create_file_in_own_library(factories, uploaded_audio_file):
context={"user": user},
)
assert s.is_valid(raise_exception=True) is True
tf = s.save()
upload = s.save()
assert tf.library == library
assert upload.library == library
def test_create_file_checks_for_user_quota(
@ -199,7 +204,7 @@ def test_create_file_checks_for_user_quota(
)
user = factories["users.User"]()
library = factories["music.Library"](actor__user=user)
s = serializers.TrackFileForOwnerSerializer(
s = serializers.UploadForOwnerSerializer(
data={
"library": library.uuid,
"source": "upload://test",
@ -211,34 +216,46 @@ def test_create_file_checks_for_user_quota(
assert s.errors["non_field_errors"] == ["upload_quota_reached"]
def test_manage_track_file_action_delete(factories):
tfs = factories["music.TrackFile"](size=5)
s = serializers.TrackFileActionSerializer(queryset=None)
def test_manage_upload_action_delete(factories, queryset_equal_list, mocker):
dispatch = mocker.patch("funkwhale_api.federation.routes.outbox.dispatch")
library1 = factories["music.Library"]()
library2 = factories["music.Library"]()
library1_uploads = factories["music.Upload"].create_batch(size=3, library=library1)
library2_uploads = factories["music.Upload"].create_batch(size=3, library=library2)
s = serializers.UploadActionSerializer(queryset=None)
s.handle_delete(tfs.__class__.objects.all())
s.handle_delete(library1_uploads[0].__class__.objects.all())
assert tfs.__class__.objects.count() == 0
assert library1_uploads[0].__class__.objects.count() == 0
dispatch.assert_any_call(
{"type": "Delete", "object": {"type": "Audio"}},
context={"uploads": library1_uploads},
)
dispatch.assert_any_call(
{"type": "Delete", "object": {"type": "Audio"}},
context={"uploads": library2_uploads},
)
def test_manage_track_file_action_relaunch_import(factories, mocker):
def test_manage_upload_action_relaunch_import(factories, mocker):
m = mocker.patch("funkwhale_api.common.utils.on_commit")
# this one is finished and should stay as is
finished = factories["music.TrackFile"](import_status="finished")
finished = factories["music.Upload"](import_status="finished")
to_relaunch = [
factories["music.TrackFile"](import_status="pending"),
factories["music.TrackFile"](import_status="skipped"),
factories["music.TrackFile"](import_status="errored"),
factories["music.Upload"](import_status="pending"),
factories["music.Upload"](import_status="skipped"),
factories["music.Upload"](import_status="errored"),
]
s = serializers.TrackFileActionSerializer(queryset=None)
s = serializers.UploadActionSerializer(queryset=None)
s.handle_relaunch_import(models.TrackFile.objects.all())
s.handle_relaunch_import(models.Upload.objects.all())
for obj in to_relaunch:
obj.refresh_from_db()
assert obj.import_status == "pending"
m.assert_any_call(tasks.import_track_file.delay, track_file_id=obj.pk)
m.assert_any_call(tasks.import_upload.delay, upload_id=obj.pk)
finished.refresh_from_db()
assert finished.import_status == "finished"

147
api/tests/music/test_tasks.py
View file

@ -11,7 +11,7 @@ from funkwhale_api.music import signals, tasks
DATA_DIR = os.path.dirname(os.path.abspath(__file__))
# DATA_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "files")
# DATA_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "uploads")
def test_can_create_track_from_file_metadata_no_mbid(db, mocker):
@ -89,64 +89,68 @@ def test_can_create_track_from_file_metadata_mbid(factories, mocker):
assert track.artist == artist
def test_track_file_import_mbid(now, factories, temp_signal):
def test_upload_import_mbid(now, factories, temp_signal, mocker):
outbox = mocker.patch("funkwhale_api.federation.routes.outbox.dispatch")
track = factories["music.Track"]()
tf = factories["music.TrackFile"](
upload = factories["music.Upload"](
track=None, import_metadata={"track": {"mbid": track.mbid}}
)
with temp_signal(signals.track_file_import_status_updated) as handler:
tasks.import_track_file(track_file_id=tf.pk)
with temp_signal(signals.upload_import_status_updated) as handler:
tasks.import_upload(upload_id=upload.pk)
tf.refresh_from_db()
upload.refresh_from_db()
assert tf.track == track
assert tf.import_status == "finished"
assert tf.import_date == now
assert upload.track == track
assert upload.import_status == "finished"
assert upload.import_date == now
handler.assert_called_once_with(
track_file=tf,
upload=upload,
old_status="pending",
new_status="finished",
sender=None,
signal=signals.track_file_import_status_updated,
signal=signals.upload_import_status_updated,
)
outbox.assert_called_once_with(
{"type": "Create", "object": {"type": "Audio"}}, context={"upload": upload}
)
def test_track_file_import_get_audio_data(factories, mocker):
def test_upload_import_get_audio_data(factories, mocker):
mocker.patch(
"funkwhale_api.music.models.TrackFile.get_audio_data",
"funkwhale_api.music.models.Upload.get_audio_data",
return_value={"size": 23, "duration": 42, "bitrate": 66},
)
track = factories["music.Track"]()
tf = factories["music.TrackFile"](
upload = factories["music.Upload"](
track=None, import_metadata={"track": {"mbid": track.mbid}}
)
tasks.import_track_file(track_file_id=tf.pk)
tasks.import_upload(upload_id=upload.pk)
tf.refresh_from_db()
assert tf.size == 23
assert tf.duration == 42
assert tf.bitrate == 66
upload.refresh_from_db()
assert upload.size == 23
assert upload.duration == 42
assert upload.bitrate == 66
def test_track_file_import_skip_existing_track_in_own_library(factories, temp_signal):
def test_upload_import_skip_existing_track_in_own_library(factories, temp_signal):
track = factories["music.Track"]()
library = factories["music.Library"]()
existing = factories["music.TrackFile"](
existing = factories["music.Upload"](
track=track,
import_status="finished",
library=library,
import_metadata={"track": {"mbid": track.mbid}},
)
duplicate = factories["music.TrackFile"](
duplicate = factories["music.Upload"](
track=track,
import_status="pending",
library=library,
import_metadata={"track": {"mbid": track.mbid}},
)
with temp_signal(signals.track_file_import_status_updated) as handler:
tasks.import_track_file(track_file_id=duplicate.pk)
with temp_signal(signals.upload_import_status_updated) as handler:
tasks.import_upload(upload_id=duplicate.pk)
duplicate.refresh_from_db()
@ -157,78 +161,80 @@ def test_track_file_import_skip_existing_track_in_own_library(factories, temp_si
}
handler.assert_called_once_with(
track_file=duplicate,
upload=duplicate,
old_status="pending",
new_status="skipped",
sender=None,
signal=signals.track_file_import_status_updated,
signal=signals.upload_import_status_updated,
)
def test_track_file_import_track_uuid(now, factories):
def test_upload_import_track_uuid(now, factories):
track = factories["music.Track"]()
tf = factories["music.TrackFile"](
upload = factories["music.Upload"](
track=None, import_metadata={"track": {"uuid": track.uuid}}
)
tasks.import_track_file(track_file_id=tf.pk)
tasks.import_upload(upload_id=upload.pk)
tf.refresh_from_db()
upload.refresh_from_db()
assert tf.track == track
assert tf.import_status == "finished"
assert tf.import_date == now
assert upload.track == track
assert upload.import_status == "finished"
assert upload.import_date == now
def test_track_file_import_error(factories, now, temp_signal):
tf = factories["music.TrackFile"](import_metadata={"track": {"uuid": uuid.uuid4()}})
with temp_signal(signals.track_file_import_status_updated) as handler:
tasks.import_track_file(track_file_id=tf.pk)
tf.refresh_from_db()
def test_upload_import_error(factories, now, temp_signal):
upload = factories["music.Upload"](
import_metadata={"track": {"uuid": uuid.uuid4()}}
)
with temp_signal(signals.upload_import_status_updated) as handler:
tasks.import_upload(upload_id=upload.pk)
upload.refresh_from_db()
assert tf.import_status == "errored"
assert tf.import_date == now
assert tf.import_details == {"error_code": "track_uuid_not_found"}
assert upload.import_status == "errored"
assert upload.import_date == now
assert upload.import_details == {"error_code": "track_uuid_not_found"}
handler.assert_called_once_with(
track_file=tf,
upload=upload,
old_status="pending",
new_status="errored",
sender=None,
signal=signals.track_file_import_status_updated,
signal=signals.upload_import_status_updated,
)
def test_track_file_import_updates_cover_if_no_cover(factories, mocker, now):
def test_upload_import_updates_cover_if_no_cover(factories, mocker, now):
mocked_update = mocker.patch("funkwhale_api.music.tasks.update_album_cover")
album = factories["music.Album"](cover="")
track = factories["music.Track"](album=album)
tf = factories["music.TrackFile"](
upload = factories["music.Upload"](
track=None, import_metadata={"track": {"uuid": track.uuid}}
)
tasks.import_track_file(track_file_id=tf.pk)
mocked_update.assert_called_once_with(album, tf)
tasks.import_upload(upload_id=upload.pk)
mocked_update.assert_called_once_with(album, upload)
def test_update_album_cover_mbid(factories, mocker):
album = factories["music.Album"](cover="")
mocked_get = mocker.patch("funkwhale_api.music.models.Album.get_image")
tasks.update_album_cover(album=album, track_file=None)
tasks.update_album_cover(album=album, upload=None)
mocked_get.assert_called_once_with()
def test_update_album_cover_file_data(factories, mocker):
album = factories["music.Album"](cover="", mbid=None)
tf = factories["music.TrackFile"](track__album=album)
upload = factories["music.Upload"](track__album=album)
mocked_get = mocker.patch("funkwhale_api.music.models.Album.get_image")
mocker.patch(
"funkwhale_api.music.metadata.Metadata.get_picture",
return_value={"hello": "world"},
)
tasks.update_album_cover(album=album, track_file=tf)
tf.get_metadata()
tasks.update_album_cover(album=album, upload=upload)
upload.get_metadata()
mocked_get.assert_called_once_with(data={"hello": "world"})
@ -239,12 +245,14 @@ def test_update_album_cover_file_cover_separate_file(ext, mimetype, factories, m
with open(image_path, "rb") as f:
image_content = f.read()
album = factories["music.Album"](cover="", mbid=None)
tf = factories["music.TrackFile"](track__album=album, source="file://" + image_path)
upload = factories["music.Upload"](
track__album=album, source="file://" + image_path
)
mocked_get = mocker.patch("funkwhale_api.music.models.Album.get_image")
mocker.patch("funkwhale_api.music.metadata.Metadata.get_picture", return_value=None)
tasks.update_album_cover(album=album, track_file=tf)
tf.get_metadata()
tasks.update_album_cover(album=album, upload=upload)
upload.get_metadata()
mocked_get.assert_called_once_with(
data={"mimetype": mimetype, "content": image_content}
)
@ -275,17 +283,23 @@ def test_scan_library_fetches_page_and_calls_scan_page(now, mocker, factories, r
def test_scan_page_fetches_page_and_creates_tracks(now, mocker, factories, r_mock):
scan_page = mocker.patch("funkwhale_api.music.tasks.scan_library_page.delay")
import_tf = mocker.patch("funkwhale_api.music.tasks.import_track_file.delay")
scan = factories["music.LibraryScan"](status="scanning", total_files=5)
tfs = factories["music.TrackFile"].build_batch(size=5, library=scan.library)
for i, tf in enumerate(tfs):
tf.fid = "https://track.test/{}".format(i)
uploads = [
factories["music.Upload"].build(
fid="https://track.test/{}".format(i),
size=42,
bitrate=66,
duration=99,
library=scan.library,
)
for i in range(5)
]
page_conf = {
"actor": scan.library.actor,
"id": scan.library.fid,
"page": Paginator(tfs, 3).page(1),
"item_serializer": federation_serializers.AudioSerializer,
"page": Paginator(uploads, 3).page(1),
"item_serializer": federation_serializers.UploadSerializer,
}
page = federation_serializers.CollectionPageSerializer(page_conf)
r_mock.get(page.data["id"], json=page.data)
@ -293,12 +307,11 @@ def test_scan_page_fetches_page_and_creates_tracks(now, mocker, factories, r_moc
tasks.scan_library_page(library_scan_id=scan.pk, page_url=page.data["id"])
scan.refresh_from_db()
lts = list(scan.library.files.all().order_by("-creation_date"))
lts = list(scan.library.uploads.all().order_by("-creation_date"))
assert len(lts) == 3
for tf in tfs[:3]:
new_tf = scan.library.files.get(fid=tf.get_federation_id())
import_tf.assert_any_call(track_file_id=new_tf.pk)
for upload in uploads[:3]:
scan.library.uploads.get(fid=upload.fid)
assert scan.status == "scanning"
assert scan.processed_files == 3
@ -312,12 +325,12 @@ def test_scan_page_fetches_page_and_creates_tracks(now, mocker, factories, r_moc
def test_scan_page_trigger_next_page_scan_skip_if_same(mocker, factories, r_mock):
patched_scan = mocker.patch("funkwhale_api.music.tasks.scan_library_page.delay")
scan = factories["music.LibraryScan"](status="scanning", total_files=5)
tfs = factories["music.TrackFile"].build_batch(size=5, library=scan.library)
uploads = factories["music.Upload"].build_batch(size=5, library=scan.library)
page_conf = {
"actor": scan.library.actor,
"id": scan.library.fid,
"page": Paginator(tfs, 3).page(1),
"item_serializer": federation_serializers.AudioSerializer,
"page": Paginator(uploads, 3).page(1),
"item_serializer": federation_serializers.UploadSerializer,
}
page = federation_serializers.CollectionPageSerializer(page_conf)
data = page.data

4
api/tests/music/test_utils.py
View file

@ -9,7 +9,7 @@ DATA_DIR = os.path.dirname(os.path.abspath(__file__))
def test_guess_mimetype_try_using_extension(factories, mocker):
mocker.patch("magic.from_buffer", return_value="audio/mpeg")
f = factories["music.TrackFile"].build(audio_file__filename="test.ogg")
f = factories["music.Upload"].build(audio_file__filename="test.ogg")
assert utils.guess_mimetype(f.audio_file) == "audio/mpeg"
@ -17,7 +17,7 @@ def test_guess_mimetype_try_using_extension(factories, mocker):
@pytest.mark.parametrize("wrong", ["application/octet-stream", "application/x-empty"])
def test_guess_mimetype_try_using_extension_if_fail(wrong, factories, mocker):
mocker.patch("magic.from_buffer", return_value=wrong)
f = factories["music.TrackFile"].build(audio_file__filename="test.mp3")
f = factories["music.Upload"].build(audio_file__filename="test.mp3")
assert utils.guess_mimetype(f.audio_file) == "audio/mpeg"

143
api/tests/music/test_views.py
View file

@ -12,7 +12,7 @@ DATA_DIR = os.path.dirname(os.path.abspath(__file__))
def test_artist_list_serializer(api_request, factories, logged_in_api_client):
track = factories["music.TrackFile"](library__privacy_level="everyone").track
track = factories["music.Upload"](library__privacy_level="everyone").track
artist = track.artist
request = api_request.get("/")
qs = artist.__class__.objects.with_albums()
@ -20,6 +20,9 @@ def test_artist_list_serializer(api_request, factories, logged_in_api_client):
qs, many=True, context={"request": request}
)
expected = {"count": 1, "next": None, "previous": None, "results": serializer.data}
for artist in serializer.data:
for album in artist["albums"]:
album["is_playable"] = True
url = reverse("api:v1:artists-list")
response = logged_in_api_client.get(url)
@ -28,7 +31,7 @@ def test_artist_list_serializer(api_request, factories, logged_in_api_client):
def test_album_list_serializer(api_request, factories, logged_in_api_client):
track = factories["music.TrackFile"](library__privacy_level="everyone").track
track = factories["music.Upload"](library__privacy_level="everyone").track
album = track.album
request = api_request.get("/")
qs = album.__class__.objects.all()
@ -46,7 +49,7 @@ def test_album_list_serializer(api_request, factories, logged_in_api_client):
def test_track_list_serializer(api_request, factories, logged_in_api_client):
track = factories["music.TrackFile"](library__privacy_level="everyone").track
track = factories["music.Upload"](library__privacy_level="everyone").track
request = api_request.get("/")
qs = track.__class__.objects.all()
serializer = serializers.TrackSerializer(
@ -65,7 +68,7 @@ def test_track_list_serializer(api_request, factories, logged_in_api_client):
def test_artist_view_filter_playable(param, expected, factories, api_request):
artists = {
"empty": factories["music.Artist"](),
"full": factories["music.TrackFile"](
"full": factories["music.Upload"](
library__privacy_level="everyone"
).track.artist,
}
@ -84,7 +87,7 @@ def test_artist_view_filter_playable(param, expected, factories, api_request):
def test_album_view_filter_playable(param, expected, factories, api_request):
artists = {
"empty": factories["music.Album"](),
"full": factories["music.TrackFile"](
"full": factories["music.Upload"](
library__privacy_level="everyone"
).track.album,
}
@ -99,32 +102,32 @@ def test_album_view_filter_playable(param, expected, factories, api_request):
assert list(queryset) == expected
def test_can_serve_track_file_as_remote_library(
def test_can_serve_upload_as_remote_library(
factories, authenticated_actor, logged_in_api_client, settings, preferences
):
preferences["common__api_authentication_required"] = True
track_file = factories["music.TrackFile"](library__privacy_level="everyone")
library_actor = track_file.library.actor
upload = factories["music.Upload"](library__privacy_level="everyone")
library_actor = upload.library.actor
factories["federation.Follow"](
approved=True, actor=authenticated_actor, target=library_actor
)
response = logged_in_api_client.get(track_file.track.listen_url)
response = logged_in_api_client.get(upload.track.listen_url)
assert response.status_code == 200
assert response["X-Accel-Redirect"] == "{}{}".format(
settings.PROTECT_FILES_PATH, track_file.audio_file.url
settings.PROTECT_FILES_PATH, upload.audio_file.url
)
def test_can_serve_track_file_as_remote_library_deny_not_following(
def test_can_serve_upload_as_remote_library_deny_not_following(
factories, authenticated_actor, settings, api_client, preferences
):
preferences["common__api_authentication_required"] = True
track_file = factories["music.TrackFile"](library__privacy_level="everyone")
response = api_client.get(track_file.track.listen_url)
upload = factories["music.Upload"](library__privacy_level="instance")
response = api_client.get(upload.track.listen_url)
assert response.status_code == 403
assert response.status_code == 404
@pytest.mark.parametrize(
@ -145,12 +148,12 @@ def test_serve_file_in_place(
settings.REVERSE_PROXY_TYPE = proxy
settings.MUSIC_DIRECTORY_PATH = "/app/music"
settings.MUSIC_DIRECTORY_SERVE_PATH = serve_path
tf = factories["music.TrackFile"](
upload = factories["music.Upload"](
in_place=True,
source="file:///app/music/hello/world.mp3",
library__privacy_level="everyone",
)
response = api_client.get(tf.track.listen_url)
response = api_client.get(upload.track.listen_url)
assert response.status_code == 200
assert response[headers[proxy]] == expected
@ -199,9 +202,11 @@ def test_serve_file_media(
settings.MUSIC_DIRECTORY_PATH = "/app/music"
settings.MUSIC_DIRECTORY_SERVE_PATH = serve_path
tf = factories["music.TrackFile"](library__privacy_level="everyone")
tf.__class__.objects.filter(pk=tf.pk).update(audio_file="tracks/hello/world.mp3")
response = api_client.get(tf.track.listen_url)
upload = factories["music.Upload"](library__privacy_level="everyone")
upload.__class__.objects.filter(pk=upload.pk).update(
audio_file="tracks/hello/world.mp3"
)
response = api_client.get(upload.track.listen_url)
assert response.status_code == 200
assert response[headers[proxy]] == expected
@ -210,32 +215,32 @@ def test_serve_file_media(
def test_can_proxy_remote_track(factories, settings, api_client, r_mock, preferences):
preferences["common__api_authentication_required"] = False
url = "https://file.test"
track_file = factories["music.TrackFile"](
upload = factories["music.Upload"](
library__privacy_level="everyone", audio_file="", source=url
)
r_mock.get(url, body=io.BytesIO(b"test"))
response = api_client.get(track_file.track.listen_url)
track_file.refresh_from_db()
response = api_client.get(upload.track.listen_url)
upload.refresh_from_db()
assert response.status_code == 200
assert response["X-Accel-Redirect"] == "{}{}".format(
settings.PROTECT_FILES_PATH, track_file.audio_file.url
settings.PROTECT_FILES_PATH, upload.audio_file.url
)
assert track_file.audio_file.read() == b"test"
assert upload.audio_file.read() == b"test"
def test_serve_updates_access_date(factories, settings, api_client, preferences):
preferences["common__api_authentication_required"] = False
track_file = factories["music.TrackFile"](library__privacy_level="everyone")
upload = factories["music.Upload"](library__privacy_level="everyone")
now = timezone.now()
assert track_file.accessed_date is None
assert upload.accessed_date is None
response = api_client.get(track_file.track.listen_url)
track_file.refresh_from_db()
response = api_client.get(upload.track.listen_url)
upload.refresh_from_db()
assert response.status_code == 200
assert track_file.accessed_date > now
assert upload.accessed_date > now
def test_listen_no_track(factories, logged_in_api_client):
@ -254,8 +259,8 @@ def test_listen_no_file(factories, logged_in_api_client):
def test_listen_no_available_file(factories, logged_in_api_client):
tf = factories["music.TrackFile"]()
url = reverse("api:v1:listen-detail", kwargs={"uuid": tf.track.uuid})
upload = factories["music.Upload"]()
url = reverse("api:v1:listen-detail", kwargs={"uuid": upload.track.uuid})
response = logged_in_api_client.get(url)
assert response.status_code == 404
@ -263,10 +268,10 @@ def test_listen_no_available_file(factories, logged_in_api_client):
def test_listen_correct_access(factories, logged_in_api_client):
logged_in_api_client.user.create_actor()
tf = factories["music.TrackFile"](
upload = factories["music.Upload"](
library__actor=logged_in_api_client.user.actor, library__privacy_level="me"
)
url = reverse("api:v1:listen-detail", kwargs={"uuid": tf.track.uuid})
url = reverse("api:v1:listen-detail", kwargs={"uuid": upload.track.uuid})
response = logged_in_api_client.get(url)
assert response.status_code == 200
@ -274,15 +279,15 @@ def test_listen_correct_access(factories, logged_in_api_client):
def test_listen_explicit_file(factories, logged_in_api_client, mocker):
mocked_serve = mocker.spy(views, "handle_serve")
tf1 = factories["music.TrackFile"](library__privacy_level="everyone")
tf2 = factories["music.TrackFile"](
library__privacy_level="everyone", track=tf1.track
upload1 = factories["music.Upload"](library__privacy_level="everyone")
upload2 = factories["music.Upload"](
library__privacy_level="everyone", track=upload1.track
)
url = reverse("api:v1:listen-detail", kwargs={"uuid": tf2.track.uuid})
response = logged_in_api_client.get(url, {"file": tf2.uuid})
url = reverse("api:v1:listen-detail", kwargs={"uuid": upload2.track.uuid})
response = logged_in_api_client.get(url, {"upload": upload2.uuid})
assert response.status_code == 200
mocked_serve.assert_called_once_with(tf2, user=logged_in_api_client.user)
mocked_serve.assert_called_once_with(upload2, user=logged_in_api_client.user)
def test_user_can_create_library(factories, logged_in_api_client):
@ -327,42 +332,60 @@ def test_user_cannot_delete_other_actors_library(factories, logged_in_api_client
assert response.status_code == 404
def test_user_cannot_get_other_actors_files(factories, logged_in_api_client):
logged_in_api_client.user.create_actor()
track_file = factories["music.TrackFile"]()
def test_library_delete_via_api_triggers_outbox(factories, mocker):
dispatch = mocker.patch("funkwhale_api.federation.routes.outbox.dispatch")
library = factories["music.Library"]()
view = views.LibraryViewSet()
view.perform_destroy(library)
dispatch.assert_called_once_with(
{"type": "Delete", "object": {"type": "Library"}}, context={"library": library}
)
url = reverse("api:v1:trackfiles-detail", kwargs={"uuid": track_file.uuid})
def test_user_cannot_get_other_actors_uploads(factories, logged_in_api_client):
logged_in_api_client.user.create_actor()
upload = factories["music.Upload"]()
url = reverse("api:v1:uploads-detail", kwargs={"uuid": upload.uuid})
response = logged_in_api_client.get(url)
assert response.status_code == 404
def test_user_cannot_delete_other_actors_files(factories, logged_in_api_client):
def test_user_cannot_delete_other_actors_uploads(factories, logged_in_api_client):
logged_in_api_client.user.create_actor()
track_file = factories["music.TrackFile"]()
upload = factories["music.Upload"]()
url = reverse("api:v1:trackfiles-detail", kwargs={"uuid": track_file.uuid})
url = reverse("api:v1:uploads-detail", kwargs={"uuid": upload.uuid})
response = logged_in_api_client.delete(url)
assert response.status_code == 404
def test_user_cannot_list_other_actors_files(factories, logged_in_api_client):
logged_in_api_client.user.create_actor()
factories["music.TrackFile"]()
def test_upload_delete_via_api_triggers_outbox(factories, mocker):
dispatch = mocker.patch("funkwhale_api.federation.routes.outbox.dispatch")
upload = factories["music.Upload"]()
view = views.UploadViewSet()
view.perform_destroy(upload)
dispatch.assert_called_once_with(
{"type": "Delete", "object": {"type": "Audio"}}, context={"uploads": [upload]}
)
url = reverse("api:v1:trackfiles-list")
def test_user_cannot_list_other_actors_uploads(factories, logged_in_api_client):
logged_in_api_client.user.create_actor()
factories["music.Upload"]()
url = reverse("api:v1:uploads-list")
response = logged_in_api_client.get(url)
assert response.status_code == 200
assert response.data["count"] == 0
def test_user_can_create_track_file(
logged_in_api_client, factories, mocker, audio_file
):
def test_user_can_create_upload(logged_in_api_client, factories, mocker, audio_file):
library = factories["music.Library"](actor__user=logged_in_api_client.user)
url = reverse("api:v1:trackfiles-list")
url = reverse("api:v1:uploads-list")
m = mocker.patch("funkwhale_api.common.utils.on_commit")
response = logged_in_api_client.post(
@ -377,14 +400,14 @@ def test_user_can_create_track_file(
assert response.status_code == 201
tf = library.files.latest("id")
upload = library.uploads.latest("id")
audio_file.seek(0)
assert tf.audio_file.read() == audio_file.read()
assert tf.source == "upload://test"
assert tf.import_reference == "test"
assert tf.track is None
m.assert_called_once_with(tasks.import_track_file.delay, track_file_id=tf.pk)
assert upload.audio_file.read() == audio_file.read()
assert upload.source == "upload://test"
assert upload.import_reference == "test"
assert upload.track is None
m.assert_called_once_with(tasks.import_upload.delay, upload_id=upload.pk)
def test_user_can_list_own_library_follows(factories, logged_in_api_client):