test_tasks.py

import os
import pytest

from funkwhale_api.providers.acoustid import get_acoustid_client
from funkwhale_api.music import tasks

DATA_DIR = os.path.dirname(os.path.abspath(__file__))


def test_set_acoustid_on_track_file(factories, mocker, preferences):
    preferences['providers_acoustid__api_key'] = 'test'
    track_file = factories['music.TrackFile'](acoustid_track_id=None)
    id = 'e475bf79-c1ce-4441-bed7-1e33f226c0a2'
    payload = {
        'results': [
            {'id': id,
             'recordings': [
                {'artists': [
                    {'id': '9c6bddde-6228-4d9f-ad0d-03f6fcb19e13',
                     'name': 'Binärpilot'}],
                 'duration': 268,
                 'id': 'f269d497-1cc0-4ae4-a0c4-157ec7d73fcb',
                 'title': 'Bend'}],
            'score': 0.860825}],
        'status': 'ok'
    }
    m = mocker.patch('acoustid.match', return_value=payload)
    r = tasks.set_acoustid_on_track_file(track_file_id=track_file.pk)
    track_file.refresh_from_db()

    assert str(track_file.acoustid_track_id) == id
    assert r == id
    m.assert_called_once_with('test', track_file.audio_file.path, parse=False)


def test_set_acoustid_on_track_file_required_high_score(factories, mocker):
    track_file = factories['music.TrackFile'](acoustid_track_id=None)
    id = 'e475bf79-c1ce-4441-bed7-1e33f226c0a2'
    payload = {
        'results': [{'score': 0.79}],
        'status': 'ok'
    }
    m = mocker.patch('acoustid.match', return_value=payload)
    r = tasks.set_acoustid_on_track_file(track_file_id=track_file.pk)
    track_file.refresh_from_db()

    assert track_file.acoustid_track_id is None


def test_import_batch_run(factories, mocker):
    job = factories['music.ImportJob']()
    mocked_job_run = mocker.patch(
        'funkwhale_api.music.tasks.import_job_run.delay')
    tasks.import_batch_run(import_batch_id=job.batch.pk)

    mocked_job_run.assert_called_once_with(import_job_id=job.pk)


@pytest.mark.skip('Acoustid is disabled')
def test_import_job_can_run_with_file_and_acoustid(
        artists, albums, tracks, preferences, factories, mocker):
    preferences['providers_acoustid__api_key'] = 'test'
    path = os.path.join(DATA_DIR, 'test.ogg')
    mbid = '9968a9d6-8d92-4051-8f76-674e157b6eed'
    acoustid_payload = {
        'results': [
            {'id': 'e475bf79-c1ce-4441-bed7-1e33f226c0a2',
             'recordings': [
                {
                 'duration': 268,
                 'id': mbid}],
            'score': 0.860825}],
        'status': 'ok'
    }
    mocker.patch(
        'funkwhale_api.music.utils.get_audio_file_data',
        return_value={'bitrate': 42, 'length': 43})
    mocker.patch(
        'funkwhale_api.musicbrainz.api.artists.get',
        return_value=artists['get']['adhesive_wombat'])
    mocker.patch(
        'funkwhale_api.musicbrainz.api.releases.get',
        return_value=albums['get']['marsupial'])
    mocker.patch(
        'funkwhale_api.musicbrainz.api.recordings.search',
        return_value=tracks['search']['8bitadventures'])
    mocker.patch('acoustid.match', return_value=acoustid_payload)

    job = factories['music.FileImportJob'](audio_file__path=path)
    f = job.audio_file
    tasks.import_job_run(import_job_id=job.pk)
    job.refresh_from_db()

    track_file = job.track_file

    with open(path, 'rb') as f:
        assert track_file.audio_file.read() == f.read()
    assert track_file.bitrate == 42
    assert track_file.duration == 43
    assert track_file.size == os.path.getsize(path)
    # audio file is deleted from import job once persisted to audio file
    assert not job.audio_file
    assert job.status == 'finished'
    assert job.source == 'file://'


def test_run_import_skipping_accoustid(factories, mocker):
    m = mocker.patch('funkwhale_api.music.tasks._do_import')
    path = os.path.join(DATA_DIR, 'test.ogg')
    job = factories['music.FileImportJob'](audio_file__path=path)
    tasks.import_job_run(import_job_id=job.pk, use_acoustid=False)
    m.assert_called_once_with(job, False, use_acoustid=False)


def test__do_import_skipping_accoustid(factories, mocker):
    t = factories['music.Track']()
    m = mocker.patch(
        'funkwhale_api.providers.audiofile.tasks.import_track_data_from_path',
        return_value=t)
    path = os.path.join(DATA_DIR, 'test.ogg')
    job = factories['music.FileImportJob'](
        mbid=None,
        audio_file__path=path)
    p = job.audio_file.path
    tasks._do_import(job, replace=False, use_acoustid=False)
    m.assert_called_once_with(p)


def test__do_import_skipping_accoustid_if_no_key(
        factories, mocker, preferences):
    preferences['providers_acoustid__api_key'] = ''
    t = factories['music.Track']()
    m = mocker.patch(
        'funkwhale_api.providers.audiofile.tasks.import_track_data_from_path',
        return_value=t)
    path = os.path.join(DATA_DIR, 'test.ogg')
    job = factories['music.FileImportJob'](
        mbid=None,
        audio_file__path=path)
    p = job.audio_file.path
    tasks._do_import(job, replace=False, use_acoustid=False)
    m.assert_called_once_with(p)


def test_import_job_skip_if_already_exists(
        artists, albums, tracks, factories, mocker):
    path = os.path.join(DATA_DIR, 'test.ogg')
    mbid = '9968a9d6-8d92-4051-8f76-674e157b6eed'
    track_file = factories['music.TrackFile'](track__mbid=mbid)
    mocker.patch(
        'funkwhale_api.providers.audiofile.tasks.import_track_data_from_path',
        return_value=track_file.track)

    job = factories['music.FileImportJob'](audio_file__path=path)
    f = job.audio_file
    tasks.import_job_run(import_job_id=job.pk)
    job.refresh_from_db()

    assert job.track_file is None
    # audio file is deleted from import job once persisted to audio file
    assert not job.audio_file
    assert job.status == 'skipped'


def test_import_job_can_be_errored(factories, mocker, preferences):
    path = os.path.join(DATA_DIR, 'test.ogg')
    mbid = '9968a9d6-8d92-4051-8f76-674e157b6eed'
    track_file = factories['music.TrackFile'](track__mbid=mbid)

    class MyException(Exception):
        pass

    mocker.patch(
        'funkwhale_api.music.tasks._do_import',
        side_effect=MyException())

    job = factories['music.FileImportJob'](
        audio_file__path=path, track_file=None)

    with pytest.raises(MyException):
        tasks.import_job_run(import_job_id=job.pk)

    job.refresh_from_db()

    assert job.track_file is None
    assert job.status == 'errored'


def test__do_import_calls_update_album_cover_if_no_cover(factories, mocker):
    path = os.path.join(DATA_DIR, 'test.ogg')
    album = factories['music.Album'](cover='')
    track = factories['music.Track'](album=album)

    mocker.patch(
        'funkwhale_api.providers.audiofile.tasks.import_track_data_from_path',
        return_value=track)

    mocked_update = mocker.patch(
        'funkwhale_api.music.tasks.update_album_cover')

    job = factories['music.FileImportJob'](
        audio_file__path=path, track_file=None)

    tasks.import_job_run(import_job_id=job.pk)

    mocked_update.assert_called_once_with(album, track.files.first())


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)

    mocked_get.assert_called_once_with()


def test_update_album_cover_file_data(factories, mocker):
    path = os.path.join(DATA_DIR, 'test.mp3')
    album = factories['music.Album'](cover='', mbid=None)
    tf = factories['music.TrackFile'](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)
    md = data = tf.get_metadata()
    mocked_get.assert_called_once_with(
        data={'hello': 'world'})


@pytest.mark.parametrize('ext,mimetype', [
    ('jpg', 'image/jpeg'),
    ('png', 'image/png'),
])
def test_update_album_cover_file_cover_separate_file(
        ext, mimetype, factories, mocker):
    mocker.patch('funkwhale_api.music.tasks.IMAGE_TYPES', [(ext, mimetype)])
    path = os.path.join(DATA_DIR, 'test.mp3')
    image_path = os.path.join(DATA_DIR, 'cover.{}'.format(ext))
    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)

    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)
    md = data = tf.get_metadata()
    mocked_get.assert_called_once_with(
        data={'mimetype': mimetype, 'content': image_content})