feat(pytest): Add TempServer for temporary tmux server management

tony · tony · commit a7aaa3ef723a · 2025-02-15T07:19:09.000-06:00
Add TempServer function to pytest_plugin.py that creates a temporary tmux server
that cleans up after itself. This is similar to the server pytest fixture but can
be used outside of pytest contexts.

- Move and rename TempTmuxServer to TempServer in pytest_plugin.py
- Add comprehensive tests for TempServer functionality:
  - Basic server creation and cleanup
  - Custom socket name support
  - Config file loading
  - Multiple server isolation
  - Proper cleanup verification
- Use fin() for cleanup instead of __del__
- Remove old implementation from test.py

This provides a more flexible way to create temporary tmux servers for testing
and development purposes, while ensuring proper cleanup.
diff --git a/src/libtmux/pytest_plugin.py b/src/libtmux/pytest_plugin.py
@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import contextlib
+import functools
 import getpass
 import logging
 import os
@@ -256,3 +257,50 @@ def session(
     assert TEST_SESSION_NAME != "tmuxp"
 
     return session
+
+
+@pytest.fixture
+def TempServer(
+    request: pytest.FixtureRequest,
+) -> t.Callable[[], t.Callable[..., Server]]:
+    """Create a temporary tmux server that cleans up after itself.
+
+    This is similar to the server pytest fixture, but can be used outside of pytest.
+    The server will be killed when the test completes.
+
+    Returns
+    -------
+    Callable[[], Callable[..., Server]]
+        A factory function that returns a partial'd Server with a unique socket_name
+
+    Examples
+    --------
+    >>> Server = request.getfixturevalue('TempServer')()  # Get unique partial
+    >>> server = Server()  # Create server instance
+    >>> server.new_session()
+    Session($... ...)
+    >>> server.is_alive()
+    True
+    >>> # Each call to TempServer() gets a new unique partial
+    >>> Server2 = request.getfixturevalue('TempServer')()
+    >>> server2 = Server2()
+    >>> server2.socket_name != server.socket_name
+    True
+    """
+    created_sockets: list[str] = []
+
+    def get_server() -> t.Callable[..., Server]:
+        socket_name = f"libtmux_test{next(namer)}"
+        created_sockets.append(socket_name)
+        return functools.partial(Server, socket_name=socket_name)
+
+    def fin() -> None:
+        """Kill all servers created with these sockets."""
+        for socket_name in created_sockets:
+            server = Server(socket_name=socket_name)
+            if server.is_alive():
+                server.kill()
+
+    request.addfinalizer(fin)
+
+    return get_server
diff --git a/tests/test_pytest_plugin.py b/tests/test_pytest_plugin.py
@@ -3,9 +3,12 @@
 from __future__ import annotations
 
 import textwrap
+import time
 import typing as t
 
 if t.TYPE_CHECKING:
+    import pathlib
+
     import pytest
 
 
@@ -71,3 +74,81 @@ def test_repo_git_remote_checkout(
     # Test
     result = pytester.runpytest(str(first_test_filename))
     result.assert_outcomes(passed=1)
+
+
+def test_temp_server(TempServer: t.Any) -> None:
+    """Test TempServer creates and cleans up server."""
+    Server = TempServer()  # Get partial'd Server
+    server = Server()  # Create server instance
+    assert server.is_alive() is False  # Server not started yet
+
+    session = server.new_session()
+    assert server.is_alive() is True
+    assert len(server.sessions) == 1
+    assert session.session_name is not None
+
+    # Test socket name is unique
+    assert server.socket_name is not None
+    assert server.socket_name.startswith("libtmux_test")
+
+
+def test_temp_server_with_config(TempServer: t.Any, tmp_path: pathlib.Path) -> None:
+    """Test TempServer with custom config."""
+    config_file = tmp_path / "tmux.conf"
+    config_file.write_text("set -g status off", encoding="utf-8")
+
+    Server = TempServer()  # Get partial'd Server
+    server = Server(config_file=str(config_file))  # Create server with config
+    session = server.new_session()
+
+    # Verify config was loaded by checking status
+    assert session.cmd("show-options", "-g", "status").stdout[0] == "status off"
+
+
+def test_temp_server_cleanup(TempServer: t.Any) -> None:
+    """Test TempServer properly cleans up after itself."""
+    Server = TempServer()  # Get partial'd Server
+    server = Server()  # Create server instance
+    socket_name = server.socket_name
+    assert socket_name is not None
+
+    # Create multiple sessions
+    server.new_session(session_name="test1")
+    server.new_session(session_name="test2")
+    assert len(server.sessions) == 2
+
+    # Verify server is alive
+    assert server.is_alive() is True
+
+    # Delete server and verify cleanup
+    server.kill()
+    time.sleep(0.1)  # Give time for cleanup
+
+    # Create new server to verify old one was cleaned up
+    Server2 = TempServer()  # Get new partial'd Server
+    new_server = Server2()  # Create new server instance
+    assert new_server.is_alive() is False  # Server not started yet
+    new_server.new_session()  # This should work if old server was cleaned up
+    assert new_server.is_alive() is True
+
+
+def test_temp_server_multiple(TempServer: t.Any) -> None:
+    """Test multiple TempServer instances can coexist."""
+    Server1 = TempServer()  # Get first partial'd Server
+    Server2 = TempServer()  # Get second partial'd Server
+
+    server1 = Server1()  # Create first server instance
+    server2 = Server2()  # Create second server instance
+
+    # Each server should have a unique socket
+    assert server1.socket_name != server2.socket_name
+
+    # Create sessions in each server
+    server1.new_session(session_name="test1")
+    server2.new_session(session_name="test2")
+
+    # Verify sessions are in correct servers
+    assert any(s.session_name == "test1" for s in server1.sessions)
+    assert any(s.session_name == "test2" for s in server2.sessions)
+    assert not any(s.session_name == "test1" for s in server2.sessions)
+    assert not any(s.session_name == "test2" for s in server1.sessions)
