Server(fix[new_session]): Harden error handling and TMUX env restoration

src/libtmux/server.py

@@ -539,46 +539,51 @@ def new_session(
         if env:
             del os.environ["TMUX"]
 
-        tmux_args: tuple[str | int, ...] = (
-            "-P",
-            "-F#{session_id}",  # output
-        )
+        try:
+            tmux_args: tuple[str | int, ...] = (
+                "-P",
+                "-F#{session_id}",  # output
+            )
 
-        if session_name is not None:
-            tmux_args += (f"-s{session_name}",)
+            if session_name is not None:
+                tmux_args += (f"-s{session_name}",)
 
-        if not attach:
-            tmux_args += ("-d",)
+            if not attach:
+                tmux_args += ("-d",)
 
-        if start_directory:
-            start_directory = pathlib.Path(start_directory).expanduser()
-            tmux_args += ("-c", str(start_directory))
+            if start_directory:
+                start_directory = pathlib.Path(start_directory).expanduser()
+                tmux_args += ("-c", str(start_directory))
 
-        if window_name:
-            tmux_args += ("-n", window_name)
+            if window_name:
+                tmux_args += ("-n", window_name)
 
-        if x is not None:
-            tmux_args += ("-x", x)
+            if x is not None:
+                tmux_args += ("-x", x)
 
-        if y is not None:
-            tmux_args += ("-y", y)
+            if y is not None:
+                tmux_args += ("-y", y)
 
-        if environment:
-            for k, v in environment.items():
-                tmux_args += (f"-e{k}={v}",)
+            if environment:
+                for k, v in environment.items():
+                    tmux_args += (f"-e{k}={v}",)
 
-        if window_command:
-            tmux_args += (window_command,)
+            if window_command:
+                tmux_args += (window_command,)
 
-        proc = self.cmd("new-session", *tmux_args)
+            proc = self.cmd("new-session", *tmux_args)
 
-        if proc.stderr:
-            raise exc.LibTmuxException(proc.stderr)
+            if proc.stderr:
+                raise exc.LibTmuxException(proc.stderr)
 
-        session_stdout = proc.stdout[0]
+            if not proc.stdout:
+                msg = "new-session produced no output"
+                raise exc.LibTmuxException(msg)
 
-        if env:
-            os.environ["TMUX"] = env
+            session_stdout = proc.stdout[0]
+        finally:
+            if env:
+                os.environ["TMUX"] = env
 
         session_formatters = dict(
             zip(

tests/test_server.py

@@ -11,6 +11,7 @@
 
 import pytest
 
+from libtmux import exc
 from libtmux.server import Server
 
 if t.TYPE_CHECKING:
@@ -104,6 +105,95 @@ def test_new_session(server: Server) -> None:
     assert server.has_session("test_new_session")
 
 
+def test_new_session_empty_stdout(
+    server: Server,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Server.new_session raises LibTmuxException on empty stdout.
+
+    Uses monkeypatch to simulate empty stdout, which cannot be triggered
+    through normal tmux operations.
+    """
+    original_cmd = server.cmd
+
+    def mock_cmd(cmd: str, *args: t.Any, **kwargs: t.Any) -> t.Any:
+        result = original_cmd(cmd, *args, **kwargs)
+        if cmd == "new-session":
+            result.stdout = []
+        return result
+
+    monkeypatch.setattr(server, "cmd", mock_cmd)
+
+    with pytest.raises(exc.LibTmuxException, match="new-session produced no output"):
+        server.new_session(session_name="test_empty_stdout")
+
+    monkeypatch.undo()
+    if server.has_session("test_empty_stdout"):
+        server.kill_session("test_empty_stdout")
+
+
+def test_new_session_restores_tmux_env_on_error(
+    server: Server,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Server.new_session restores TMUX env var even when an exception is raised.
+
+    Uses monkeypatch to simulate empty stdout, which cannot be triggered
+    through normal tmux operations.
+    """
+    original_cmd = server.cmd
+    sentinel = "/tmp/libtmux-test-fake-socket,12345,0"
+
+    monkeypatch.setenv("TMUX", sentinel)
+
+    def mock_cmd(cmd: str, *args: t.Any, **kwargs: t.Any) -> t.Any:
+        result = original_cmd(cmd, *args, **kwargs)
+        if cmd == "new-session":
+            result.stdout = []
+        return result
+
+    monkeypatch.setattr(server, "cmd", mock_cmd)
+
+    with pytest.raises(exc.LibTmuxException, match="new-session produced no output"):
+        server.new_session(session_name="test_env_restore")
+
+    assert os.environ.get("TMUX") == sentinel
+
+    monkeypatch.undo()
+    if server.has_session("test_env_restore"):
+        server.kill_session("test_env_restore")
+
+
+def test_new_session_restores_tmux_env_on_setup_error(
+    server: Server,
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: pathlib.Path,
+) -> None:
+    """Server.new_session restores TMUX env var when setup code before cmd() raises.
+
+    Uses monkeypatch to make pathlib.Path.expanduser raise, which cannot be
+    triggered through normal tmux operations. This verifies the try/finally
+    covers the arg-building phase, not just the cmd() call.
+    """
+    sentinel = "/tmp/libtmux-test-fake-socket,99999,0"
+
+    monkeypatch.setenv("TMUX", sentinel)
+
+    def broken_expanduser(self: pathlib.Path) -> t.Any:
+        msg = "injected setup error"
+        raise RuntimeError(msg)
+
+    monkeypatch.setattr(pathlib.Path, "expanduser", broken_expanduser)
+
+    with pytest.raises(RuntimeError, match="injected setup error"):
+        server.new_session(
+            session_name="test_setup_error",
+            start_directory=tmp_path,
+        )
+
+    assert os.environ.get("TMUX") == sentinel
+
+
 def test_new_session_no_name(server: Server) -> None:
     """Server.new_session works with no name."""
     first_session = server.new_session()