diff --git a/CHANGES b/CHANGES index e7d822880..64b36ec79 100644 --- a/CHANGES +++ b/CHANGES @@ -14,6 +14,13 @@ $ pip install --user --upgrade --pre libtmux +### New features + +#### Detect if server active (#448) + +- `Server.is_alive()` +- `Server.raise_if_dead()` + ### Internal - Remove unused `sphinx-click` development dependency diff --git a/src/libtmux/server.py b/src/libtmux/server.py index 7f5d0affa..cfc0bda44 100644 --- a/src/libtmux/server.py +++ b/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)) + + """ + 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: """ diff --git a/tests/test_server.py b/tests/test_server.py index 782150086..d7830fd0b 100644 --- a/tests/test_server.py +++ b/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()