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
|
# Copyright 2021 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 os
import threading
from time import sleep, time
from quodlibet import config
from quodlibet.ext.events.qlscrobbler import QLSubmitQueue
from quodlibet.formats import AudioFile
from quodlibet.util.picklehelper import pickle_load
from senf import fsnative
from tests import run_gtk_loop, init_fake_app, destroy_fake_app
from tests.plugin import PluginTestCase
A_SONG = AudioFile(
{"~filename": fsnative("fake.mp3"), "artist": "Foo bar", "title": "The Title"}
)
class TScrobbler(PluginTestCase):
@classmethod
def setUpClass(cls):
config.init()
init_fake_app()
@classmethod
def tearDownClass(cls):
destroy_fake_app()
config.quit()
def setUp(self):
self.mod = self.modules["QLScrobbler"]
self.plugin = self.plugins["QLScrobbler"].cls()
# It's a class instance, so make sure :(
self.plugin.queue.queue.clear()
self.SCROBBLER_CACHE_FILE = self.mod.QLSubmitQueue.SCROBBLER_CACHE_FILE
try:
os.unlink(self.SCROBBLER_CACHE_FILE)
except FileNotFoundError:
pass
def tearDown(self):
del self.mod
def test_queue(self):
queue: QLSubmitQueue = self.mod.QLSubmitQueue()
thread = threading.Thread(target=queue.run, daemon=True)
thread.start()
songs = [A_SONG]
for song in songs:
queue.submit(song)
assert len(queue.queue) == 1
queue.dump_queue()
loaded = self.load_queue()
assert all(
actual["a"] == expected["artist"] and actual["t"] == expected["title"]
for actual, expected in zip(loaded, songs, strict=False)
)
def load_queue(self) -> list[dict]:
try:
with open(self.SCROBBLER_CACHE_FILE, "rb") as f:
return pickle_load(f)
except FileNotFoundError:
return []
def test_enabled_disabled(self):
self.plugin.enabled()
assert self.plugin._tid
# Disable
self.plugin.disabled()
assert not self.plugin._tid
def test_autosave(self):
self.plugin.AUTOSAVE_INTERVAL = 0.1
self.plugin.enabled()
assert not self.load_queue()
assert not self.plugin.queue.queue, "Queue not empty in test"
self.plugin.queue.submit(A_SONG)
assert len(self.plugin.queue.queue) == 1, "Song wasn't queued"
queue = self.retry_queue()
assert len(queue) == 1, "Queued song didn't get persisted"
self.plugin.disabled()
def retry_queue(self):
"""Allows GTK looks, flushing of disk buffers etc"""
queue = None
start = time()
while time() - start < 2:
run_gtk_loop()
queue = self.load_queue()
if queue:
break
sleep(0.1)
return queue
|
