diff --git a/emmental/listenbrainz/__init__.py b/emmental/listenbrainz/__init__.py index 3d2863e..bc137ab 100644 --- a/emmental/listenbrainz/__init__.py +++ b/emmental/listenbrainz/__init__.py @@ -28,6 +28,8 @@ class ListenBrainz(GObject.GObject): def __parse_task(self, op: str, *args) -> bool: match op: + case "clear-token": + self._thread.clear_user_token() case "set-token": self._thread.set_user_token(*args) @@ -50,7 +52,9 @@ class ListenBrainz(GObject.GObject): def __notify_user_token(self, listenbrainz: GObject.GObject, param: GObject.ParamSpec) -> None: - self._queue.push("set-token", self.user_token) + match self.user_token: + case "": self._queue.push("clear-token") + case _: self._queue.push("set-token", self.user_token) self.__idle_start() def __source_stop(self, srcid: str) -> None: diff --git a/emmental/listenbrainz/thread.py b/emmental/listenbrainz/thread.py index 3ee69d8..ee44bf7 100644 --- a/emmental/listenbrainz/thread.py +++ b/emmental/listenbrainz/thread.py @@ -25,9 +25,17 @@ class Thread(thread.Thread): def do_run_task(self, task: thread.Data) -> None: """Call a specific listenbrainz operation.""" match task.op: + case "clear-token": + self._client.set_auth_token(None, check_validity=False) + self.set_result("clear-token") case "set-token": self.__set_user_token(task.token) + def clear_user_token(self) -> None: + """Schedule clearing the user token.""" + self.__print("clearing user token") + self.set_task(op="clear-token") + def get_result(self, **kwargs) -> thread.Data: """Get the result of a listenbrainz task.""" if (res := super().get_result(**kwargs)) is not None: diff --git a/tests/listenbrainz/test_listenbrainz.py b/tests/listenbrainz/test_listenbrainz.py index b7a031b..5d8367f 100644 --- a/tests/listenbrainz/test_listenbrainz.py +++ b/tests/listenbrainz/test_listenbrainz.py @@ -86,3 +86,25 @@ class TestListenBrainz(unittest.TestCase): 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) diff --git a/tests/listenbrainz/test_task.py b/tests/listenbrainz/test_task.py index 8383082..6e07f7a 100644 --- a/tests/listenbrainz/test_task.py +++ b/tests/listenbrainz/test_task.py @@ -30,3 +30,14 @@ class TestTaskQueue(unittest.TestCase): self.queue.push("set-token", "abcde") self.queue.clear("set-token") self.assertIsNone(self.queue._set_token) + + def test_push_clear_token(self): + """Test calling push() with the 'clear-token' operation.""" + self.queue.push("clear-token") + self.assertTupleEqual(self.queue._set_token, ("clear-token",)) + self.assertTupleEqual(self.queue.pop(), ("clear-token",)) + self.assertIsNone(self.queue._set_token) + + self.queue.push("clear-token") + self.queue.clear("clear-token") + self.assertIsNone(self.queue._set_token) diff --git a/tests/listenbrainz/test_thread.py b/tests/listenbrainz/test_thread.py index c59dfd6..294d2cc 100644 --- a/tests/listenbrainz/test_thread.py +++ b/tests/listenbrainz/test_thread.py @@ -24,6 +24,21 @@ class TestThread(unittest.TestCase): 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,