funkwhale/api/tests/federation/test_models.py

285 wiersze
8.1 KiB
Python

import pytest
from django import db
from funkwhale_api.federation import models
from funkwhale_api.music import models as music_models
def test_cannot_duplicate_actor(factories):
actor = factories["federation.Actor"]()
with pytest.raises(db.IntegrityError):
factories["federation.Actor"](
domain=actor.domain, preferred_username=actor.preferred_username
)
def test_cannot_duplicate_follow(factories):
follow = factories["federation.Follow"]()
with pytest.raises(db.IntegrityError):
factories["federation.Follow"](target=follow.target, actor=follow.actor)
def test_follow_federation_url(factories):
follow = factories["federation.Follow"](local=True)
expected = f"{follow.actor.fid}#follows/{follow.uuid}"
assert follow.get_federation_id() == expected
def test_actor_get_quota(factories):
library = factories["music.Library"]()
factories["music.Upload"](
library=library,
import_status="pending",
audio_file__from_path=None,
audio_file__data=b"a",
)
factories["music.Upload"](
library=library,
import_status="skipped",
audio_file__from_path=None,
audio_file__data=b"aa",
)
factories["music.Upload"](
library=library,
import_status="errored",
audio_file__from_path=None,
audio_file__data=b"aaa",
)
factories["music.Upload"](
library=library,
import_status="finished",
audio_file__from_path=None,
audio_file__data=b"aaaa",
)
factories["music.Upload"](
library=library,
import_status="draft",
audio_file__from_path=None,
audio_file__data=b"aaaaa",
)
# this one is imported in place and don't count
factories["music.Upload"](
library=library,
import_status="finished",
source="file://test",
audio_file=None,
size=42,
)
# this one is imported in place but count because there is a mapped file
factories["music.Upload"](
library=library,
import_status="finished",
source="file://test2",
audio_file__from_path=None,
audio_file__data=b"aaaa",
)
# this one is in a channel
channel = factories["audio.Channel"](attributed_to=library.actor)
factories["music.Upload"](
library=channel.library,
import_status="finished",
audio_file__from_path=None,
audio_file__data=b"aaaaa",
)
expected = {
"total": 24,
"pending": 1,
"skipped": 2,
"errored": 3,
"finished": 13,
"draft": 5,
}
assert library.actor.get_current_usage() == expected
@pytest.mark.parametrize(
"value, expected",
[
("Domain.com", "domain.com"),
("hello-WORLD.com", "hello-world.com"),
("posés.com", "posés.com"),
],
)
def test_domain_name_saved_properly(value, expected, factories):
domain = factories["federation.Domain"](name=value)
assert domain.name == expected
def test_external_domains(factories, settings):
d1 = factories["federation.Domain"]()
d2 = factories["federation.Domain"]()
settings.FEDERATION_HOSTNAME = d1.pk
assert list(models.Domain.objects.external()) == [d2]
def test_domain_stats(factories):
expected = {
"actors": 0,
"libraries": 0,
"tracks": 0,
"albums": 0,
"channels": 0,
"uploads": 0,
"artists": 0,
"outbox_activities": 0,
"received_library_follows": 0,
"emitted_library_follows": 0,
"media_total_size": 0,
"media_downloaded_size": 0,
}
domain = factories["federation.Domain"]()
assert domain.get_stats() == expected
def test_actor_stats(factories):
expected = {
"libraries": 0,
"tracks": 0,
"albums": 0,
"uploads": 0,
"artists": 0,
"reports": 0,
"channels": 0,
"requests": 0,
"outbox_activities": 0,
"received_library_follows": 0,
"emitted_library_follows": 0,
"media_total_size": 0,
"media_downloaded_size": 0,
}
actor = factories["federation.Actor"]()
assert actor.get_stats() == expected
def test_actor_can_manage_false(mocker, factories):
obj = mocker.Mock()
actor = factories["federation.Actor"]()
assert actor.can_manage(obj) is False
def test_actor_can_manage_attributed_to(mocker, factories):
actor = factories["federation.Actor"]()
obj = mocker.Mock(attributed_to_id=actor.pk)
assert actor.can_manage(obj) is True
def test_actor_can_manage_domain_not_service_actor(mocker, factories):
actor = factories["federation.Actor"]()
obj = mocker.Mock(fid=f"https://{actor.domain_id}/hello")
assert actor.can_manage(obj) is False
def test_actor_can_manage_domain_service_actor(mocker, factories):
actor = factories["federation.Actor"]()
actor.domain.service_actor = actor
actor.domain.save()
obj = mocker.Mock(fid=f"https://{actor.domain_id}/hello")
assert actor.can_manage(obj) is True
def test_can_create_fetch_for_object(factories):
track = factories["music.Track"](fid="http://test.domain")
fetch = factories["federation.Fetch"](object=track)
assert fetch.url == "http://test.domain"
assert fetch.status == "pending"
assert fetch.detail == {}
assert fetch.object == track
@pytest.mark.parametrize(
"initial_approved, updated_approved, initial_playable_tracks, updated_playable_tracks",
[
(
True,
False,
{"owner": [0], "follower": [0], "local_actor": [], None: []},
{"owner": [0], "follower": [], "local_actor": [], None: []},
),
(
False,
True,
{"owner": [0], "follower": [], "local_actor": [], None: []},
{"owner": [0], "follower": [0], "local_actor": [], None: []},
),
],
)
def test_update_library_follow_approved_create_entries(
initial_approved,
updated_approved,
initial_playable_tracks,
updated_playable_tracks,
factories,
):
actors = {
"owner": factories["federation.Actor"](local=True),
"follower": factories["federation.Actor"](local=True),
"local_actor": factories["federation.Actor"](local=True),
None: None,
}
library = factories["music.Library"](actor=actors["owner"], privacy_level="me")
tracks = [
factories["music.Upload"](playable=True, library=library).track,
factories["music.Upload"](library=library, import_status="pending").track,
]
follow = factories["federation.LibraryFollow"](
target=library, actor=actors["follower"], approved=initial_approved
)
for actor_name, expected in initial_playable_tracks.items():
actor = actors[actor_name]
expected_tracks = [tracks[i] for i in expected]
assert list(music_models.Track.objects.playable_by(actor)) == expected_tracks
follow.approved = updated_approved
follow.save()
for actor_name, expected in updated_playable_tracks.items():
actor = actors[actor_name]
expected_tracks = [tracks[i] for i in expected]
assert list(music_models.Track.objects.playable_by(actor)) == expected_tracks
def test_update_library_follow_delete_delete_denormalization_entries(
factories,
):
updated_playable_tracks = {"owner": [0], "follower": []}
actors = {
"owner": factories["federation.Actor"](local=True),
"follower": factories["federation.Actor"](local=True),
}
library = factories["music.Library"](actor=actors["owner"], privacy_level="me")
tracks = [
factories["music.Upload"](playable=True, library=library).track,
factories["music.Upload"](library=library, import_status="pending").track,
]
follow = factories["federation.LibraryFollow"](
target=library, actor=actors["follower"], approved=True
)
follow.delete()
for actor_name, expected in updated_playable_tracks.items():
actor = actors[actor_name]
expected_tracks = [tracks[i] for i in expected]
assert list(music_models.Track.objects.playable_by(actor)) == expected_tracks