added Track.get_active_cmus

symuid/__init__.py

@@ -1,6 +1,7 @@
 import datetime as dt
 import os
 import re
+import subprocess
 import typing
 
 import mutagen
@@ -171,3 +172,26 @@ class Track:
                     except NotImplementedError as exc:
                         if unsupported_cb is not None:
                             unsupported_cb(track_path, exc)
+
+    _CMUS_REMOTE_COMMAND = "cmus-remote"
+    _CMUS_STATUS_FILE_PATH_REGEX = re.compile(r"^file (.*)$", flags=re.MULTILINE)
+    _CMUS_STATUS_ENCODING = "utf-8"
+
+    @classmethod
+    def get_active_cmus(cls) -> typing.Optional["Track"]:
+        try:
+            status_proc = subprocess.run(
+                [cls._CMUS_REMOTE_COMMAND, "-Q"],
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                check=True,
+                encoding=cls._CMUS_STATUS_ENCODING,
+            )
+        except subprocess.CalledProcessError as exc:
+            if exc.returncode == 1 and "cmus is not running" in exc.stderr:
+                return None
+            raise
+        active_path_match = cls._CMUS_STATUS_FILE_PATH_REGEX.search(status_proc.stdout)
+        if not active_path_match:
+            return None
+        return cls(path=active_path_match.group(1))

tests/test_track.py

@@ -1,6 +1,10 @@
 import datetime as dt
 import os
 import re
+import shlex
+import stat
+import subprocess
+import unittest.mock
 
 import mutagen
 import pytest
@@ -219,3 +223,111 @@ def test_walk(tracks_dir_path):
         "ogg-opus-empty.opus",
         "ogg-vorbis-empty.ogg",
     }
+
+
+@pytest.mark.parametrize(
+    ("status_stdout", "expected_path"),
+    [
+        (
+            b"status playing\n"
+            b"file /some/path/possibly including/spaces.mp3\n"
+            b"duration 210\n"
+            b"position 42\n"
+            b"tag artist some artist\n"
+            b"tag album some album\n"
+            b"tag title some title\n"
+            b"tag date 2012-03-04\n"
+            b"tag bpm 0\n"
+            b"set aaa_mode all\n"
+            b"set continue true\n"
+            b"set play_library true\n"
+            b"set play_sorted false\n"
+            b"set replaygain disabled\n"
+            b"set replaygain_limit true\n"
+            b"set replaygain_preamp 0.000000\n"
+            b"set repeat false\n"
+            b"set repeat_current false\n"
+            b"set shuffle true\n"
+            b"set softvol false\n"
+            b"set vol_left 42\n"
+            b"set vol_right 42\n",
+            "/some/path/possibly including/spaces.mp3",
+        ),
+    ],
+)
+def test_get_active_cmus(tmpdir, status_stdout, expected_path):
+    status_path = tmpdir.join("status")
+    with status_path.open("wb") as status_file:
+        status_file.write(status_stdout)
+    cmd_mock_path = tmpdir.join("cmus-remote-mock")
+    with cmd_mock_path.open("w") as cmd_file:
+        cmd_file.write("#!/bin/sh\n")
+        cmd_file.write('[ "$@" = "-Q" ] || exit 1\n')
+        cmd_file.write("cat {}\n".format(shlex.quote(status_path.strpath)))
+    cmd_mock_path.chmod(stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR)
+    with unittest.mock.patch("symuid.Track.__init__", return_value=None) as init_mock:
+        with unittest.mock.patch(
+            "symuid.Track._CMUS_REMOTE_COMMAND", cmd_mock_path.strpath
+        ):
+            assert isinstance(symuid.Track.get_active_cmus(), symuid.Track)
+    init_mock.assert_called_once()
+    init_args, init_kwargs = init_mock.call_args
+    assert init_args == tuple()
+    assert init_kwargs == {"path": expected_path}
+
+
+def test_get_active_cmus_none(tmpdir):
+    status_path = tmpdir.join("status")
+    with status_path.open("wb") as status_file:
+        status_file.write(
+            b"status stopped\n"
+            b"set aaa_mode all\n"
+            b"set continue true\n"
+            b"set play_library true\n"
+            b"set play_sorted false\n"
+            b"set replaygain disabled\n"
+            b"set replaygain_limit true\n"
+            b"set replaygain_preamp 0.000000\n"
+            b"set repeat false\n"
+            b"set repeat_current false\n"
+            b"set shuffle true\n"
+            b"set softvol false\n"
+            b"set vol_left 0\n"
+            b"set vol_right 0\n"
+        )
+    cmd_mock_path = tmpdir.join("cmus-remote-mock")
+    with cmd_mock_path.open("w") as cmd_file:
+        cmd_file.write("#!/bin/sh\n")
+        cmd_file.write('[ "$@" = "-Q" ] || exit 1\n')
+        cmd_file.write("cat {}\n".format(shlex.quote(status_path.strpath)))
+    cmd_mock_path.chmod(stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR)
+    with unittest.mock.patch(
+        "symuid.Track._CMUS_REMOTE_COMMAND", cmd_mock_path.strpath
+    ):
+        assert symuid.Track.get_active_cmus() is None
+
+
+def test_get_active_cmus_not_running(tmpdir):
+    cmd_mock_path = tmpdir.join("cmus-remote-mock")
+    with cmd_mock_path.open("w") as cmd_file:
+        cmd_file.write("#!/bin/sh\n")
+        cmd_file.write("echo cmus-remote: cmus is not running >> /dev/stderr\n")
+        cmd_file.write("exit 1\n")
+    cmd_mock_path.chmod(stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR)
+    with unittest.mock.patch(
+        "symuid.Track._CMUS_REMOTE_COMMAND", cmd_mock_path.strpath
+    ):
+        assert symuid.Track.get_active_cmus() is None
+
+
+def test_get_active_cmus_unexpected_error(tmpdir):
+    cmd_mock_path = tmpdir.join("cmus-remote-mock")
+    with cmd_mock_path.open("w") as cmd_file:
+        cmd_file.write("#!/bin/sh\n")
+        cmd_file.write("exit 42\n")
+    cmd_mock_path.chmod(stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR)
+    with unittest.mock.patch(
+        "symuid.Track._CMUS_REMOTE_COMMAND", cmd_mock_path.strpath
+    ):
+        with pytest.raises(subprocess.CalledProcessError):
+            symuid.Track.get_active_cmus()