190 lines
9.5 KiB
Python
190 lines
9.5 KiB
Python
# Copyright 2024 (c) Anna Schumaker.
|
|
"""Tests our ListenBrainz client thread."""
|
|
import emmental.listenbrainz.thread
|
|
import io
|
|
import liblistenbrainz
|
|
import requests
|
|
import unittest
|
|
|
|
|
|
@unittest.mock.patch("sys.stdout", new_callable=io.StringIO)
|
|
class TestThread(unittest.TestCase):
|
|
"""ListenBrainz Thread test case."""
|
|
|
|
def setUp(self):
|
|
"""Set up common variables."""
|
|
self.thread = emmental.listenbrainz.thread.Thread()
|
|
|
|
def tearDown(self):
|
|
"""Clean up."""
|
|
self.thread.stop()
|
|
|
|
def test_init(self, mock_stdout: io.StringIO):
|
|
"""Test that the ListenBrainz thread was initialized properly."""
|
|
self.assertIsInstance(self.thread, emmental.thread.Thread)
|
|
self.assertIsInstance(self.thread._client,
|
|
liblistenbrainz.client.ListenBrainz)
|
|
|
|
def test_clear_user_token(self, mock_stdout: io.StringIO):
|
|
"""Test clearing the user token."""
|
|
with unittest.mock.patch.object(self.thread._client,
|
|
"set_auth_token") as mock_set_auth:
|
|
self.thread.clear_user_token()
|
|
self.assertFalse(self.thread.ready.is_set())
|
|
self.assertEqual(self.thread._task, {"op": "clear-token"})
|
|
self.assertEqual(mock_stdout.getvalue(),
|
|
"listenbrainz: clearing user token\n")
|
|
|
|
self.thread.ready.wait()
|
|
mock_set_auth.assert_called_with(None, check_validity=False)
|
|
self.assertEqual(self.thread.get_result(),
|
|
{"op": "clear-token", "valid": True,
|
|
"offline": False})
|
|
|
|
def test_set_user_token(self, mock_stdout: io.StringIO):
|
|
"""Test setting the user auth token."""
|
|
with unittest.mock.patch.object(self.thread._client,
|
|
"set_auth_token") as mock_set_auth:
|
|
self.thread.set_user_token("abcde")
|
|
self.assertFalse(self.thread.ready.is_set())
|
|
self.assertEqual(self.thread._task,
|
|
{"op": "set-token", "token": "abcde"})
|
|
self.assertEqual(mock_stdout.getvalue(),
|
|
"listenbrainz: setting user token\n")
|
|
|
|
self.thread.ready.wait()
|
|
mock_set_auth.assert_called_with("abcde")
|
|
self.assertEqual(self.thread.get_result(),
|
|
{"op": "set-token", "token": "abcde",
|
|
"valid": True, "offline": False})
|
|
|
|
def test_set_user_token_exceptions(self, mock_stdout: io.StringIO):
|
|
"""Test exception handling when setting the user auth token."""
|
|
with unittest.mock.patch.object(self.thread._client,
|
|
"set_auth_token") as mock_set_auth:
|
|
mock_set_auth.side_effect = \
|
|
liblistenbrainz.errors.InvalidAuthTokenException()
|
|
self.thread.set_user_token("abcde")
|
|
self.thread.ready.wait()
|
|
self.assertEqual(self.thread.get_result(),
|
|
{"op": "set-token", "token": "abcde",
|
|
"valid": False, "offline": False})
|
|
self.assertEqual(mock_stdout.getvalue(),
|
|
"listenbrainz: setting user token\n" +
|
|
"listenbrainz: user token is invalid\n")
|
|
|
|
mock_set_auth.side_effect = requests.exceptions.ConnectionError()
|
|
self.thread.set_user_token("abcde")
|
|
self.thread.ready.wait()
|
|
self.assertEqual(self.thread.get_result(),
|
|
{"op": "set-token", "token": "abcde",
|
|
"valid": True, "offline": True})
|
|
self.assertRegex(mock_stdout.getvalue(), "listenbrainz: offline")
|
|
|
|
def test_submit_now_playing(self, mock_stdout: io.StringIO):
|
|
"""Test submitting the now playing track."""
|
|
listen = liblistenbrainz.Listen("Track Name", "Artist Name")
|
|
with unittest.mock.patch.object(self.thread._client,
|
|
"submit_playing_now") as mock_submit:
|
|
self.thread.submit_now_playing(listen)
|
|
self.assertFalse(self.thread.ready.is_set())
|
|
self.assertEqual(self.thread._task, {"op": "now-playing",
|
|
"listen": listen})
|
|
self.assertEqual(mock_stdout.getvalue(),
|
|
"listenbrainz: now playing 'Track Name' " +
|
|
"by 'Artist Name'\n")
|
|
|
|
self.thread.ready.wait()
|
|
mock_submit.assert_called_with(listen)
|
|
self.assertEqual(self.thread.get_result(),
|
|
{"op": "now-playing", "valid": True,
|
|
"offline": False})
|
|
|
|
def test_submit_now_playing_exceptions(self, mock_stdout: io.StringIO):
|
|
"""Test exception handling when submitting the now playing track."""
|
|
listen = liblistenbrainz.Listen("Track Name", "Artist Name")
|
|
with unittest.mock.patch.object(self.thread._client,
|
|
"submit_playing_now") as mock_submit:
|
|
mock_submit.side_effect = \
|
|
liblistenbrainz.errors.ListenBrainzAPIException(401)
|
|
self.thread.submit_now_playing(listen)
|
|
self.thread.ready.wait()
|
|
self.assertEqual(self.thread.get_result(),
|
|
{"op": "now-playing", "valid": False,
|
|
"offline": False})
|
|
self.assertEqual(mock_stdout.getvalue(),
|
|
"listenbrainz: now playing 'Track Name' " +
|
|
"by 'Artist Name'\n" +
|
|
"listenbrainz: user token is invalid\n")
|
|
|
|
mock_submit.side_effect = requests.exceptions.ConnectionError()
|
|
self.thread.submit_now_playing(listen)
|
|
self.thread.ready.wait()
|
|
self.assertEqual(self.thread.get_result(),
|
|
{"op": "now-playing", "valid": True,
|
|
"offline": True})
|
|
self.assertRegex(mock_stdout.getvalue(), "listenbrainz: offline")
|
|
|
|
def test_submit_single_listen(self, mock_stdout: io.StringIO):
|
|
"""Test submitting a single listen."""
|
|
listens = [liblistenbrainz.Listen("Track Name", "Artist Name")]
|
|
with unittest.mock.patch.object(self.thread._client,
|
|
"submit_single_listen") as mock_submit:
|
|
self.thread.submit_listens(listens)
|
|
self.assertFalse(self.thread.ready.is_set())
|
|
self.assertEqual(self.thread._task, {"op": "submit-listens",
|
|
"listens": listens})
|
|
self.assertEqual(mock_stdout.getvalue(),
|
|
"listenbrainz: submitting 1 listen\n")
|
|
|
|
self.thread.ready.wait()
|
|
mock_submit.assert_called_with(listens[0])
|
|
self.assertEqual(self.thread.get_result(),
|
|
{"op": "submit-listens", "listens": listens,
|
|
"valid": True, "offline": False})
|
|
|
|
def test_submit_multiple_listens(self, mock_stdout: io.StringIO):
|
|
"""Test submitting multiple listens."""
|
|
listens = [liblistenbrainz.Listen("Track 1", "Artist"),
|
|
liblistenbrainz.Listen("Track 2", "Artist"),
|
|
liblistenbrainz.Listen("Track 3", "Artist")]
|
|
with unittest.mock.patch.object(self.thread._client,
|
|
"submit_multiple_listens") \
|
|
as mock_submit:
|
|
self.thread.submit_listens(listens)
|
|
self.assertFalse(self.thread.ready.is_set())
|
|
self.assertEqual(self.thread._task, {"op": "submit-listens",
|
|
"listens": listens})
|
|
self.assertEqual(mock_stdout.getvalue(),
|
|
"listenbrainz: submitting 3 listens\n")
|
|
|
|
self.thread.ready.wait()
|
|
mock_submit.assert_called_with(listens)
|
|
self.assertEqual(self.thread.get_result(),
|
|
{"op": "submit-listens", "listens": listens,
|
|
"valid": True, "offline": False})
|
|
|
|
def test_submit_listens_exceptions(self, mock_stdout: io.StringIO):
|
|
"""Test exception handling when submitting listens."""
|
|
listens = [liblistenbrainz.Listen("Track Name", "Artist Name")]
|
|
with unittest.mock.patch.object(self.thread._client,
|
|
"submit_single_listen") as mock_submit:
|
|
mock_submit.side_effect = \
|
|
liblistenbrainz.errors.ListenBrainzAPIException(401)
|
|
self.thread.submit_listens(listens)
|
|
self.thread.ready.wait()
|
|
self.assertEqual(self.thread.get_result(),
|
|
{"op": "submit-listens", "listens": listens,
|
|
"valid": False, "offline": False})
|
|
self.assertEqual(mock_stdout.getvalue(),
|
|
"listenbrainz: submitting 1 listen\n" +
|
|
"listenbrainz: user token is invalid\n")
|
|
|
|
mock_submit.side_effect = requests.exceptions.ConnectionError()
|
|
self.thread.submit_listens(listens)
|
|
self.thread.ready.wait()
|
|
self.assertEqual(self.thread.get_result(),
|
|
{"op": "submit-listens", "listens": listens,
|
|
"valid": True, "offline": True})
|
|
self.assertRegex(mock_stdout.getvalue(), "listenbrainz: offline")
|