logout command to remove cached credentials

Author: kingosticks

src/mopidy_spotify/__init__.py

@@ -50,3 +50,15 @@ def setup(self, registry):
         from mopidy_spotify.backend import SpotifyBackend
 
         registry.add("backend", SpotifyBackend)
+
+    def get_command(self):
+        from .commands import SpotifyCommand
+
+        return SpotifyCommand()
+
+    @classmethod
+    def get_credentials_dir(cls, config):
+        data_dir = cls.get_data_dir(config)
+        credentials_dir = data_dir / "credentials-cache"
+        credentials_dir.mkdir(mode=0o700, exist_ok=True)
+        return credentials_dir

src/mopidy_spotify/backend.py

@@ -37,13 +37,9 @@ class SpotifyPlaybackProvider(backend.PlaybackProvider):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self._cache_location = Extension().get_cache_dir(self.backend._config)
-        self._data_location = Extension().get_data_dir(self.backend._config)
+        self._credentials_dir = Extension().get_credentials_dir(self.backend._config)
         self._config = self.backend._config["spotify"]
 
-        self._credentials_dir = self._data_location / "credentials-cache"
-        if not self._credentials_dir.exists():
-            self._credentials_dir.mkdir(mode=0o700)
-
     def on_source_setup(self, source):
         source.set_property("bitrate", str(self._config["bitrate"]))
         source.set_property("cache-credentials", self._credentials_dir)

src/mopidy_spotify/commands.py

@@ -0,0 +1,38 @@
+import logging
+import os
+from pathlib import Path
+
+from mopidy import commands
+
+from mopidy_spotify import Extension
+
+logger = logging.getLogger(__name__)
+
+
+class SpotifyCommand(commands.Command):
+    def __init__(self):
+        super().__init__()
+        self.add_child("logout", LogoutCommand())
+
+
+class LogoutCommand(commands.Command):
+    help = "Logout from Spotify account."
+
+    def run(self, _args, config):
+        credentials_dir = Extension().get_credentials_dir(config)
+        try:
+            for root, dirs, files in os.walk(credentials_dir, topdown=False):
+                root_path = Path(root)
+                for name in files:
+                    file_path = root_path / name
+                    file_path.unlink()
+                    logger.debug(f"Removed file {file_path}")
+                for name in dirs:
+                    dir_path = root_path / name
+                    dir_path.rmdir()
+                    logger.debug(f"Removed directory {dir_path}")
+            credentials_dir.rmdir()
+        except Exception as error:
+            logger.warning(f"Failed to logout from Spotify: {error}")
+        else:
+            logger.info("Logout from Spotify complete")

tests/test_commands.py

@@ -0,0 +1,16 @@
+from unittest.mock import sentinel
+
+from mopidy_spotify import Extension
+from mopidy_spotify.commands import LogoutCommand
+
+
+def test_logout_command(tmp_path):
+    config = {"core": {"data_dir": tmp_path}}
+    credentials_dir = Extension().get_credentials_dir(config)
+    (credentials_dir / "foo").mkdir()
+    (credentials_dir / "bar").touch()
+
+    cmd = LogoutCommand()
+    cmd.run(sentinel.args, config)
+
+    assert not credentials_dir.is_dir()

tests/test_extension.py

@@ -40,3 +40,16 @@ def test_setup():
     ext.setup(registry)
 
     registry.add.assert_called_with("backend", backend_lib.SpotifyBackend)
+
+
+def test_get_credentials_dir(tmp_path):
+    config = {"core": {"data_dir": tmp_path}}
+
+    ext = Extension()
+    result = ext.get_credentials_dir(config)
+    assert result == tmp_path / "spotify" / "credentials-cache"
+    assert result.is_dir()
+    assert result.stat().st_mode == 0o40700
+
+    result2 = ext.get_credentials_dir(config)  # check exists_ok
+    assert result == result2