TestServer: Server, but partial'd to run on a test socket (#565)

#### New `Server` params

`Server` now accepts 2 new optional params, `socket_name_factory` and `on_init` callbacks (#565):

- `socket_name_factory`: Callable that generates unique socket names for new servers
- `on_init`: Callback that runs after server initialization
- Useful for creating multiple servers with unique names and tracking server instances
- Socket name factory is tried after socket_name, maintaining backward compatibility

#### New test fixture: `TestServer`

Add `TestServer` pytest fixture for creating temporary tmux servers (#565):

- Creates servers with unique socket names that clean up after themselves
- Useful for testing interactions between multiple tmux servers
- Includes comprehensive test coverage and documentation
- Available in doctest namespace

Author: tony

CHANGES

@@ -11,9 +11,27 @@ $ pip install --user --upgrade --pre libtmux
 
 ## libtmux 0.43.x (Yet to be released)
 
+<!-- To maintainers and contributors: Please add notes for the forthcoming version below -->
+
 - _Future release notes will be placed here_
 
-<!-- To maintainers and contributors: Please add notes for the forthcoming version above -->
+### Features
+
+Server now accepts 2 new optional params, `socket_name_factory` and `on_init` callbacks (#565):
+
+- `socket_name_factory`: Callable that generates unique socket names for new servers
+- `on_init`: Callback that runs after server initialization
+- Useful for creating multiple servers with unique names and tracking server instances
+- Socket name factory is tried after socket_name, maintaining backward compatibility
+
+#### New test fixture: `TestServer`
+
+Add `TestServer` pytest fixture for creating temporary tmux servers (#565):
+
+- Creates servers with unique socket names that clean up after themselves
+- Useful for testing interactions between multiple tmux servers
+- Includes comprehensive test coverage and documentation
+- Available in doctest namespace
 
 ## libtmux 0.42.1 (2024-02-15)
 
@@ -88,6 +106,7 @@ _Maintenance only, no bug fixes or new features_
   ```sh
   ruff check --select ALL . --fix --unsafe-fixes --preview --show-fixes; ruff format .
   ```
+
 - Tests: Stability fixes for legacy `test_select_pane` test (#552)
 
 ## libtmux 0.39.0 (2024-11-26)

conftest.py

@@ -41,6 +41,7 @@ def add_doctest_fixtures(
         doctest_namespace["Window"] = Window
         doctest_namespace["Pane"] = Pane
         doctest_namespace["server"] = request.getfixturevalue("server")
+        doctest_namespace["Server"] = request.getfixturevalue("TestServer")
         session: Session = request.getfixturevalue("session")
         doctest_namespace["session"] = session
         doctest_namespace["window"] = session.active_window

docs/pytest-plugin/index.md

@@ -93,6 +93,34 @@ def test_something(session):
 
 The above will assure the libtmux session launches with `-x 800 -y 600`.
 
+(temp_server)=
+
+### Creating temporary servers
+
+If you need multiple independent tmux servers in your tests, the {func}`TestServer fixture <libtmux.pytest_plugin.TestServer>` provides a factory that creates servers with unique socket names. Each server is automatically cleaned up when the test completes.
+
+```python
+def test_something(TestServer):
+    Server = TestServer()  # Get unique partial'd Server
+    server = Server()      # Create server instance
+    
+    session = server.new_session()
+    assert server.is_alive()
+```
+
+You can also use it with custom configurations, similar to the {ref}`server fixture <Setting a tmux configuration>`:
+
+```python
+def test_with_config(TestServer, tmp_path):
+    config_file = tmp_path / "tmux.conf"
+    config_file.write_text("set -g status off")
+    
+    Server = TestServer()
+    server = Server(config_file=str(config_file))
+```
+
+This is particularly useful when testing interactions between multiple tmux servers or when you need to verify behavior across server restarts.
+
 (set_home)=
 
 ### Setting a temporary home directory

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,58 @@ def session(
     assert TEST_SESSION_NAME != "tmuxp"
 
     return session
+
+
+@pytest.fixture
+def TestServer(
+    request: pytest.FixtureRequest,
+) -> type[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
+    -------
+    type[Server]
+        A factory function that returns a Server with a unique socket_name
+
+    Examples
+    --------
+    >>> server = Server()  # Create server instance
+    >>> server.new_session()
+    Session($... ...)
+    >>> server.is_alive()
+    True
+    >>> # Each call creates a new server with unique socket
+    >>> server2 = Server()
+    >>> server2.socket_name != server.socket_name
+    True
+    """
+    created_sockets: list[str] = []
+
+    def on_init(server: Server) -> None:
+        """Track created servers for cleanup."""
+        created_sockets.append(server.socket_name or "default")
+
+    def socket_name_factory() -> str:
+        """Generate unique socket names."""
+        return f"libtmux_test{next(namer)}"
+
+    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 t.cast(
+        "type[Server]",
+        functools.partial(
+            Server,
+            on_init=on_init,
+            socket_name_factory=socket_name_factory,
+        ),
+    )

src/libtmux/server.py

@@ -59,6 +59,8 @@ class Server(EnvironmentMixin):
     socket_path : str, optional
     config_file : str, optional
     colors : str, optional
+    on_init : callable, optional
+    socket_name_factory : callable, optional
 
     Examples
     --------
@@ -110,6 +112,8 @@ def __init__(
         socket_path: str | pathlib.Path | None = None,
         config_file: str | None = None,
         colors: int | None = None,
+        on_init: t.Callable[[Server], None] | None = None,
+        socket_name_factory: t.Callable[[], str] | None = None,
         **kwargs: t.Any,
     ) -> None:
         EnvironmentMixin.__init__(self, "-g")
@@ -120,6 +124,8 @@ def __init__(
             self.socket_path = socket_path
         elif socket_name is not None:
             self.socket_name = socket_name
+        elif socket_name_factory is not None:
+            self.socket_name = socket_name_factory()
 
         tmux_tmpdir = pathlib.Path(os.getenv("TMUX_TMPDIR", "/tmp"))
         socket_name = self.socket_name or "default"
@@ -137,6 +143,9 @@ def __init__(
         if colors:
             self.colors = colors
 
+        if on_init is not None:
+            on_init(self)
+
     def is_alive(self) -> bool:
         """Return True if tmux server alive.
 

tests/test_pytest_plugin.py

@@ -3,11 +3,16 @@
 from __future__ import annotations
 
 import textwrap
+import time
 import typing as t
 
 if t.TYPE_CHECKING:
+    import pathlib
+
     import pytest
 
+    from libtmux.server import Server
+
 
 def test_plugin(
     pytester: pytest.Pytester,
@@ -71,3 +76,83 @@ def test_repo_git_remote_checkout(
     # Test
     result = pytester.runpytest(str(first_test_filename))
     result.assert_outcomes(passed=1)
+
+
+def test_test_server(TestServer: t.Callable[..., Server]) -> None:
+    """Test TestServer creates and cleans up server."""
+    server = TestServer()
+    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")
+
+    # Each call creates a new server with unique socket
+    server2 = TestServer()
+    assert server2.socket_name is not None
+    assert server2.socket_name.startswith("libtmux_test")
+    assert server2.socket_name != server.socket_name
+
+
+def test_test_server_with_config(
+    TestServer: t.Callable[..., Server],
+    tmp_path: pathlib.Path,
+) -> None:
+    """Test TestServer with config file."""
+    config_file = tmp_path / "tmux.conf"
+    config_file.write_text("set -g status off", encoding="utf-8")
+
+    server = TestServer(config_file=str(config_file))
+    session = server.new_session()
+
+    # Verify config was loaded
+    assert session.cmd("show-options", "-g", "status").stdout[0] == "status off"
+
+
+def test_test_server_cleanup(TestServer: t.Callable[..., Server]) -> None:
+    """Test TestServer properly cleans up after itself."""
+    server = TestServer()
+    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
+    new_server = TestServer()
+    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_test_server_multiple(TestServer: t.Callable[..., Server]) -> None:
+    """Test multiple TestServer instances can coexist."""
+    server1 = TestServer()
+    server2 = TestServer()
+
+    # 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)

tests/test_server.py

@@ -228,3 +228,71 @@ def test_raise_if_dead_does_not_raise_if_alive(server: Server) -> None:
     """Verify new_session() does not raise if tmux server is alive."""
     server.new_session()
     server.raise_if_dead()
+
+
+def test_on_init(server: Server) -> None:
+    """Verify on_init callback is called during Server initialization."""
+    called_with: list[Server] = []
+
+    def on_init(server: Server) -> None:
+        called_with.append(server)
+
+    myserver = Server(socket_name="test_on_init", on_init=on_init)
+    try:
+        assert len(called_with) == 1
+        assert called_with[0] is myserver
+    finally:
+        if myserver.is_alive():
+            myserver.kill()
+
+
+def test_socket_name_factory(server: Server) -> None:
+    """Verify socket_name_factory generates socket names."""
+    socket_names: list[str] = []
+
+    def socket_name_factory() -> str:
+        name = f"test_socket_{len(socket_names)}"
+        socket_names.append(name)
+        return name
+
+    myserver = Server(socket_name_factory=socket_name_factory)
+    try:
+        assert myserver.socket_name == "test_socket_0"
+        assert socket_names == ["test_socket_0"]
+
+        # Creating another server should use factory again
+        myserver2 = Server(socket_name_factory=socket_name_factory)
+        try:
+            assert myserver2.socket_name == "test_socket_1"
+            assert socket_names == ["test_socket_0", "test_socket_1"]
+        finally:
+            if myserver2.is_alive():
+                myserver2.kill()
+    finally:
+        if myserver.is_alive():
+            myserver.kill()
+        if myserver2.is_alive():
+            myserver2.kill()
+
+
+def test_socket_name_precedence(server: Server) -> None:
+    """Verify socket_name takes precedence over socket_name_factory."""
+
+    def socket_name_factory() -> str:
+        return "from_factory"
+
+    myserver = Server(
+        socket_name="explicit_name",
+        socket_name_factory=socket_name_factory,
+    )
+    myserver2 = Server(socket_name_factory=socket_name_factory)
+    try:
+        assert myserver.socket_name == "explicit_name"
+
+        # Without socket_name, factory is used
+        assert myserver2.socket_name == "from_factory"
+    finally:
+        if myserver.is_alive():
+            myserver.kill()
+        if myserver2.is_alive():
+            myserver2.kill()