1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
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
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
|
# Copyright 2012 Christoph Reiter
# 2023 Nick Boultbee
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
import gi
from tests import run_gtk_loop
gi.require_version('Soup', '3.0')
from gi.repository import Gtk
from dataclasses import dataclass, field
from time import time, sleep
from typing import Any, Optional, List
import pytest as pytest
from quodlibet.formats import AudioFile
from quodlibet.util.cover.http import ApiCoverSourcePlugin
from senf import fsnative
from tests.plugin import PluginTestCase, plugins
AN_ARTIST = "The Beatles"
AN_ALBUM = "Let It Be"
A_SONG = AudioFile({"album": AN_ALBUM, "artist": AN_ARTIST})
AN_MBID = "82a4adf2-008b-3236-bb7a-bd93d7ed9677"
def delay_rerun(self, *args):
# Try to recover from any network blips
sleep(10)
return True
class TCovers(PluginTestCase):
def setUp(self) -> None:
self.song = A_SONG
self.blank_song = AudioFile()
def test_cover_path_lastfm(self):
plugin_cls = self.plugins["lastfm-cover"].cls
assert isinstance(plugin_cls(self.song).cover_path, fsnative)
assert isinstance(plugin_cls(self.blank_song).cover_path, fsnative)
def test_cover_path_musicbrainz(self):
plugin_cls = self.plugins["musicbrainz-cover"].cls
assert isinstance(plugin_cls(self.song).cover_path, fsnative)
assert isinstance(plugin_cls(self.blank_song).cover_path, fsnative)
def test_cover_path_discogs(self):
plugin_cls = self.plugins["discogs-cover"].cls
assert isinstance(plugin_cls(self.song).cover_path, fsnative)
assert isinstance(plugin_cls(self.blank_song).cover_path, fsnative)
@dataclass
class Results:
covers: List[Any] = field(default_factory=list)
success: Optional[bool] = None
@pytest.mark.network
@pytest.mark.flaky(max_runs=3, min_passes=1, rerun_filter=delay_rerun)
@pytest.mark.parametrize("plugin_class_name",
["lastfm-cover", "discogs-cover", "musicbrainz-cover"])
def test_live_cover_download(plugin_class_name):
results = Results()
# Just in case overhanging events
run_gtk_loop()
def search_complete(source, data, results):
results.covers = data
def good(source, data, results):
results.success = True
header = data.read(4)
data.close()
assert header.startswith(b"\x89PNG") or header.startswith(b"\xFF\xD8")
def bad(source, error, results):
# For debugging
results.covers = error
results.success = False
plugin_cls = plugins[plugin_class_name].cls
song = A_SONG
if "musicbrainz" in plugin_class_name:
song["musicbrainz_albumid"] = AN_MBID
plugin: ApiCoverSourcePlugin = plugin_cls(song)
sig = plugin.connect("search-complete", search_complete, results)
sig2 = plugin.connect("fetch-success", good, results)
sig3 = plugin.connect("fetch-failure", bad, results)
try:
start = time()
if "musicbrainz" in plugin_class_name:
# Isn't called by fetch_cover()
plugin.search()
plugin.fetch_cover()
while time() - start < 10 and results.success is None:
Gtk.main_iteration_do(False)
assert results.success is not None, "No signal triggered"
assert results.success, f"Didn't succeed: {results.covers}"
covers = results.covers
assert covers, "Didn't get search results"
first = covers[0]
assert first["cover"].startswith("http")
assert "dimensions" in first
if "album" in first:
# Only lastfm populates this currently
assert first["album"] == AN_ALBUM, f"Downloaded wrong cover: {covers}"
finally:
plugin.disconnect(sig)
plugin.disconnect(sig2)
plugin.disconnect(sig3)
|
