244 lines
12 KiB
Python
244 lines
12 KiB
Python
# Copyright 2024 (c) Anna Schumaker.
|
|
"""Tests our custom ListenBrainz GObject."""
|
|
import datetime
|
|
import emmental.listenbrainz
|
|
import io
|
|
import pathlib
|
|
import tests.util
|
|
import unittest
|
|
from gi.repository import GObject
|
|
from gi.repository import GLib
|
|
|
|
|
|
@unittest.mock.patch("sys.stdout", new_callable=io.StringIO)
|
|
@unittest.mock.patch("gi.repository.GLib.source_remove")
|
|
@unittest.mock.patch("gi.repository.GLib.idle_add", return_value=42)
|
|
class TestListenBrainz(tests.util.TestCase):
|
|
"""ListenBrainz GObject test case."""
|
|
|
|
def setUp(self):
|
|
"""Set up common variables."""
|
|
super().setUp()
|
|
self.listenbrainz = emmental.listenbrainz.ListenBrainz(self.sql)
|
|
self.library = self.sql.libraries.create(pathlib.Path("/a/b"))
|
|
self.album = self.sql.albums.create("Test Album", "Test Artist",
|
|
release="1988-06",
|
|
mbid="mbid-release")
|
|
self.medium = self.sql.media.create(self.album, "", number=1)
|
|
self.year = self.sql.years.create(1988)
|
|
self.track = self.sql.tracks.create(self.library,
|
|
pathlib.Path("/a/b/c.ogg"),
|
|
self.medium, self.year,
|
|
title="Track 1", number=1,
|
|
artist="Track Artist", length=10)
|
|
|
|
@unittest.mock.patch("gi.repository.GLib.source_remove")
|
|
def tearDown(self, mock_source_remove: unittest.mock.Mock):
|
|
"""Clean up."""
|
|
self.listenbrainz.stop()
|
|
|
|
def test_init(self, mock_idle_add: unittest.mock.Mock,
|
|
mock_source_remove: unittest.mock.Mock,
|
|
mock_stdout: io.StringIO):
|
|
"""Test that the ListenBrainz GObject was set up properly."""
|
|
self.assertIsInstance(self.listenbrainz, GObject.GObject)
|
|
self.assertIsInstance(self.listenbrainz._queue,
|
|
emmental.listenbrainz.task.Queue)
|
|
self.assertIsInstance(self.listenbrainz._thread,
|
|
emmental.listenbrainz.thread.Thread)
|
|
self.assertIsNone(self.listenbrainz._idle_id)
|
|
self.assertEqual(self.listenbrainz.sql, self.sql)
|
|
|
|
def test_stop(self, mock_idle_add: unittest.mock.Mock,
|
|
mock_source_remove: unittest.mock.Mock,
|
|
mock_stdout: io.StringIO):
|
|
"""Test stopping the thread during shutdown."""
|
|
self.listenbrainz._idle_id = 12345
|
|
|
|
self.listenbrainz.stop()
|
|
self.assertFalse(self.listenbrainz._thread.is_alive())
|
|
self.assertIsNone(self.listenbrainz._idle_id)
|
|
mock_source_remove.assert_called_with(12345)
|
|
|
|
def test_set_user_token(self, mock_idle_add: unittest.mock.Mock,
|
|
mock_source_remove: unittest.mock.Mock,
|
|
mock_stdout: io.StringIO):
|
|
"""Test setting the user-token property."""
|
|
self.assertEqual(self.listenbrainz.user_token, "")
|
|
self.assertTrue(self.listenbrainz.valid_token)
|
|
|
|
idle_work = self.listenbrainz._ListenBrainz__idle_work
|
|
with unittest.mock.patch.object(self.listenbrainz._thread,
|
|
"set_user_token") as mock_set_token:
|
|
self.listenbrainz.user_token = "abc"
|
|
self.assertEqual(self.listenbrainz._queue._set_token,
|
|
("set-token", "abc"))
|
|
self.assertEqual(self.listenbrainz._idle_id, 42)
|
|
mock_idle_add.assert_called_with(idle_work)
|
|
|
|
self.assertEqual(idle_work(), GLib.SOURCE_CONTINUE)
|
|
mock_set_token.assert_called_with("abc")
|
|
|
|
mock_idle_add.reset_mock()
|
|
self.listenbrainz.user_token = "abcde"
|
|
self.assertEqual(self.listenbrainz._queue._set_token,
|
|
("set-token", "abcde"))
|
|
self.assertEqual(self.listenbrainz._idle_id, 42)
|
|
mock_idle_add.assert_not_called()
|
|
|
|
self.listenbrainz._thread.set_result(op="set-token", token="abc",
|
|
valid=True)
|
|
self.assertEqual(idle_work(), GLib.SOURCE_CONTINUE)
|
|
mock_set_token.assert_called_with("abcde")
|
|
|
|
self.listenbrainz._thread.ready.clear()
|
|
self.assertEqual(idle_work(), GLib.SOURCE_CONTINUE)
|
|
|
|
for valid in [False, True]:
|
|
with self.subTest(valid=valid):
|
|
self.listenbrainz._thread.set_result(op="set-token",
|
|
token="abcde",
|
|
valid=valid)
|
|
self.assertEqual(idle_work(), GLib.SOURCE_REMOVE)
|
|
self.assertEqual(self.listenbrainz.valid_token, valid)
|
|
self.assertIsNone(self.listenbrainz._idle_id)
|
|
|
|
def test_clear_user_token(self, mock_idle_add: unittest.mock.Mock,
|
|
mock_source_remove: unittest.mock.Mock,
|
|
mock_stdout: io.StringIO):
|
|
"""Test clearing the user-token property."""
|
|
idle_work = self.listenbrainz._ListenBrainz__idle_work
|
|
with unittest.mock.patch.object(self.listenbrainz._thread,
|
|
"clear_user_token") as mock_clear:
|
|
self.listenbrainz.valid_token = False
|
|
self.listenbrainz.user_token = ""
|
|
self.assertEqual(self.listenbrainz._queue._set_token,
|
|
("clear-token",))
|
|
self.assertEqual(self.listenbrainz._idle_id, 42)
|
|
mock_idle_add.assert_called_with(idle_work)
|
|
|
|
self.assertEqual(idle_work(), GLib.SOURCE_CONTINUE)
|
|
mock_clear.assert_called()
|
|
|
|
self.listenbrainz._thread.set_result(op="clear-token", valid=True)
|
|
self.assertEqual(idle_work(), GLib.SOURCE_REMOVE)
|
|
self.assertTrue(self.listenbrainz.valid_token)
|
|
self.assertIsNone(self.listenbrainz._idle_id)
|
|
|
|
def test_submit_now_playing(self, mock_idle_add: unittest.mock.Mock,
|
|
mock_source_remove: unittest.mock.Mock,
|
|
mock_stdout: io.StringIO):
|
|
"""Test setting the now-playing property."""
|
|
self.assertIsNone(self.listenbrainz.now_playing)
|
|
|
|
self.listenbrainz.user_token = "abcde"
|
|
self.listenbrainz.valid_token = True
|
|
self.listenbrainz._queue.pop()
|
|
self.listenbrainz._ListenBrainz__source_stop("_idle_id")
|
|
|
|
self.listenbrainz.now_playing = self.track
|
|
self.assertTupleEqual(self.listenbrainz._queue._now_playing,
|
|
("now-playing", self.track))
|
|
self.assertEqual(self.listenbrainz._idle_id, 42)
|
|
|
|
idle_work = self.listenbrainz._ListenBrainz__idle_work
|
|
with unittest.mock.patch.object(self.listenbrainz._thread,
|
|
"submit_now_playing") as mock_playing:
|
|
self.assertEqual(idle_work(), GLib.SOURCE_CONTINUE)
|
|
mock_playing.assert_called()
|
|
self.assertIsInstance(mock_playing.call_args.args[0],
|
|
emmental.listenbrainz.listen.Listen)
|
|
|
|
for valid in [True, False]:
|
|
with self.subTest(valid=valid):
|
|
self.listenbrainz._thread.set_result(op="now-playing",
|
|
valid=valid)
|
|
self.assertEqual(idle_work(), GLib.SOURCE_REMOVE)
|
|
self.assertEqual(self.listenbrainz.valid_token, valid)
|
|
|
|
def test_submit_now_playing_later(self, mock_idle_add: unittest.mock.Mock,
|
|
mock_source_remove: unittest.mock.Mock,
|
|
mock_stdout: io.StringIO):
|
|
"""Test the now-playing property when ListenBrainz is disconnected."""
|
|
self.assertIsNone(self.listenbrainz.now_playing)
|
|
|
|
self.listenbrainz.now_playing = self.track
|
|
self.assertTupleEqual(self.listenbrainz._queue._now_playing,
|
|
("now-playing", self.track))
|
|
self.assertIsNone(self.listenbrainz._idle_id)
|
|
|
|
self.listenbrainz.user_token = "abcde"
|
|
self.listenbrainz.valid_token = False
|
|
self.listenbrainz._queue.pop()
|
|
self.listenbrainz._ListenBrainz__source_stop("_idle_id")
|
|
self.listenbrainz.now_playing = self.track
|
|
self.assertTupleEqual(self.listenbrainz._queue._now_playing,
|
|
("now-playing", self.track))
|
|
self.assertIsNone(self.listenbrainz._idle_id)
|
|
|
|
self.listenbrainz.valid_token = True
|
|
self.listenbrainz._queue._now_playing = "abcde"
|
|
self.listenbrainz.now_playing = None
|
|
self.assertIsNone(self.listenbrainz._queue._now_playing)
|
|
self.assertIsNone(self.listenbrainz._idle_id)
|
|
|
|
def test_submit_listens(self, mock_idle_add: unittest.mock.Mock,
|
|
mock_source_remove: unittest.mock.Mock,
|
|
mock_stdout: io.StringIO):
|
|
"""Test submitting recently listened tracks."""
|
|
ts1 = datetime.datetime.utcnow()
|
|
ts2 = datetime.datetime.utcnow()
|
|
idle_work = self.listenbrainz._ListenBrainz__idle_work
|
|
listens = [emmental.listenbrainz.listen.Listen(self.track, listenid=1,
|
|
listened_at=ts1),
|
|
emmental.listenbrainz.listen.Listen(self.track, listenid=2,
|
|
listened_at=ts2)]
|
|
|
|
self.listenbrainz.user_token = "abcde"
|
|
self.listenbrainz.valid_token = True
|
|
self.listenbrainz._queue.pop()
|
|
self.listenbrainz._ListenBrainz__source_stop("_idle_id")
|
|
|
|
self.listenbrainz.submit_listens("ignored", "args")
|
|
self.assertIsNotNone(self.listenbrainz._idle_id)
|
|
|
|
with unittest.mock.patch.object(self.sql.tracks,
|
|
"get_n_listens") as mock_get_listens:
|
|
mock_get_listens.return_value = [(1, self.track, ts1),
|
|
(2, self.track, ts2)]
|
|
|
|
with unittest.mock.patch.object(self.listenbrainz._thread,
|
|
"submit_listens") as mock_submit:
|
|
self.assertEqual(idle_work(), GLib.SOURCE_CONTINUE)
|
|
mock_get_listens.assert_called_with(50)
|
|
mock_submit.assert_called()
|
|
|
|
with unittest.mock.patch.object(self.sql.tracks,
|
|
"delete_listens") as mock_delete:
|
|
for valid in [True, False]:
|
|
mock_delete.reset_mock()
|
|
with self.subTest(valid=valid):
|
|
self.listenbrainz._thread.set_result(op="submit-listens",
|
|
listens=listens,
|
|
valid=valid)
|
|
self.assertEqual(idle_work(), GLib.SOURCE_REMOVE)
|
|
self.assertEqual(self.listenbrainz.valid_token, valid)
|
|
if valid:
|
|
mock_delete.assert_called_with([1, 2])
|
|
else:
|
|
mock_delete.assert_not_called()
|
|
|
|
def test_submit_listens_later(self, mock_idle_add: unittest.mock.Mock,
|
|
mock_source_remove: unittest.mock.Mock,
|
|
mock_stdout: io.StringIO):
|
|
"""Test submitting listens when ListenBrainz is disconnected."""
|
|
self.listenbrainz.submit_listens("ignored", "args")
|
|
self.assertIsNone(self.listenbrainz._idle_id)
|
|
|
|
self.listenbrainz.user_token = "abcde"
|
|
self.listenbrainz.valid_token = False
|
|
self.listenbrainz._queue.pop()
|
|
self.listenbrainz._idle_id = None
|
|
self.listenbrainz.submit_listens("ignored", "args")
|
|
self.assertIsNone(self.listenbrainz._idle_id)
|