73 lines
3.2 KiB
Python
73 lines
3.2 KiB
Python
# Copyright 2024 (c) Anna Schumaker.
|
|
"""Tests our ListenBrainz client thread."""
|
|
import emmental.listenbrainz.thread
|
|
import io
|
|
import liblistenbrainz
|
|
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})
|
|
|
|
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})
|
|
|
|
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})
|
|
self.assertEqual(mock_stdout.getvalue(),
|
|
"listenbrainz: setting user token\n" +
|
|
"listenbrainz: user token is invalid\n")
|