funkwhale/api/tests/conftest.py

419 wiersze
10 KiB
Python
Czysty Zwykły widok Historia

import contextlib
import datetime
2018-07-13 12:10:39 +00:00
import io
import os
2018-07-13 12:10:39 +00:00
import PIL
import random
import shutil
import tempfile
2018-09-22 12:29:30 +00:00
import uuid
2018-09-22 12:29:30 +00:00
from faker.providers import internet as internet_provider
2018-06-10 08:55:16 +00:00
import factory
import pytest
2018-09-23 12:38:42 +00:00
from django.core.management import call_command
2018-03-25 15:17:51 +00:00
from django.contrib.auth.models import AnonymousUser
from django.core.cache import cache as django_cache, caches
from django.core.files import uploadedfile
from django.utils import timezone
from django.test import client
from django.db import connection
from django.db.migrations.executor import MigrationExecutor
2018-09-13 15:18:23 +00:00
from django.db.models import QuerySet
from dynamic_preferences.registries import global_preferences_registry
from rest_framework import fields as rest_fields
2018-06-10 08:55:16 +00:00
from rest_framework.test import APIClient, APIRequestFactory
from funkwhale_api.activity import record
from funkwhale_api.users.permissions import HasUserPermission
2018-09-22 12:29:30 +00:00
class FunkwhaleProvider(internet_provider.Provider):
"""
Our own faker data generator, since built-in ones are sometimes
not random enough
"""
def federation_url(self, prefix=""):
def path_generator():
return "{}/{}".format(prefix, uuid.uuid4())
domain = self.domain_name()
protocol = "https"
path = path_generator()
return "{}://{}/{}".format(protocol, domain, path)
factory.Faker.add_provider(FunkwhaleProvider)
2018-09-13 15:18:23 +00:00
@pytest.fixture
def queryset_equal_queries():
"""
Unitting querysets is hard because we have to compare queries
by hand. Let's monkey patch querysets to do that for us.
"""
def __eq__(self, other):
if isinstance(other, QuerySet):
return str(other.query) == str(self.query)
else:
return False
setattr(QuerySet, "__eq__", __eq__)
yield __eq__
delattr(QuerySet, "__eq__")
@pytest.fixture
def queryset_equal_list():
"""
Unitting querysets is hard because we usually simply wants to ensure
a querysets contains the same objects as a list, let's monkey patch
querysets to to that for us.
"""
def __eq__(self, other):
if isinstance(other, (list, tuple)):
return list(self) == list(other)
else:
return False
setattr(QuerySet, "__eq__", __eq__)
yield __eq__
delattr(QuerySet, "__eq__")
@pytest.fixture(scope="session", autouse=True)
def factories_autodiscover():
from django.apps import apps
from funkwhale_api import factories
2018-06-09 13:36:16 +00:00
app_names = [app.name for app in apps.app_configs.values()]
factories.registry.autodiscover(app_names)
@pytest.fixture(autouse=True)
def cache():
2018-04-28 16:39:48 +00:00
"""
Returns a django Cache instance for cache-related operations
"""
yield django_cache
django_cache.clear()
@pytest.fixture(autouse=True)
def local_cache():
yield caches["local"]
caches["local"].clear()
@pytest.fixture
def factories(db):
2018-04-28 16:39:48 +00:00
"""
Returns a dictionnary containing all registered factories with keys such as
users.User or music.Track
"""
from funkwhale_api import factories
2018-06-09 13:36:16 +00:00
for v in factories.registry.values():
try:
v._meta.strategy = factory.CREATE_STRATEGY
except AttributeError:
# probably not a class based factory
pass
yield factories.registry
@pytest.fixture
def nodb_factories():
2018-04-28 16:39:48 +00:00
"""
Returns a dictionnary containing all registered factories with a build strategy
that does not require access to the database
"""
from funkwhale_api import factories
2018-06-09 13:36:16 +00:00
for v in factories.registry.values():
try:
v._meta.strategy = factory.BUILD_STRATEGY
except AttributeError:
# probably not a class based factory
pass
yield factories.registry
@pytest.fixture
def preferences(db, cache):
2018-04-28 16:39:48 +00:00
"""
return a dynamic_preferences manager for global_preferences
"""
manager = global_preferences_registry.manager()
manager.all()
yield manager
@pytest.fixture
def tmpdir():
2018-04-28 16:39:48 +00:00
"""
Returns a temporary directory path where you can write things during your
test
"""
d = tempfile.mkdtemp()
yield d
shutil.rmtree(d)
2018-04-16 19:57:53 +00:00
@pytest.fixture
def tmpfile():
2018-04-28 16:39:48 +00:00
"""
Returns a temporary file where you can write things during your test
"""
2018-04-16 19:57:53 +00:00
yield tempfile.NamedTemporaryFile()
@pytest.fixture
def logged_in_client(db, factories, client):
2018-04-28 16:39:48 +00:00
"""
Returns a logged-in, non-API client with an authenticated ``User``
stored in the ``user`` attribute
"""
2018-06-09 13:36:16 +00:00
user = factories["users.User"]()
assert client.login(username=user.username, password="test")
setattr(client, "user", user)
yield client
2018-06-09 13:36:16 +00:00
delattr(client, "user")
2018-03-25 15:17:51 +00:00
@pytest.fixture
def anonymous_user():
2018-04-28 16:39:48 +00:00
"""Returns a AnonymousUser() instance"""
2018-03-25 15:17:51 +00:00
return AnonymousUser()
@pytest.fixture
def api_client(client):
2018-04-28 16:39:48 +00:00
"""
Return an API client without any authentication
"""
return APIClient()
2018-02-20 22:59:50 +00:00
@pytest.fixture
def logged_in_api_client(db, factories, api_client):
2018-04-28 16:39:48 +00:00
"""
Return a logged-in API client with an authenticated ``User``
stored in the ``user`` attribute
"""
2018-06-09 13:36:16 +00:00
user = factories["users.User"]()
assert api_client.login(username=user.username, password="test")
api_client.force_authenticate(user=user)
2018-06-09 13:36:16 +00:00
setattr(api_client, "user", user)
2018-02-20 22:59:50 +00:00
yield api_client
2018-06-09 13:36:16 +00:00
delattr(api_client, "user")
2018-02-20 22:59:50 +00:00
2018-02-22 22:33:29 +00:00
@pytest.fixture
def superuser_api_client(db, factories, api_client):
2018-04-28 16:39:48 +00:00
"""
Return a logged-in API client with an authenticated superuser
stored in the ``user`` attribute
"""
2018-06-09 13:36:16 +00:00
user = factories["users.SuperUser"]()
assert api_client.login(username=user.username, password="test")
setattr(api_client, "user", user)
2018-02-22 22:33:29 +00:00
yield api_client
2018-06-09 13:36:16 +00:00
delattr(api_client, "user")
2018-02-22 22:33:29 +00:00
@pytest.fixture
def superuser_client(db, factories, client):
2018-04-28 16:39:48 +00:00
"""
Return a logged-in, non-API client with an authenticated ``User``
stored in the ``user`` attribute
"""
2018-06-09 13:36:16 +00:00
user = factories["users.SuperUser"]()
assert client.login(username=user.username, password="test")
setattr(client, "user", user)
yield client
2018-06-09 13:36:16 +00:00
delattr(client, "user")
@pytest.fixture
def api_request():
2018-04-28 16:39:48 +00:00
"""
Returns a dummy API request object you can pass to API views
"""
return APIRequestFactory()
@pytest.fixture
def fake_request():
2018-04-28 16:39:48 +00:00
"""
Returns a dummy, non-API request object you can pass to regular views
"""
return client.RequestFactory()
@pytest.fixture
def activity_registry():
state = list(record.registry.items())
yield record.registry
record.registry.clear()
for key, value in state:
record.registry[key] = value
@pytest.fixture
def activity_muted(activity_registry, mocker):
2018-06-09 13:36:16 +00:00
yield mocker.patch.object(record, "send")
2018-03-25 20:34:30 +00:00
@pytest.fixture(autouse=True)
def media_root(settings):
2018-04-28 16:39:48 +00:00
"""
Sets settings.MEDIA_ROOT to a temporary path and returns this path
"""
2018-03-25 20:34:30 +00:00
tmp_dir = tempfile.mkdtemp()
settings.MEDIA_ROOT = tmp_dir
yield settings.MEDIA_ROOT
shutil.rmtree(tmp_dir)
2018-09-24 18:44:22 +00:00
@pytest.fixture(autouse=True)
def disabled_musicbrainz(mocker):
# we ensure no music brainz requests gets out
yield mocker.patch(
"musicbrainzngs.musicbrainz._safe_read",
side_effect=Exception("Disabled network calls"),
)
2018-09-23 12:38:42 +00:00
@pytest.fixture(autouse=True)
def r_mock(requests_mock):
2018-04-28 16:39:48 +00:00
"""
Returns a requests_mock.mock() object you can use to mock HTTP calls made
using python-requests
"""
2018-09-23 12:38:42 +00:00
yield requests_mock
@pytest.fixture
def authenticated_actor(factories, mocker):
2018-04-28 16:39:48 +00:00
"""
Returns an authenticated ActivityPub actor
"""
2018-06-09 13:36:16 +00:00
actor = factories["federation.Actor"]()
mocker.patch(
2018-06-09 13:36:16 +00:00
"funkwhale_api.federation.authentication.SignatureAuthentication.authenticate_actor",
return_value=actor,
)
yield actor
@pytest.fixture
def assert_user_permission():
2018-06-09 13:36:16 +00:00
def inner(view, permissions, operator="and"):
assert HasUserPermission in view.permission_classes
2018-06-09 13:36:16 +00:00
assert getattr(view, "permission_operator", "and") == operator
assert set(view.required_permissions) == set(permissions)
2018-06-09 13:36:16 +00:00
return inner
@pytest.fixture
def to_api_date():
def inner(value):
if isinstance(value, datetime.datetime):
f = rest_fields.DateTimeField()
return f.to_representation(value)
if isinstance(value, datetime.date):
f = rest_fields.DateField()
return f.to_representation(value)
2018-06-09 13:36:16 +00:00
raise ValueError("Invalid value: {}".format(value))
return inner
@pytest.fixture()
def now(mocker):
now = timezone.now()
mocker.patch("django.utils.timezone.now", return_value=now)
return now
2018-07-13 12:10:39 +00:00
@pytest.fixture()
def avatar():
i = PIL.Image.new("RGBA", (400, 400), random.choice(["red", "blue", "yellow"]))
f = io.BytesIO()
i.save(f, "png")
f.name = "avatar.png"
f.seek(0)
yield f
f.close()
@pytest.fixture()
def audio_file():
data_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "music")
path = os.path.join(data_dir, "test.ogg")
assert os.path.exists(path)
with open(path, "rb") as f:
yield f
@pytest.fixture()
def uploaded_audio_file(audio_file):
yield uploadedfile.SimpleUploadedFile(
name=audio_file.name, content=audio_file.read()
)
@pytest.fixture()
def temp_signal(mocker):
"""
Connect a temporary handler to a given signal. This is helpful to validate
a signal is dispatched properly, without mocking.
"""
@contextlib.contextmanager
def connect(signal):
stub = mocker.stub()
signal.connect(stub)
try:
yield stub
finally:
signal.disconnect(stub)
return connect
@pytest.fixture()
def stdout():
yield io.StringIO()
@pytest.fixture
def spa_html(r_mock, settings):
yield r_mock.get(
settings.FUNKWHALE_SPA_HTML_ROOT + "index.html", text="<head></head>"
)
@pytest.fixture
def no_api_auth(preferences):
preferences["common__api_authentication_required"] = False
@pytest.fixture()
def migrator(transactional_db):
yield MigrationExecutor(connection)
call_command("migrate", interactive=False)
@pytest.fixture(autouse=True)
def rsa_small_key(settings):
# smaller size for faster generation, since it's CPU hungry
settings.RSA_KEY_SIZE = 512