refactor(Server): (attached_)sessions return list if server off

Author: tony

src/libtmux/server.py

@@ -6,6 +6,8 @@
 """
 import logging
 import os
+import shutil
+import subprocess
 import typing as t
 
 from libtmux.common import tmux_cmd
@@ -115,6 +117,42 @@ def __init__(
         if colors:
             self.colors = colors
 
+    def is_alive(self) -> bool:
+        """If server alive or not.
+
+        >>> tmux = Server(socket_name="no_exist")
+        >>> assert not tmux.is_alive()
+        """
+        try:
+            res = self.cmd("list-sessions")
+            return res.returncode == 0
+        except Exception:
+            return False
+
+    def raise_if_dead(self) -> None:
+        """Raise if server not connected.
+
+        >>> tmux = Server(socket_name="no_exist")
+        >>> try:
+        ...     tmux.raise_if_dead()
+        ... except Exception as e:
+        ...     print(type(e))
+        <class 'subprocess.CalledProcessError'>
+        """
+        tmux_bin = shutil.which("tmux")
+        if tmux_bin is None:
+            raise exc.TmuxCommandNotFound()
+
+        cmd_args: t.List[str] = ["list-sessions"]
+        if self.socket_name:
+            cmd_args.insert(0, f"-L{self.socket_name}")
+        if self.socket_path:
+            cmd_args.insert(0, f"-S{self.socket_path}")
+        if self.config_file:
+            cmd_args.insert(0, f"-f{self.config_file}")
+
+        subprocess.check_call([tmux_bin] + cmd_args)
+
     def cmd(self, *args: t.Any, **kwargs: t.Any) -> tmux_cmd:
         """
         Execute tmux command and return output.
@@ -207,7 +245,10 @@ def list_sessions(self) -> t.List[Session]:
     @property
     def sessions(self) -> t.List[Session]:
         """Property / alias to return :meth:`~.list_sessions`."""
-        return self.list_sessions()
+        try:
+            return self.list_sessions()
+        except Exception:
+            return []
 
     #: Alias :attr:`sessions` for :class:`~libtmux.common.TmuxRelationalObject`
     children = sessions  # type: ignore
@@ -348,7 +389,7 @@ def _update_panes(self) -> "Server":
         return self
 
     @property
-    def attached_sessions(self) -> t.Optional[t.List[Session]]:
+    def attached_sessions(self) -> t.List[Session]:
         """
         Return active :class:`Session` objects.
 
@@ -357,19 +398,22 @@ def attached_sessions(self) -> t.Optional[t.List[Session]]:
         list of :class:`Session`
         """
 
-        sessions = self._sessions
-        attached_sessions = list()
+        try:
+            sessions = self._sessions
+            attached_sessions = list()
 
-        for session in sessions:
-            attached = session.get("session_attached")
-            # for now session_active is a unicode
-            if attached != "0":
-                logger.debug(f"session {session.get('name')} attached")
-                attached_sessions.append(session)
-            else:
-                continue
+            for session in sessions:
+                attached = session.get("session_attached")
+                # for now session_active is a unicode
+                if attached != "0":
+                    logger.debug(f"session {session.get('name')} attached")
+                    attached_sessions.append(session)
+                else:
+                    continue
 
-        return [Session(server=self, **s) for s in attached_sessions] or None
+            return [Session(server=self, **s) for s in attached_sessions] or []
+        except Exception:
+            return []
 
     def has_session(self, target_session: str, exact: bool = True) -> bool:
         """

tests/test_server.py

@@ -1,6 +1,8 @@
 """Test for libtmux Server object."""
 import logging
 
+import pytest
+
 from libtmux.common import has_gte_version
 from libtmux.server import Server
 from libtmux.session import Session
@@ -123,3 +125,34 @@ def test_new_session_shell(server: Server) -> None:
         assert pane_start_command.replace('"', "") == cmd
     else:
         assert pane_start_command == cmd
+
+
+def test_no_server_sessions() -> None:
+    server = Server(socket_name="test_attached_session_no_server")
+    assert server.sessions == []
+
+
+def test_no_server_attached_sessions() -> None:
+    server = Server(socket_name="test_no_server_attached_sessions")
+    assert server.attached_sessions == []
+
+
+def test_no_server_is_alive() -> None:
+    dead_server = Server(socket_name="test_no_server_is_alive")
+    assert not dead_server.is_alive()
+
+
+def test_with_server_is_alive(server: Server) -> None:
+    server.new_session()
+    assert server.is_alive()
+
+
+def test_no_server_raise_if_dead() -> None:
+    dead_server = Server(socket_name="test_attached_session_no_server")
+    with pytest.raises(Exception):
+        dead_server.raise_if_dead()
+
+
+def test_with_server_raise_if_dead(server: Server) -> None:
+    server.new_session()
+    server.raise_if_dead()