diff --git a/README.md b/README.md index 805808b3e..0cc5d91bf 100644 --- a/README.md +++ b/README.md @@ -246,6 +246,46 @@ Window(@1 1:..., Session($1 ...)) Session($1 ...) ``` +# Async support + +libtmux provides async versions of key methods for use in async applications: + +```python +import asyncio +from libtmux import Server + +async def main(): + server = Server() + + # Create session asynchronously + session = await server.anew_session( + session_name="async_session", + start_directory="~/" + ) + + # Create windows concurrently + windows = await asyncio.gather( + session.anew_window(window_name="editor"), + session.anew_window(window_name="terminal"), + session.anew_window(window_name="logs"), + ) + + # Check session exists + exists = await server.ahas_session("async_session") + print(f"Session exists: {exists}") # True + +asyncio.run(main()) +``` + +Available async methods (using 'a' prefix convention): +- `Server.ahas_session()` - Check if session exists +- `Server.anew_session()` - Create new session +- `Session.anew_window()` - Create new window +- `Session.arename_session()` - Rename session +- `Window.akill()` - Kill window + +See the [async API documentation](https://libtmux.git-pull.com/api/async.html) for details. + # Python support Unsupported / no security releases or bug fixes: diff --git a/docs/_templates/sidebar/projects.html b/docs/_templates/sidebar/projects.html index 7b46e0bce..97420c1ad 100644 --- a/docs/_templates/sidebar/projects.html +++ b/docs/_templates/sidebar/projects.html @@ -49,7 +49,7 @@

web - social-embed + social-embed

diff --git a/docs/api/async.md b/docs/api/async.md new file mode 100644 index 000000000..4337bfae2 --- /dev/null +++ b/docs/api/async.md @@ -0,0 +1,486 @@ +(async)= + +# Async Operations + +libtmux provides async versions of key operations for use in async applications. +These methods use the 'a' prefix naming convention (e.g., `anew_session`, `ahas_session`) +and leverage `await self.acmd()` for non-blocking communication with tmux. + +## Overview + +Async methods enable: +- **Non-blocking operations**: Don't block the event loop while waiting for tmux +- **Concurrent execution**: Run multiple tmux operations in parallel with `asyncio.gather()` +- **Better performance**: Significant speedup when performing multiple operations +- **Async integration**: Seamless integration with async frameworks (FastAPI, aiohttp, etc.) + +## When to Use Async Methods + +**Use async methods when:** +- Your application is built with asyncio +- You need to perform multiple tmux operations concurrently +- You're integrating with async frameworks +- You want to avoid blocking operations in an event loop + +**Use sync methods when:** +- You're writing simple scripts +- You don't need concurrency +- You prefer simpler, more straightforward code +- You're not in an async context + +## Available Async Methods + +### Server Async Methods + +```{eval-rst} +.. currentmodule:: libtmux + +.. autosummary:: + :toctree: _autosummary + + Server.ahas_session + Server.anew_session +``` + +#### Server.ahas_session() + +Check if a session exists asynchronously. + +```python +exists = await server.ahas_session("my_session") +``` + +See {meth}`Server.ahas_session` for full documentation. + +#### Server.anew_session() + +Create a new session asynchronously. + +```python +session = await server.anew_session( + session_name="my_project", + start_directory="~/code/myproject", + environment={"NODE_ENV": "development"} +) +``` + +See {meth}`Server.anew_session` for full documentation. + +### Session Async Methods + +```{eval-rst} +.. autosummary:: + :toctree: _autosummary + + Session.anew_window + Session.arename_session +``` + +#### Session.anew_window() + +Create a new window asynchronously. + +```python +window = await session.anew_window( + window_name="editor", + start_directory="/tmp" +) +``` + +See {meth}`Session.anew_window` for full documentation. + +#### Session.arename_session() + +Rename a session asynchronously. + +```python +session = await session.arename_session("new_name") +``` + +See {meth}`Session.arename_session` for full documentation. + +### Window Async Methods + +```{eval-rst} +.. autosummary:: + :toctree: _autosummary + + Window.akill +``` + +#### Window.akill() + +Kill a window asynchronously. + +```python +await window.akill() + +# Or kill all windows except this one +await window.akill(all_except=True) +``` + +See {meth}`Window.akill` for full documentation. + +### Pane Async Methods + +```{eval-rst} +.. autosummary:: + :toctree: _autosummary + + Pane.asend_keys + Pane.acapture_pane + Pane.asplit +``` + +#### Pane.asend_keys() + +Send keys to pane asynchronously, enabling non-blocking command execution. + +```python +# Basic usage +await pane.asend_keys('echo "Hello"', enter=True) + +# Send without executing +await pane.asend_keys('ls -la', enter=False) + +# Literal mode (special chars as text) +await pane.asend_keys('C-c', literal=True) + +# Suppress shell history +await pane.asend_keys('secret_command', suppress_history=True) + +# Concurrent execution across multiple panes +await asyncio.gather( + pane1.asend_keys('echo "pane1"'), + pane2.asend_keys('echo "pane2"'), + pane3.asend_keys('echo "pane3"'), +) +``` + +See {meth}`Pane.asend_keys` for full documentation. + +#### Pane.acapture_pane() + +Capture pane output asynchronously, enabling concurrent monitoring. + +```python +# Capture visible pane +output = await pane.acapture_pane() + +# Capture with history (last 10 lines) +output = await pane.acapture_pane(start=-10) + +# Capture specific range +output = await pane.acapture_pane(start=0, end=5) + +# Capture complete scrollback +output = await pane.acapture_pane(start="-", end="-") + +# Concurrent capture from multiple panes +outputs = await asyncio.gather( + pane1.acapture_pane(), + pane2.acapture_pane(), + pane3.acapture_pane(), +) +``` + +See {meth}`Pane.acapture_pane` for full documentation. + +#### Pane.asplit() + +Split pane asynchronously, enabling rapid layout creation. + +```python +# Default split (below) +new_pane = await pane.asplit() + +# Vertical split (right) +from libtmux.pane import PaneDirection +new_pane = await pane.asplit(direction=PaneDirection.Right) + +# With custom directory and size +new_pane = await pane.asplit( + start_directory='/tmp', + size="30%" +) + +# With shell command (auto-closes) +new_pane = await pane.asplit(shell='echo "done"') + +# Concurrent splits for rapid layout +new_panes = await asyncio.gather( + pane.asplit(direction=PaneDirection.Right), + pane.asplit(direction=PaneDirection.Below), +) +``` + +See {meth}`Pane.asplit` for full documentation. + +## Usage Patterns + +### Basic Async Pattern + +```python +import asyncio +from libtmux import Server + +async def main(): + server = Server() + + # Create session + session = await server.anew_session(session_name="my_session") + + # Create window + window = await session.anew_window(window_name="my_window") + + # Check session exists + exists = await server.ahas_session("my_session") + print(f"Session exists: {exists}") + +asyncio.run(main()) +``` + +### Concurrent Operations + +One of the key benefits of async methods is the ability to run multiple operations concurrently: + +```python +import asyncio + +async def setup_project_workspace(): + server = Server() + + # Create multiple sessions concurrently + frontend, backend, database = await asyncio.gather( + server.anew_session( + session_name="frontend", + start_directory="~/project/frontend" + ), + server.anew_session( + session_name="backend", + start_directory="~/project/backend" + ), + server.anew_session( + session_name="database", + start_directory="~/project/database" + ), + ) + + # Set up windows in each session concurrently + await asyncio.gather( + frontend.anew_window(window_name="editor"), + frontend.anew_window(window_name="server"), + backend.anew_window(window_name="api"), + backend.anew_window(window_name="tests"), + database.anew_window(window_name="console"), + ) + + return frontend, backend, database +``` + +### Integration with Async Frameworks + +#### FastAPI Example + +```python +from fastapi import FastAPI +from libtmux import Server + +app = FastAPI() +server = Server() + +@app.post("/sessions/") +async def create_session(name: str, directory: str = None): + """Create a tmux session via API.""" + session = await server.anew_session( + session_name=name, + start_directory=directory + ) + + return { + "session_id": session.session_id, + "session_name": session.session_name, + } + +@app.get("/sessions/{name}") +async def check_session(name: str): + """Check if a session exists.""" + exists = await server.ahas_session(name) + return {"exists": exists} +``` + +#### aiohttp Example + +```python +from aiohttp import web +from libtmux import Server + +async def handle_create_session(request): + server = Server() + data = await request.json() + + session = await server.anew_session( + session_name=data["name"], + start_directory=data.get("directory") + ) + + return web.json_response({ + "session_id": session.session_id, + "session_name": session.session_name, + }) + +app = web.Application() +app.router.add_post('/sessions', handle_create_session) +``` + +### Error Handling + +```python +from libtmux import exc + +async def safe_session_creation(server, name): + """Create session with proper error handling.""" + try: + # Check if session already exists + if await server.ahas_session(name): + print(f"Session {name} already exists") + return None + + # Create new session + session = await server.anew_session(session_name=name) + return session + + except exc.BadSessionName as e: + print(f"Invalid session name: {e}") + return None + + except exc.LibTmuxException as e: + print(f"tmux error: {e}") + return None +``` + +### Cleanup Patterns + +```python +async def managed_session(): + """Use context manager pattern for cleanup.""" + server = Server() + session = None + + try: + # Create resources + session = await server.anew_session(session_name="temp_session") + window = await session.anew_window(window_name="work") + + # Do work... + yield session + + finally: + # Clean up resources + if session and server.has_session(session.session_name): + await session.kill_session() +``` + +## Performance Characteristics + +### Concurrent vs Sequential + +**Sequential (slower):** + +```python +# Creates sessions one at a time +session1 = await server.anew_session("session1") +session2 = await server.anew_session("session2") +session3 = await server.anew_session("session3") +# Takes ~3x the time of one operation +``` + +**Concurrent (faster):** + +```python +# Creates all sessions in parallel +sessions = await asyncio.gather( + server.anew_session("session1"), + server.anew_session("session2"), + server.anew_session("session3"), +) +# Takes ~1x the time of one operation +``` + +### Benchmarks + +Typical performance improvements with async concurrent operations: + +- **3 sessions created concurrently**: ~2-3x faster than sequential +- **10 windows created concurrently**: ~5-8x faster than sequential +- **Checking 20 sessions concurrently**: ~10-15x faster than sequential + +Actual performance depends on system resources and tmux response time. + +## Comparison with Sync API + +| Feature | Sync API | Async API | +|---------|----------|-----------| +| Method naming | `new_session()` | `anew_session()` | +| Execution | Blocking | Non-blocking | +| Concurrency | Sequential only | True concurrency | +| Use case | Scripts, simple apps | Async apps, high performance | +| Complexity | Simpler | More complex | +| Event loop | Not required | Required (asyncio) | + +## Implementation Details + +### The 'a' Prefix Convention + +Async methods use the 'a' prefix naming convention: +- `has_session()` → `ahas_session()` +- `new_session()` → `anew_session()` +- `new_window()` → `anew_window()` +- `rename_session()` → `arename_session()` +- `kill()` → `akill()` + +This makes it clear which methods are async and prevents naming conflicts. + +### Under the Hood + +Async methods use `await self.acmd()` instead of `self.cmd()`: + +```python +# Sync version +def has_session(self, target_session: str) -> bool: + proc = self.cmd("has-session", target=target_session) + return bool(not proc.returncode) + +# Async version +async def ahas_session(self, target_session: str) -> bool: + proc = await self.acmd("has-session", target=target_session) + return bool(not proc.returncode) +``` + +The `acmd()` method uses `AsyncTmuxCmd` which leverages `asyncio.create_subprocess_exec()` +for non-blocking subprocess execution. + +## Roadmap + +This is the **foundation of async support** in libtmux (v0.48.0). The current async API provides: + +✅ Core session management (create, check, rename) +✅ Window management (create, kill) +✅ Full parameter support matching sync methods +✅ Concurrent operation support + +**Future enhancements may include:** +- Additional async wrapper methods for panes +- Async context managers +- Async iterators for tmux objects +- Performance optimizations + +Async support is actively being expanded. Contributions welcome! + +## See Also + +- {ref}`servers` - Server class documentation +- {ref}`sessions` - Session class documentation +- {ref}`windows` - Window class documentation +- {ref}`quickstart` - Basic libtmux usage +- [Python asyncio documentation](https://docs.python.org/3/library/asyncio.html) diff --git a/docs/api/index.md b/docs/api/index.md index 99d614fee..c0e0f35d3 100644 --- a/docs/api/index.md +++ b/docs/api/index.md @@ -11,6 +11,7 @@ servers sessions windows panes +async constants common exceptions diff --git a/docs/api/servers.md b/docs/api/servers.md index deb7fbf89..25b784c61 100644 --- a/docs/api/servers.md +++ b/docs/api/servers.md @@ -9,6 +9,17 @@ tmux initializes a server automatically on first running (e.g. executing `tmux`) +## Async Methods + +Server provides async versions of key methods for use in async applications: + +- {meth}`~Server.ahas_session` - Check if session exists asynchronously +- {meth}`~Server.anew_session` - Create new session asynchronously + +See {ref}`async` for comprehensive async documentation. + +## API Reference + ```{eval-rst} .. autoclass:: libtmux.Server :members: diff --git a/docs/api/sessions.md b/docs/api/sessions.md index 008d64df9..aac881c3a 100644 --- a/docs/api/sessions.md +++ b/docs/api/sessions.md @@ -6,6 +6,17 @@ - Contain {ref}`Windows` (which contain {ref}`Panes`) - Identified by `$`, e.g. `$313` +## Async Methods + +Session provides async versions of key methods for use in async applications: + +- {meth}`~Session.anew_window` - Create new window asynchronously +- {meth}`~Session.arename_session` - Rename session asynchronously + +See {ref}`async` for comprehensive async documentation. + +## API Reference + ```{eval-rst} .. autoclass:: libtmux.Session :members: diff --git a/docs/api/windows.md b/docs/api/windows.md index 728554b0d..f19365f80 100644 --- a/docs/api/windows.md +++ b/docs/api/windows.md @@ -6,6 +6,16 @@ - Contain {ref}`Panes` - Identified by `@`, e.g. `@313` +## Async Methods + +Window provides async versions of key methods for use in async applications: + +- {meth}`~Window.akill` - Kill window asynchronously + +See {ref}`async` for comprehensive async documentation. + +## API Reference + ```{module} libtmux ``` diff --git a/docs/quickstart.md b/docs/quickstart.md index 90edfcbfc..ae2c7cd5c 100644 --- a/docs/quickstart.md +++ b/docs/quickstart.md @@ -441,6 +441,86 @@ automatically sent, the leading space character prevents adding it to the user's shell history. Omitting `enter=false` means the default behavior (sending the command) is done, without needing to use `pane.enter()` after. +## Async Support + +For async applications, libtmux provides async versions of key methods using the 'a' prefix naming convention. + +### Basic Async Usage + +```python +import asyncio +from libtmux import Server + +async def main(): + server = Server() + + # Create session asynchronously + session = await server.anew_session(session_name="async_demo") + + # Create window asynchronously + window = await session.anew_window(window_name="async_window") + + # Check session exists + exists = await server.ahas_session("async_demo") + print(f"Session exists: {exists}") # True + +asyncio.run(main()) +``` + +### Concurrent Operations + +One of the key benefits of async methods is concurrent execution: + +```python +import asyncio + +async def setup_workspace(): + server = Server() + + # Create multiple sessions concurrently + frontend, backend, database = await asyncio.gather( + server.anew_session( + session_name="frontend", + start_directory="~/project/frontend" + ), + server.anew_session( + session_name="backend", + start_directory="~/project/backend" + ), + server.anew_session( + session_name="database", + start_directory="~/project/database" + ), + ) + + # Set up windows in each session concurrently + await asyncio.gather( + frontend.anew_window(window_name="editor"), + frontend.anew_window(window_name="server"), + backend.anew_window(window_name="api"), + backend.anew_window(window_name="tests"), + ) + + return frontend, backend, database +``` + +### Available Async Methods + +- {meth}`Server.ahas_session` - Check if session exists +- {meth}`Server.anew_session` - Create new session +- {meth}`Session.anew_window` - Create new window +- {meth}`Session.arename_session` - Rename session +- {meth}`Window.akill` - Kill window + +### When to Use Async + +Use async methods when: +- Your application uses asyncio +- You need to perform multiple tmux operations concurrently +- You're integrating with async frameworks (FastAPI, aiohttp, etc.) + +For more details, see {ref}`async`. + ## Final notes These objects created use tmux's internal usage of ID's to make servers, diff --git a/pyproject.toml b/pyproject.toml index 2deddc21c..e8fa3a0a6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -67,6 +67,7 @@ dev = [ "typing-extensions; python_version < '3.11'", "gp-libs", "pytest", + "pytest-asyncio", "pytest-rerunfailures", "pytest-mock", "pytest-watcher", @@ -97,6 +98,7 @@ testing = [ "typing-extensions; python_version < '3.11'", "gp-libs", "pytest", + "pytest-asyncio", "pytest-rerunfailures", "pytest-mock", "pytest-watcher", diff --git a/src/libtmux/common.py b/src/libtmux/common.py index ac9b9b7f1..fec2e4f7d 100644 --- a/src/libtmux/common.py +++ b/src/libtmux/common.py @@ -7,6 +7,7 @@ from __future__ import annotations +import asyncio import logging import re import shutil @@ -267,6 +268,149 @@ def __init__(self, *args: t.Any) -> None: ) +class AsyncTmuxCmd: + """ + An asyncio-compatible class for running any tmux command via subprocess. + + Attributes + ---------- + cmd : list[str] + The full command (including the "tmux" binary path). + stdout : list[str] + Lines of stdout output from tmux. + stderr : list[str] + Lines of stderr output from tmux. + returncode : int + The process return code. + + Examples + -------- + >>> import asyncio + >>> + >>> async def main(): + ... proc = await AsyncTmuxCmd.run('-V') + ... if proc.stderr: + ... raise exc.LibTmuxException( + ... f"Error invoking tmux: {proc.stderr}" + ... ) + ... print("tmux version:", proc.stdout) + ... + >>> asyncio.run(main()) + tmux version: [...] + + This is equivalent to calling: + + .. code-block:: console + + $ tmux -V + """ + + def __init__( + self, + cmd: list[str], + stdout: list[str], + stderr: list[str], + returncode: int, + ) -> None: + """ + Store the results of a completed tmux subprocess run. + + Parameters + ---------- + cmd : list[str] + The command used to invoke tmux. + stdout : list[str] + Captured lines from tmux stdout. + stderr : list[str] + Captured lines from tmux stderr. + returncode : int + Subprocess exit code. + """ + self.cmd: list[str] = cmd + self.stdout: list[str] = stdout + self.stderr: list[str] = stderr + self.returncode: int = returncode + + @classmethod + async def run(cls, *args: t.Any) -> AsyncTmuxCmd: + """ + Execute a tmux command asynchronously and capture its output. + + Parameters + ---------- + *args : str + Arguments to be passed after the "tmux" binary name. + + Returns + ------- + AsyncTmuxCmd + An instance containing the cmd, stdout, stderr, and returncode. + + Raises + ------ + exc.TmuxCommandNotFound + If no "tmux" executable is found in the user's PATH. + exc.LibTmuxException + If there's any unexpected exception creating or communicating + with the tmux subprocess. + """ + tmux_bin: str | None = shutil.which("tmux") + if not tmux_bin: + msg = "tmux executable not found in PATH" + raise exc.TmuxCommandNotFound( + msg, + ) + + cmd: list[str] = [tmux_bin] + [str(c) for c in args] + + try: + process: asyncio.subprocess.Process = await asyncio.create_subprocess_exec( + *cmd, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + stdout_bytes, stderr_bytes = await process.communicate() + returncode: int = ( + process.returncode if process.returncode is not None else -1 + ) + + except Exception as e: + logger.exception("Exception for %s", " ".join(cmd)) + msg = f"Exception while running tmux command: {e}" + raise exc.LibTmuxException( + msg, + ) from e + + # Decode bytes to string with error handling + stdout = stdout_bytes.decode(errors="backslashreplace") + stderr = stderr_bytes.decode(errors="backslashreplace") + + # Split on newlines and filter empty lines + stdout_split: list[str] = stdout.split("\n") + # remove trailing newlines from stdout + while stdout_split and stdout_split[-1] == "": + stdout_split.pop() + + stderr_split = stderr.split("\n") + stderr_split = list(filter(None, stderr_split)) # filter empty values + + # Workaround for tmux "has-session" command behavior + if "has-session" in cmd and stderr_split and not stdout_split: + # If `has-session` fails, it might output an error on stderr + # with nothing on stdout. We replicate the original logic here: + stdout_split = [stderr_split[0]] + + logger.debug("stdout for %s: %s", " ".join(cmd), stdout_split) + logger.debug("stderr for %s: %s", " ".join(cmd), stderr_split) + + return cls( + cmd=cmd, + stdout=stdout_split, + stderr=stderr_split, + returncode=returncode, + ) + + def get_version() -> LooseVersion: """Return tmux version. diff --git a/src/libtmux/pane.py b/src/libtmux/pane.py index 7f126f452..7da8451c3 100644 --- a/src/libtmux/pane.py +++ b/src/libtmux/pane.py @@ -14,7 +14,7 @@ import warnings from libtmux import exc -from libtmux.common import has_gte_version, has_lt_version, tmux_cmd +from libtmux.common import AsyncTmuxCmd, has_gte_version, has_lt_version, tmux_cmd from libtmux.constants import ( PANE_DIRECTION_FLAG_MAP, RESIZE_ADJUSTMENT_DIRECTION_FLAG_MAP, @@ -24,6 +24,9 @@ from libtmux.formats import FORMAT_SEPARATOR from libtmux.neo import Obj, fetch_obj +__all__ = ["Pane", "PaneDirection"] + + if t.TYPE_CHECKING: import sys import types @@ -202,6 +205,509 @@ def cmd( return self.server.cmd(cmd, *args, target=target) + async def acmd( + self, + cmd: str, + *args: t.Any, + target: str | int | None = None, + ) -> AsyncTmuxCmd: + """Execute tmux subcommand within pane context. + + Automatically binds target by adding ``-t`` for object's pane ID to the + command. Pass ``target`` to keyword arguments to override. + + Examples + -------- + >>> import asyncio + >>> async def test_acmd(): + ... result = await pane.acmd('split-window', '-P') + ... print(result.stdout[0]) + >>> asyncio.run(test_acmd()) + libtmux...:... + + From raw output to an enriched `Pane` object: + + >>> async def test_from_pane(): + ... pane_id_result = await pane.acmd( + ... 'split-window', '-P', '-F#{pane_id}' + ... ) + ... return Pane.from_pane_id( + ... pane_id=pane_id_result.stdout[0], + ... server=session.server + ... ) + >>> asyncio.run(test_from_pane()) + Pane(%... Window(@... ...:..., Session($1 libtmux_...))) + + Parameters + ---------- + target : str, optional + Optional custom target override. By default, the target is the pane ID. + + Returns + ------- + :meth:`server.cmd` + """ + if target is None: + target = self.pane_id + + return await self.server.acmd(cmd, *args, target=target) + + async def asend_keys( + self, + cmd: str, + enter: bool | None = True, + suppress_history: bool | None = False, + literal: bool | None = False, + ) -> None: + r"""``$ tmux send-keys`` to the pane asynchronously. + + This is the async version of :meth:`send_keys`. It uses ``await self.acmd()`` + for non-blocking command execution, making it suitable for async applications + and enabling concurrent command execution across multiple panes. + + A leading space character is added to cmd to avoid polluting the + user's history when suppress_history is True. + + Parameters + ---------- + cmd : str + Text or input into pane + enter : bool, optional + Send enter after sending the input, default True. + suppress_history : bool, optional + Prepend a space to command to suppress shell history, default False. + + .. versionchanged:: 0.14 + + Default changed from True to False. + literal : bool, optional + Send keys literally, default False. + + See Also + -------- + :meth:`send_keys` : Synchronous version of this method + :meth:`acapture_pane` : Capture pane output asynchronously + :meth:`acmd` : Execute arbitrary tmux commands asynchronously + + Notes + ----- + This method is non-blocking and suitable for use in async applications. + It's particularly powerful when sending commands to multiple panes concurrently + using ``asyncio.gather()``, which can significantly improve performance + compared to sequential execution. + + .. versionadded:: 0.48.0 + + Added async send_keys support. + + Examples + -------- + Basic command execution: + + >>> import asyncio + >>> async def test_basic_send(): + ... test_session = await server.anew_session("asend_basic") + ... pane = test_session.active_pane + ... await pane.asend_keys('echo "Hello world"', enter=True) + ... # Wait a moment for command to execute + ... await asyncio.sleep(0.1) + ... output = pane.capture_pane() + ... has_hello = any("Hello world" in line for line in output) + ... await server.acmd("kill-session", target="asend_basic") + ... return has_hello + >>> asyncio.run(test_basic_send()) + True + + Send without enter: + + >>> import asyncio + >>> async def test_no_enter(): + ... test_session = await server.anew_session("asend_no_enter") + ... pane = test_session.active_pane + ... await pane.asend_keys('echo test', enter=False) + ... await server.acmd("kill-session", target="asend_no_enter") + ... # Command sent but not executed (no enter) + >>> asyncio.run(test_no_enter()) + + Literal mode (special characters sent as-is): + + >>> import asyncio + >>> async def test_literal(): + ... test_session = await server.anew_session("asend_literal") + ... pane = test_session.active_pane + ... await pane.asend_keys('C-c', literal=True, enter=False) + ... await server.acmd("kill-session", target="asend_literal") + ... # Sends literal "C-c" text, not Ctrl-C signal + >>> asyncio.run(test_literal()) + + Concurrent command execution across multiple panes: + + >>> import asyncio + >>> async def test_concurrent_send(): + ... test_session = await server.anew_session("asend_concurrent") + ... window = test_session.active_window + ... pane1 = window.active_pane + ... pane2 = window.split() + ... pane3 = window.split() + ... # Send commands to all panes concurrently + ... await asyncio.gather( + ... pane1.asend_keys('echo pane1'), + ... pane2.asend_keys('echo pane2'), + ... pane3.asend_keys('echo pane3'), + ... ) + ... await server.acmd("kill-session", target="asend_concurrent") + ... # All three commands sent in parallel + >>> asyncio.run(test_concurrent_send()) + """ + prefix = " " if suppress_history else "" + + if literal: + await self.acmd("send-keys", "-l", prefix + cmd) + else: + await self.acmd("send-keys", prefix + cmd) + + if enter: + await self.acmd("send-keys", "Enter") + + async def acapture_pane( + self, + start: t.Literal["-"] | int | None = None, + end: t.Literal["-"] | int | None = None, + ) -> str | list[str]: + """Capture text from pane asynchronously. + + This is the async version of :meth:`capture_pane`. It uses ``await self.acmd()`` + for non-blocking output capture, making it suitable for async applications + and enabling concurrent output capture from multiple panes. + + ``$ tmux capture-pane`` to pane. + ``$ tmux capture-pane -S -10`` to pane. + ``$ tmux capture-pane -E 3`` to pane. + ``$ tmux capture-pane -S - -E -`` to pane. + + Parameters + ---------- + start : str or int, optional + Specify the starting line number. + - Zero is the first line of the visible pane + - Positive numbers are lines in the visible pane + - Negative numbers are lines in the history + - ``"-"`` is the start of the history + Default: None (capture visible pane only) + end : str or int, optional + Specify the ending line number. + - Zero is the first line of the visible pane + - Positive numbers are lines in the visible pane + - Negative numbers are lines in the history + - ``"-"`` is the end of the visible pane + Default: None (capture to end of visible pane) + + Returns + ------- + str or list[str] + Captured pane content + + See Also + -------- + :meth:`capture_pane` : Synchronous version of this method + :meth:`asend_keys` : Send keys to pane asynchronously + :meth:`acmd` : Execute arbitrary tmux commands asynchronously + + Notes + ----- + This method is non-blocking and suitable for async applications. It is + especially helpful when capturing output from multiple panes concurrently + via ``asyncio.gather()``, which can significantly improve performance + compared to sequential capture. + + .. versionadded:: 0.48.0 + + Added async capture_pane support. + + Examples + -------- + Basic pane output capture: + + >>> import asyncio + >>> async def test_basic_capture(): + ... test_session = await server.anew_session("acapture_basic") + ... pane = test_session.active_pane + ... await pane.asend_keys('echo "Test output"') + ... await asyncio.sleep(0.1) + ... output = await pane.acapture_pane() + ... has_test = any("Test output" in line for line in output) + ... await server.acmd("kill-session", target="acapture_basic") + ... return has_test + >>> asyncio.run(test_basic_capture()) + True + + Capture with line range: + + >>> import asyncio + >>> async def test_range_capture(): + ... test_session = await server.anew_session("acapture_range") + ... pane = test_session.active_pane + ... # Send multiple lines + ... await pane.asend_keys('echo line1') + ... await pane.asend_keys('echo line2') + ... await pane.asend_keys('echo line3') + ... await asyncio.sleep(0.1) + ... # Capture last 5 lines + ... output = await pane.acapture_pane(start=-5, end="-") + ... is_list = isinstance(output, list) + ... await server.acmd("kill-session", target="acapture_range") + ... return is_list + >>> asyncio.run(test_range_capture()) + True + + Concurrent output capture from multiple panes: + + >>> import asyncio + >>> async def test_concurrent_capture(): + ... test_session = await server.anew_session("acapture_concurrent") + ... window = test_session.active_window + ... pane1 = window.active_pane + ... pane2 = window.split() + ... pane3 = window.split() + ... # Send commands to all panes + ... await asyncio.gather( + ... pane1.asend_keys('echo output1'), + ... pane2.asend_keys('echo output2'), + ... pane3.asend_keys('echo output3'), + ... ) + ... await asyncio.sleep(0.1) + ... # Capture output from all panes concurrently + ... outputs = await asyncio.gather( + ... pane1.acapture_pane(), + ... pane2.acapture_pane(), + ... pane3.acapture_pane(), + ... ) + ... await server.acmd("kill-session", target="acapture_concurrent") + ... return len(outputs) + >>> asyncio.run(test_concurrent_capture()) + 3 + """ + cmd_parts: list[str] = ["capture-pane", "-p"] + if start is not None: + cmd_parts.extend(["-S", str(start)]) + if end is not None: + cmd_parts.extend(["-E", str(end)]) + result = await self.acmd(*cmd_parts) + return result.stdout + + async def asplit( + self, + /, + target: int | str | None = None, + start_directory: StrPath | None = None, + attach: bool = False, + direction: PaneDirection | None = None, + full_window_split: bool | None = None, + zoom: bool | None = None, + shell: str | None = None, + size: str | int | None = None, + environment: dict[str, str] | None = None, + ) -> Pane: + """Split window asynchronously and return :class:`Pane`. + + This is the async version of :meth:`split`. It uses ``await self.acmd()`` + for non-blocking pane creation, making it suitable for async applications + and enabling concurrent pane creation. + + By default, splits beneath the current pane. + + Parameters + ---------- + target : optional + Optional, custom *target-pane*, used by :meth:`Window.asplit`. + attach : bool, optional + Make new pane the current pane after creating it, default False. + start_directory : str or PathLike, optional + Working directory in which the new pane is created. + direction : PaneDirection, optional + Direction to split: Above, Below (default), Left, or Right. + full_window_split : bool, optional + Split across full window width or height, rather than active pane. + zoom : bool, optional + Expand pane after creation. + shell : str, optional + Execute a command when splitting the pane. The pane will close + when the command exits. + + .. warning:: + + When this command exits, the pane will close. This feature is + useful for long-running processes where automatic cleanup is desired. + size : int or str, optional + Cell/row count or percentage to occupy with respect to current window. + Examples: ``50`` (50 cells), ``"50%"`` (50 percent). + environment : dict[str, str], optional + Environmental variables for new pane. + + .. note:: + + Requires tmux 3.0+. On older versions, this parameter is ignored + with a warning. + + Returns + ------- + :class:`Pane` + The newly created pane object + + Raises + ------ + :exc:`exc.LibTmuxException` + If tmux command execution fails (e.g., pane too small) + + See Also + -------- + :meth:`split` : Synchronous version of this method + :meth:`asend_keys` : Send keys to pane asynchronously + :meth:`acapture_pane` : Capture pane output asynchronously + + Notes + ----- + This method is non-blocking and suitable for use in async applications. + It's particularly powerful when creating multiple panes concurrently + using ``asyncio.gather()``, which can significantly improve performance + compared to sequential creation. + + .. versionadded:: 0.48.0 + + Added async split_window support. + + Examples + -------- + Basic horizontal split (default - below current pane): + + >>> import asyncio + >>> async def test_basic_split(): + ... test_session = await server.anew_session("asplit_basic") + ... pane = test_session.active_pane + ... new_pane = await pane.asplit() + ... pane_count = len(test_session.active_window.panes) + ... await server.acmd("kill-session", target="asplit_basic") + ... return pane_count + >>> asyncio.run(test_basic_split()) + 2 + + Vertical split with custom directory: + + >>> import asyncio + >>> async def test_vertical_split(): + ... test_session = await server.anew_session("asplit_vertical") + ... pane = test_session.active_pane + ... new_pane = await pane.asplit( + ... direction=PaneDirection.Right, + ... start_directory='/tmp' + ... ) + ... pane_count = len(test_session.active_window.panes) + ... await server.acmd("kill-session", target="asplit_vertical") + ... return pane_count + >>> asyncio.run(test_vertical_split()) + 2 + + Split with size specification: + + >>> import asyncio + >>> async def test_split_with_size(): + ... test_session = await server.anew_session("asplit_size") + ... pane = test_session.active_pane + ... new_pane = await pane.asplit(size="30%") + ... pane_count = len(test_session.active_window.panes) + ... await server.acmd("kill-session", target="asplit_size") + ... return pane_count + >>> asyncio.run(test_split_with_size()) + 2 + + Concurrent multi-pane creation: + + >>> import asyncio + >>> async def test_concurrent_splits(): + ... test_session = await server.anew_session("asplit_concurrent") + ... window = test_session.active_window + ... base_pane = window.active_pane + ... # Create multiple panes concurrently + ... new_panes = await asyncio.gather( + ... base_pane.asplit(direction=PaneDirection.Below), + ... base_pane.asplit(direction=PaneDirection.Right), + ... ) + ... pane_count = len(window.panes) + ... await server.acmd("kill-session", target="asplit_concurrent") + ... return pane_count >= 3 + >>> asyncio.run(test_concurrent_splits()) + True + """ + tmux_formats = ["#{pane_id}" + FORMAT_SEPARATOR] + + tmux_args: tuple[str, ...] = () + + if direction: + tmux_args += tuple(PANE_DIRECTION_FLAG_MAP[direction]) + else: + tmux_args += tuple(PANE_DIRECTION_FLAG_MAP[PaneDirection.Below]) + + if size is not None: + if has_lt_version("3.1"): + if isinstance(size, str) and size.endswith("%"): + tmux_args += (f"-p{str(size).rstrip('%')}",) + else: + warnings.warn( + 'Ignored size. Use percent in tmux < 3.1, e.g. "size=50%"', + stacklevel=2, + ) + else: + tmux_args += (f"-l{size}",) + + if full_window_split: + tmux_args += ("-f",) + + if zoom: + tmux_args += ("-Z",) + + tmux_args += ("-P", "-F{}".format("".join(tmux_formats))) # output + + if start_directory: + # as of 2014-02-08 tmux 1.9-dev doesn't expand ~ in new-window -c. + start_path = pathlib.Path(start_directory).expanduser() + tmux_args += (f"-c{start_path}",) + + if not attach: + tmux_args += ("-d",) + + if environment: + if has_gte_version("3.0"): + for k, v in environment.items(): + tmux_args += (f"-e{k}={v}",) + else: + logger.warning( + "Environment flag ignored, tmux 3.0 or newer required.", + ) + + if shell: + tmux_args += (shell,) + + pane_cmd = await self.acmd("split-window", *tmux_args, target=target) + + # tmux < 1.7. This is added in 1.7. + if pane_cmd.stderr: + if "pane too small" in pane_cmd.stderr: + raise exc.LibTmuxException(pane_cmd.stderr) + + raise exc.LibTmuxException( + pane_cmd.stderr, + self.__dict__, + self.window.panes, + ) + + pane_output = pane_cmd.stdout[0] + + pane_formatters = dict( + zip(["pane_id"], pane_output.split(FORMAT_SEPARATOR), strict=False), + ) + + return self.from_pane_id(server=self.server, pane_id=pane_formatters["pane_id"]) + """ Commands (tmux-like) """ @@ -335,6 +841,10 @@ def capture_pane( Negative numbers are lines in the history. `-` is the end of the visible pane Default: None + + See Also + -------- + :meth:`acapture_pane` : Async version of this method """ cmd = ["capture-pane", "-p"] if start is not None: @@ -370,6 +880,10 @@ def send_keys( literal : bool, optional Send keys literally, default False. + See Also + -------- + :meth:`asend_keys` : Async version of this method + Examples -------- >>> pane = window.split(shell='sh') @@ -587,6 +1101,10 @@ def split( environment: dict, optional Environmental variables for new pane. tmux 3.0+ only. Passthrough to ``-e``. + See Also + -------- + :meth:`asplit` : Async version of this method + Examples -------- >>> (pane.at_left, pane.at_right, diff --git a/src/libtmux/server.py b/src/libtmux/server.py index 17b290c34..6b919b97b 100644 --- a/src/libtmux/server.py +++ b/src/libtmux/server.py @@ -24,6 +24,7 @@ from libtmux.window import Window from .common import ( + AsyncTmuxCmd, EnvironmentMixin, PaneDict, SessionDict, @@ -250,8 +251,12 @@ def cmd( Output of `tmux -L ... new-window -P -F#{window_id}` to a `Window` object: - >>> Window.from_window_id(window_id=session.cmd( - ... 'new-window', '-P', '-F#{window_id}').stdout[0], server=session.server) + >>> Window.from_window_id( + ... window_id=session.cmd( + ... 'new-window', '-P', '-F#{window_id}' + ... ).stdout[0], + ... server=session.server, + ... ) Window(@4 3:..., Session($1 libtmux_...)) Create a pane from a window: @@ -262,7 +267,9 @@ def cmd( Output of `tmux -L ... split-window -P -F#{pane_id}` to a `Pane` object: >>> Pane.from_pane_id(pane_id=window.cmd( - ... 'split-window', '-P', '-F#{pane_id}').stdout[0], server=window.server) + ... 'split-window', '-P', '-F#{pane_id}').stdout[0], + ... server=window.server + ... ) Pane(%... Window(@... ...:..., Session($1 libtmux_...))) Parameters @@ -300,6 +307,516 @@ def cmd( return tmux_cmd(*svr_args, *cmd_args) + async def acmd( + self, + cmd: str, + *args: t.Any, + target: str | int | None = None, + ) -> AsyncTmuxCmd: + """Execute tmux command respective of socket name and file, return output. + + Examples + -------- + >>> import asyncio + >>> async def test_acmd(): + ... result = await server.acmd('display-message', 'hi') + ... print(result.stdout) + >>> asyncio.run(test_acmd()) + [] + + New session: + + >>> async def test_new_session(): + ... result = await server.acmd( + ... 'new-session', '-d', '-P', '-F#{session_id}' + ... ) + ... print(result.stdout[0]) + >>> asyncio.run(test_new_session()) + $... + + Output of `tmux -L ... new-window -P -F#{window_id}` to a `Window` object: + + >>> async def test_new_window(): + ... result = await session.acmd('new-window', '-P', '-F#{window_id}') + ... window_id = result.stdout[0] + ... window = Window.from_window_id(window_id=window_id, server=server) + ... print(window) + >>> asyncio.run(test_new_window()) + Window(@... ...:..., Session($... libtmux_...)) + + Create a pane from a window: + + >>> async def test_split_window(): + ... result = await server.acmd('split-window', '-P', '-F#{pane_id}') + ... print(result.stdout[0]) + >>> asyncio.run(test_split_window()) + %... + + Output of `tmux -L ... split-window -P -F#{pane_id}` to a `Pane` object: + + >>> async def test_pane(): + ... result = await window.acmd('split-window', '-P', '-F#{pane_id}') + ... pane_id = result.stdout[0] + ... pane = Pane.from_pane_id(pane_id=pane_id, server=server) + ... print(pane) + >>> asyncio.run(test_pane()) + Pane(%... Window(@... ...:..., Session($1 libtmux_...))) + + Parameters + ---------- + target : str, optional + Optional custom target. + + Returns + ------- + :class:`common.AsyncTmuxCmd` + """ + svr_args: list[str | int] = [cmd] + cmd_args: list[str | int] = [] + if self.socket_name: + svr_args.insert(0, f"-L{self.socket_name}") + if self.socket_path: + svr_args.insert(0, f"-S{self.socket_path}") + if self.config_file: + svr_args.insert(0, f"-f{self.config_file}") + if self.colors: + if self.colors == 256: + svr_args.insert(0, "-2") + elif self.colors == 88: + svr_args.insert(0, "-8") + else: + raise exc.UnknownColorOption + + cmd_args = ["-t", str(target), *args] if target is not None else [*args] + + return await AsyncTmuxCmd.run(*svr_args, *cmd_args) + + async def ahas_session(self, target_session: str, exact: bool = True) -> bool: + """Return True if session exists asynchronously. + + This is the async version of :meth:`has_session`. It uses + ``await self.acmd()`` for non-blocking session existence checks, making + it suitable for async applications. + + Equivalent to:: + + $ tmux has-session -t + + Parameters + ---------- + target_session : str + Session name to check for existence + exact : bool, optional + Match the session name exactly. When True (default), tmux will only + match exact session names. When False, tmux uses fnmatch(3) pattern + matching. Internally prepends ``=`` to the session when exact=True. + + .. note:: + + Exact matching requires tmux 2.1+. On older versions, this parameter + is ignored and fnmatch behavior is used. + + Returns + ------- + bool + True if session exists, False otherwise + + Raises + ------ + :exc:`exc.BadSessionName` + If target_session contains invalid characters (periods or colons) + + See Also + -------- + :meth:`has_session` : Synchronous version of this method + :meth:`anew_session` : Create a new session asynchronously + :meth:`kill_session` : Kill a session + + Notes + ----- + This method is non-blocking and suitable for use in async applications. + It's particularly useful when checking multiple sessions concurrently using + ``asyncio.gather()``. + + .. versionadded:: 0.48.0 + + Added async session existence check support. + + Examples + -------- + Basic session existence check: + + >>> import asyncio + >>> async def check_session_exists(): + ... session = await server.anew_session("test_ahas_basic") + ... exists = await server.ahas_session("test_ahas_basic") + ... await server.acmd("kill-session", target="test_ahas_basic") + ... return exists + >>> asyncio.run(check_session_exists()) + True + + Checking for nonexistent session: + + >>> import asyncio + >>> async def check_nonexistent(): + ... return await server.ahas_session("nonexistent_xyz_123") + >>> asyncio.run(check_nonexistent()) + False + + Checking multiple sessions concurrently: + + >>> import asyncio + >>> async def check_multiple_sessions(): + ... # Create sessions concurrently + ... await asyncio.gather( + ... server.anew_session("ahas_s1"), + ... server.anew_session("ahas_s2"), + ... server.anew_session("ahas_s3"), + ... ) + ... # Check all sessions concurrently + ... results = await asyncio.gather( + ... server.ahas_session("ahas_s1"), + ... server.ahas_session("ahas_s2"), + ... server.ahas_session("ahas_s3"), + ... ) + ... # Cleanup + ... await asyncio.gather( + ... server.acmd("kill-session", target="ahas_s1"), + ... server.acmd("kill-session", target="ahas_s2"), + ... server.acmd("kill-session", target="ahas_s3"), + ... ) + ... return results + >>> asyncio.run(check_multiple_sessions()) + [True, True, True] + + Using exact matching: + + >>> import asyncio + >>> async def test_exact_matching(): + ... session = await server.anew_session("exact_match_test") + ... # Exact match - must match full name + ... exact_result = await server.ahas_session("exact_match_test", exact=True) + ... # Partial name should not match with exact=True + ... partial_result = await server.ahas_session("exact", exact=True) + ... await server.acmd("kill-session", target="exact_match_test") + ... return (exact_result, partial_result) + >>> asyncio.run(test_exact_matching()) + (True, False) + """ + session_check_name(target_session) + + if exact and has_gte_version("2.1"): + target_session = f"={target_session}" + + proc = await self.acmd("has-session", target=target_session) + + return bool(not proc.returncode) + + async def anew_session( + self, + session_name: str | None = None, + kill_session: bool = False, + attach: bool = False, + start_directory: StrPath | None = None, + window_name: str | None = None, + window_command: str | None = None, + x: int | DashLiteral | None = None, + y: int | DashLiteral | None = None, + environment: dict[str, str] | None = None, + *args: t.Any, + **kwargs: t.Any, + ) -> Session: + """Create new session asynchronously, returns new :class:`Session`. + + This is the async version of :meth:`new_session`. It uses ``await self.acmd()`` + for non-blocking session creation, making it suitable for async applications + and enabling concurrent session creation. + + Uses ``-P`` flag to print session info, ``-F`` for return formatting, + and returns a new :class:`Session` object. + + Equivalent to:: + + $ tmux new-session -d -s + + .. note:: + + ``attach=False`` (default) creates the session in the background:: + + $ tmux new-session -d + + This is typically desired for async operations to avoid blocking. + + Parameters + ---------- + session_name : str, optional + Name for the new session. If not provided, tmux will auto-generate + a name (typically sequential numbers: 0, 1, 2, etc.). + + Equivalent to:: + + $ tmux new-session -s + + attach : bool, optional + Create session in the foreground (True) or background (False, default). + For async operations, background creation is typically preferred. + + ``attach=False`` is equivalent to:: + + $ tmux new-session -d + + Other Parameters + ---------------- + kill_session : bool, optional + If True, kill the existing session with the same name before creating + a new one. If False (default) and a session with the same name exists, + raises :exc:`exc.TmuxSessionExists`. + + Useful for testing workspaces and ensuring a clean slate. + + start_directory : str or PathLike, optional + Working directory in which the new session is created. All windows + and panes in the session will default to this directory. + + Supports pathlib.Path objects and tilde expansion (``~/``). + + Equivalent to:: + + $ tmux new-session -c + + window_name : str, optional + Name for the initial window created in the session. + + Equivalent to:: + + $ tmux new-session -n + + window_command : str, optional + Shell command to execute when starting the session. The window will + automatically close when the command exits. + + .. warning:: + + When this command exits, the window will close. This feature is + useful for long-running processes where automatic cleanup is desired. + + x : int or '-', optional + Force the specified width (in columns) instead of the tmux default + for a detached session. Use '-' for tmux default. + + y : int or '-', optional + Force the specified height (in rows) instead of the tmux default + for a detached session. Use '-' for tmux default. + + environment : dict[str, str], optional + Dictionary of environment variables to set in the new session. + Each key-value pair will be set as an environment variable. + + .. note:: + + Requires tmux 3.2+. On older versions, this parameter is ignored + with a warning. + + Equivalent to:: + + $ tmux new-session -e KEY1=value1 -e KEY2=value2 + + Returns + ------- + :class:`Session` + The newly created session object + + Raises + ------ + :exc:`exc.BadSessionName` + If session_name contains invalid characters (periods or colons) + :exc:`exc.TmuxSessionExists` + If a session with the same name already exists and kill_session=False + :exc:`exc.LibTmuxException` + If tmux command execution fails + + See Also + -------- + :meth:`new_session` : Synchronous version of this method + :meth:`ahas_session` : Check if a session exists asynchronously + :meth:`kill_session` : Kill a session + :class:`Session` : Session object documentation + + Notes + ----- + This method is non-blocking and suitable for use in async applications. + It's particularly powerful when creating multiple sessions concurrently + using ``asyncio.gather()``, which can significantly improve performance + compared to sequential creation. + + The method temporarily removes the ``TMUX`` environment variable during + session creation to allow creating sessions from within tmux itself. + + .. versionadded:: 0.48.0 + + Added async session creation support. + + Examples + -------- + Sessions can be created without a session name (auto-generated IDs): + + >>> import asyncio + >>> async def test_auto_generated(): + ... session1 = await server.anew_session() + ... session2 = await server.anew_session() + ... # Both have auto-generated names + ... has_session1 = session1.session_name is not None + ... has_session2 = session2.session_name is not None + ... # Cleanup + ... await asyncio.gather( + ... server.acmd("kill-session", target=session1.session_name), + ... server.acmd("kill-session", target=session2.session_name), + ... ) + ... return (has_session1, has_session2) + >>> asyncio.run(test_auto_generated()) + (True, True) + + With a custom `session_name`: + + >>> import asyncio + >>> async def test_custom_name(): + ... session = await server.anew_session(session_name='my_project') + ... name = session.session_name + ... await server.acmd("kill-session", target="my_project") + ... return name + >>> asyncio.run(test_custom_name()) + 'my_project' + + With custom working directory: + + >>> import asyncio + >>> async def test_start_directory(): + ... from pathlib import Path + ... session = await server.anew_session( + ... session_name='dev_session', + ... start_directory='/tmp' + ... ) + ... # Verify session was created + ... exists = await server.ahas_session('dev_session') + ... await server.acmd("kill-session", target="dev_session") + ... return exists + >>> asyncio.run(test_start_directory()) + True + + Creating multiple sessions concurrently: + + >>> import asyncio + >>> async def test_concurrent_creation(): + ... sessions = await asyncio.gather( + ... server.anew_session(session_name='anew_frontend'), + ... server.anew_session(session_name='anew_backend'), + ... server.anew_session(session_name='anew_database'), + ... ) + ... names = [s.session_name for s in sessions] + ... # Cleanup + ... await asyncio.gather( + ... server.acmd("kill-session", target="anew_frontend"), + ... server.acmd("kill-session", target="anew_backend"), + ... server.acmd("kill-session", target="anew_database"), + ... ) + ... return (len(sessions), names) + >>> asyncio.run(test_concurrent_creation()) + (3, ['anew_frontend', 'anew_backend', 'anew_database']) + + With custom window configuration: + + >>> import asyncio + >>> async def test_custom_window(): + ... session = await server.anew_session( + ... session_name='custom_window_test', + ... window_name='main' + ... ) + ... window_name = session.active_window.window_name + ... await server.acmd("kill-session", target="custom_window_test") + ... return window_name + >>> asyncio.run(test_custom_window()) + 'main' + """ + if session_name is not None: + session_check_name(session_name) + + if await self.ahas_session(session_name): + if kill_session: + await self.acmd("kill-session", target=session_name) + logger.info("session %s exists. killed it.", session_name) + else: + msg = f"Session named {session_name} exists" + raise exc.TmuxSessionExists( + msg, + ) + + logger.debug("creating session %s", session_name) + + env = os.environ.get("TMUX") + + if env: + del os.environ["TMUX"] + + tmux_args: tuple[str | int, ...] = ( + "-P", + "-F#{session_id}", # output + ) + + if session_name is not None: + tmux_args += (f"-s{session_name}",) + + if not attach: + tmux_args += ("-d",) + + if start_directory: + # as of 2014-02-08 tmux 1.9-dev doesn't expand ~ in new-session -c. + start_directory = pathlib.Path(start_directory).expanduser() + tmux_args += ("-c", str(start_directory)) + + if window_name: + tmux_args += ("-n", window_name) + + if x is not None: + tmux_args += ("-x", x) + + if y is not None: + tmux_args += ("-y", y) + + if environment: + if has_gte_version("3.2"): + for k, v in environment.items(): + tmux_args += (f"-e{k}={v}",) + else: + logger.warning( + "Environment flag ignored, tmux 3.2 or newer required.", + ) + + if window_command: + tmux_args += (window_command,) + + proc = await self.acmd("new-session", *tmux_args) + + if proc.stderr: + raise exc.LibTmuxException(proc.stderr) + + session_stdout = proc.stdout[0] + + if env: + os.environ["TMUX"] = env + + session_formatters = dict( + zip( + ["session_id"], + session_stdout.split(formats.FORMAT_SEPARATOR), + strict=False, + ), + ) + + return Session.from_session_id( + server=self, + session_id=session_formatters["session_id"], + ) + @property def attached_sessions(self) -> list[Session]: """Return active :class:`Session`s. @@ -334,6 +851,10 @@ def has_session(self, target_session: str, exact: bool = True) -> bool: Returns ------- bool + + See Also + -------- + :meth:`ahas_session` : Async version of this method """ session_check_name(target_session) @@ -492,6 +1013,10 @@ def new_session( ------ :exc:`exc.BadSessionName` + See Also + -------- + :meth:`anew_session` : Async version of this method + Examples -------- Sessions can be created without a session name (0.14.2+): diff --git a/src/libtmux/session.py b/src/libtmux/session.py index 26b55426d..4c93fc103 100644 --- a/src/libtmux/session.py +++ b/src/libtmux/session.py @@ -22,6 +22,7 @@ from . import exc from .common import ( + AsyncTmuxCmd, EnvironmentMixin, WindowDict, handle_option_error, @@ -235,6 +236,426 @@ def cmd( target = self.session_id return self.server.cmd(cmd, *args, target=target) + async def acmd( + self, + cmd: str, + *args: t.Any, + target: str | int | None = None, + ) -> AsyncTmuxCmd: + """Execute tmux subcommand within session context. + + Automatically binds target by adding ``-t`` for object's session ID to the + command. Pass ``target`` to keyword arguments to override. + + Examples + -------- + >>> import asyncio + >>> async def test_acmd(): + ... result = await session.acmd('new-window', '-P') + ... print(result.stdout[0]) + >>> asyncio.run(test_acmd()) + libtmux...:....0 + + From raw output to an enriched `Window` object: + + >>> async def test_from_window(): + ... window_id_result = await session.acmd( + ... 'new-window', '-P', '-F#{window_id}' + ... ) + ... return Window.from_window_id( + ... window_id=window_id_result.stdout[0], + ... server=session.server + ... ) + >>> asyncio.run(test_from_window()) + Window(@... ...:..., Session($1 libtmux_...)) + + Parameters + ---------- + target : str, optional + Optional custom target override. By default, the target is the session ID. + + Returns + ------- + :meth:`server.cmd` + + Notes + ----- + .. versionchanged:: 0.34 + + Passing target by ``-t`` is ignored. Use ``target`` keyword argument instead. + + .. versionchanged:: 0.8 + + Renamed from ``.tmux`` to ``.cmd``. + """ + if target is None: + target = self.session_id + return await self.server.acmd(cmd, *args, target=target) + + async def arename_session(self, new_name: str) -> Session: + """Rename session asynchronously and return session object. + + This is the async version of :meth:`rename_session`. It uses + ``await self.acmd()`` for non-blocking session renaming, making it + suitable for async applications. + + Equivalent to:: + + $ tmux rename-session + + Parameters + ---------- + new_name : str + New name for the session. Must not contain periods (.) or colons (:). + + Returns + ------- + :class:`Session` + Returns self (the session object) with updated name + + Raises + ------ + :exc:`exc.BadSessionName` + If new_name contains invalid characters (periods or colons) + :exc:`exc.LibTmuxException` + If tmux command execution fails + + See Also + -------- + :meth:`rename_session` : Synchronous version of this method + :meth:`Session.session_name` : Property to get current session name + + Notes + ----- + This method is non-blocking and suitable for use in async applications. + The session object is automatically refreshed after renaming to ensure + the session_name property reflects the new name. + + On tmux 2.7 with BSD systems, a "no current client" warning may be + raised but is safely ignored as it's a known issue fixed in later versions. + + .. versionadded:: 0.48.0 + + Added async session renaming support. + + Examples + -------- + Basic session rename: + + >>> import asyncio + >>> async def test_rename(): + ... test_session = await server.anew_session("arename_original") + ... renamed_session = await test_session.arename_session("arename_new") + ... new_name = renamed_session.session_name + ... await server.acmd("kill-session", target="arename_new") + ... return new_name + >>> asyncio.run(test_rename()) + 'arename_new' + + Rename and verify: + + >>> import asyncio + >>> async def test_rename_verify(): + ... test_session = await server.anew_session("arename_verify_old") + ... old_name = test_session.session_name + ... await test_session.arename_session("arename_verify_new") + ... # Verify the rename + ... has_new = test_session.server.has_session("arename_verify_new") + ... has_old = test_session.server.has_session(old_name) + ... await server.acmd("kill-session", target="arename_verify_new") + ... return (test_session.session_name, has_new, has_old) + >>> asyncio.run(test_rename_verify()) + ('arename_verify_new', True, False) + + Chaining operations: + + >>> import asyncio + >>> async def test_chaining(): + ... # arename_session returns self, allowing chaining + ... test_session = await server.anew_session("arename_chain_old") + ... renamed_session = await test_session.arename_session( + ... "arename_chain_new" + ... ) + ... window = await renamed_session.anew_window(window_name="main") + ... result = (renamed_session.session_name, window.window_name) + ... await server.acmd("kill-session", target="arename_chain_new") + ... return result + >>> asyncio.run(test_chaining()) + ('arename_chain_new', 'main') + """ + session_check_name(new_name) + + proc = await self.acmd("rename-session", new_name) + + if proc.stderr: + if has_version("2.7") and "no current client" in proc.stderr: + """tmux 2.7 raises "no current client" warning on BSD systems. + + Should be fixed next release: + + - https://www.mail-archive.com/tech@openbsd.org/msg45186.html + - https://marc.info/?l=openbsd-cvs&m=152183263526828&w=2 + """ + else: + raise exc.LibTmuxException(proc.stderr) + + self.refresh() + + return self + + async def anew_window( + self, + window_name: str | None = None, + *, + start_directory: StrPath | None = None, + attach: bool = False, + window_index: str = "", + window_shell: str | None = None, + environment: dict[str, str] | None = None, + direction: WindowDirection | None = None, + target_window: str | None = None, + ) -> Window: + """Create new window asynchronously, returns new :class:`Window`. + + This is the async version of :meth:`new_window`. It uses ``await self.acmd()`` + for non-blocking window creation, making it suitable for async applications + and enabling concurrent window creation. + + By default, this will make the window active. For the new window + to be created and not set to current, pass in ``attach=False``. + + Equivalent to:: + + $ tmux new-window -n + + Parameters + ---------- + window_name : str, optional + Name for the new window. If not provided, tmux will auto-name + based on the shell command running in the window. + + start_directory : str or PathLike, optional + Working directory in which the new window is created. All panes + in the window will default to this directory. + + Supports pathlib.Path objects and tilde expansion (``~/``). + + Equivalent to:: + + $ tmux new-window -c + + attach : bool, optional + Make the new window the current (active) window after creating it. + Default is False, meaning the window is created in the background. + + When False (default):: + + $ tmux new-window -d + + window_index : str, optional + Create the new window at the given index position. Default is empty + string which creates the window in the next available position. + + Use to control window ordering or create windows at specific indices. + + window_shell : str, optional + Shell command to execute when starting the window. The window will + automatically close when the command exits. + + .. warning:: + + When this command exits, the window will close. This feature is + useful for long-running processes where automatic cleanup is desired. + + environment : dict[str, str], optional + Dictionary of environment variables to set in the new window. + Each key-value pair will be set as an environment variable. + + .. note:: + + Requires tmux 3.0+. On older versions, this parameter is ignored + with a warning. + + direction : WindowDirection, optional + Insert the new window before or after the target window. + Values: "before" or "after". + + .. note:: + + Requires tmux 3.2+. On older versions, this parameter is ignored + with a warning. + + target_window : str, optional + Target window identifier for positioning the new window when using + the direction parameter. + + .. note:: + + Requires tmux 3.2+. On older versions, this parameter is ignored + with a warning. + + Returns + ------- + :class:`Window` + The newly created window object + + Raises + ------ + :exc:`exc.LibTmuxException` + If tmux command execution fails + + See Also + -------- + :meth:`new_window` : Synchronous version of this method + :meth:`Session.kill_window` : Kill a window + :class:`Window` : Window object documentation + + Notes + ----- + This method is non-blocking and suitable for use in async applications. + It's particularly powerful when creating multiple windows concurrently + using ``asyncio.gather()``, which can significantly improve performance + compared to sequential creation. + + .. versionadded:: 0.48.0 + + Added async window creation support. + + .. versionchanged:: 0.28.0 + + ``attach`` default changed from ``True`` to ``False``. + + Examples + -------- + Basic window creation: + + >>> import asyncio + >>> async def test_basic_window(): + ... test_session = await server.anew_session("anew_window_basic") + ... window = await test_session.anew_window(window_name='editor') + ... name = window.window_name + ... await server.acmd("kill-session", target="anew_window_basic") + ... return name + >>> asyncio.run(test_basic_window()) + 'editor' + + With custom working directory: + + >>> import asyncio + >>> async def test_start_directory(): + ... from pathlib import Path + ... test_session = await server.anew_session("anew_window_dir") + ... window = await test_session.anew_window( + ... window_name='project', + ... start_directory='/tmp' + ... ) + ... # Verify window was created + ... name = window.window_name + ... await server.acmd("kill-session", target="anew_window_dir") + ... return name + >>> asyncio.run(test_start_directory()) + 'project' + + Creating multiple windows concurrently: + + >>> import asyncio + >>> async def test_concurrent_windows(): + ... test_session = await server.anew_session("anew_window_concurrent") + ... windows = await asyncio.gather( + ... test_session.anew_window(window_name='editor'), + ... test_session.anew_window(window_name='terminal'), + ... test_session.anew_window(window_name='logs'), + ... ) + ... names = [w.window_name for w in windows] + ... await server.acmd("kill-session", target="anew_window_concurrent") + ... return (len(windows), names) + >>> asyncio.run(test_concurrent_windows()) + (3, ['editor', 'terminal', 'logs']) + + With specific window index: + + >>> import asyncio + >>> async def test_window_index(): + ... test_session = await server.anew_session("anew_window_index") + ... window = await test_session.anew_window( + ... window_name='custom', + ... window_index='5' + ... ) + ... # Verify window was created with correct name + ... name = window.window_name + ... await server.acmd("kill-session", target="anew_window_index") + ... return name + >>> asyncio.run(test_window_index()) + 'custom' + """ + window_args: tuple[str, ...] = () + + if not attach: + window_args += ("-d",) + + window_args += ("-P",) + + # Catch empty string and default (`None`) + if start_directory: + # as of 2014-02-08 tmux 1.9-dev doesn't expand ~ in new-window -c. + start_directory = pathlib.Path(start_directory).expanduser() + window_args += (f"-c{start_directory}",) + + window_args += ("-F#{window_id}",) # output + if window_name is not None and isinstance(window_name, str): + window_args += ("-n", window_name) + + if environment: + if has_gte_version("3.0"): + for k, v in environment.items(): + window_args += (f"-e{k}={v}",) + else: + logger.warning( + "Environment flag ignored, requires tmux 3.0 or newer.", + ) + + if direction is not None: + if has_gte_version("3.2"): + window_args += (WINDOW_DIRECTION_FLAG_MAP[direction],) + else: + logger.warning( + "Direction flag ignored, requires tmux 3.1 or newer.", + ) + + target: str | None = None + if window_index is not None: + # empty string for window_index will use the first one available + target = f"{self.session_id}:{window_index}" + if target_window: + if has_gte_version("3.2"): + target = target_window + else: + logger.warning( + "Window target ignored, requires tmux 3.1 or newer.", + ) + elif window_index is not None: + # empty string for window_index will use the first one available + window_args += (f"-t{self.session_id}:{window_index}",) + + if window_shell: + window_args += (window_shell,) + + cmd = await self.acmd("new-window", *window_args, target=target) + + if cmd.stderr: + raise exc.LibTmuxException(cmd.stderr) + + window_output = cmd.stdout[0] + + window_formatters = dict( + zip(["window_id"], window_output.split(FORMAT_SEPARATOR), strict=False), + ) + + return Window.from_window_id( + server=self.server, + window_id=window_formatters["window_id"], + ) + """ Commands (tmux-like) """ @@ -563,6 +984,10 @@ def rename_session(self, new_name: str) -> Session: Raises ------ :exc:`exc.BadSessionName` + + See Also + -------- + :meth:`arename_session` : Async version of this method """ session_check_name(new_name) @@ -632,6 +1057,7 @@ def new_window( See Also -------- + :meth:`anew_window` : Async version of this method :meth:`Window.new_window()` Examples diff --git a/src/libtmux/window.py b/src/libtmux/window.py index e20eb26f3..70d971ef6 100644 --- a/src/libtmux/window.py +++ b/src/libtmux/window.py @@ -25,7 +25,7 @@ from libtmux.pane import Pane from . import exc -from .common import PaneDict, WindowOptionDict, handle_option_error +from .common import AsyncTmuxCmd, PaneDict, WindowOptionDict, handle_option_error if t.TYPE_CHECKING: import sys @@ -228,6 +228,177 @@ def cmd( return self.server.cmd(cmd, *args, target=target) + async def acmd( + self, + cmd: str, + *args: t.Any, + target: str | int | None = None, + ) -> AsyncTmuxCmd: + """Execute tmux subcommand within window context. + + Automatically binds target by adding ``-t`` for object's window ID to the + command. Pass ``target`` to keyword arguments to override. + + Examples + -------- + Create a pane from a window: + + >>> import asyncio + >>> async def test_acmd(): + ... result = await window.acmd('split-window', '-P', '-F#{pane_id}') + ... print(result.stdout[0]) + >>> asyncio.run(test_acmd()) + %... + + Magic, directly to a `Pane`: + + >>> async def test_from_pane(): + ... pane_id_result = await session.acmd( + ... 'split-window', '-P', '-F#{pane_id}' + ... ) + ... return Pane.from_pane_id( + ... pane_id=pane_id_result.stdout[0], + ... server=session.server + ... ) + >>> asyncio.run(test_from_pane()) + Pane(%... Window(@... ...:..., Session($1 libtmux_...))) + + Parameters + ---------- + target : str, optional + Optional custom target override. By default, the target is the window ID. + + Returns + ------- + :meth:`server.cmd` + """ + if target is None: + target = self.window_id + + return await self.server.acmd(cmd, *args, target=target) + + async def akill( + self, + all_except: bool | None = None, + ) -> None: + """Kill :class:`Window` asynchronously. + + This is the async version of :meth:`kill`. It uses ``await self.acmd()`` + for non-blocking window destruction, making it suitable for async applications. + + Equivalent to:: + + $ tmux kill-window + + When ``all_except=True``:: + + $ tmux kill-window -a + + Parameters + ---------- + all_except : bool, optional + If True, kill all windows in the session except this one. + If False or None (default), kill only this window. + + Useful for cleaning up all other windows while keeping one active. + + Raises + ------ + :exc:`exc.LibTmuxException` + If tmux command execution fails + + See Also + -------- + :meth:`kill` : Synchronous version of this method + :meth:`Session.kill_window` : Kill a window by target + :meth:`Session.anew_window` : Create a new window asynchronously + + Notes + ----- + This method is non-blocking and suitable for use in async applications. + After killing a window, the window object should not be used as it no + longer represents a valid tmux window. + + When ``all_except=True``, all other windows in the session are destroyed, + leaving only the current window active. This is useful for cleaning up + a session to a single window. + + .. versionadded:: 0.48.0 + + Added async window killing support. + + Examples + -------- + Kill a single window: + + >>> import asyncio + >>> async def test_kill_single(): + ... test_session = await server.anew_session("akill_single") + ... # Create a window + ... window = await test_session.anew_window(window_name='temp') + ... window_id = window.window_id + ... # Kill it + ... await window.akill() + ... # Verify it no longer exists + ... windows = test_session.windows + ... exists = any(w.window_id == window_id for w in windows) + ... await server.acmd("kill-session", target="akill_single") + ... return exists + >>> asyncio.run(test_kill_single()) + False + + Kill all windows except one: + + >>> import asyncio + >>> async def test_kill_all_except(): + ... test_session = await server.anew_session("akill_except") + ... # Create multiple windows + ... keep_window = await test_session.anew_window(window_name='main') + ... await test_session.anew_window(window_name='temp1') + ... await test_session.anew_window(window_name='temp2') + ... await test_session.anew_window(window_name='temp3') + ... # Kill all except keep_window (kills initial + temp windows) + ... await keep_window.akill(all_except=True) + ... # Count remaining windows (should be 1: only keep_window) + ... window_count = len(test_session.windows) + ... await server.acmd("kill-session", target="akill_except") + ... return window_count + >>> asyncio.run(test_kill_all_except()) + 1 + + Concurrent window cleanup: + + >>> import asyncio + >>> async def test_concurrent_cleanup(): + ... test_session = await server.anew_session("akill_concurrent") + ... # Create some temporary windows + ... temp_windows = await asyncio.gather( + ... test_session.anew_window(window_name='temp1'), + ... test_session.anew_window(window_name='temp2'), + ... test_session.anew_window(window_name='temp3'), + ... ) + ... # Kill all temporary windows concurrently + ... await asyncio.gather(*[w.akill() for w in temp_windows]) + ... # Count remaining windows (should be 1: initial window) + ... window_count = len(test_session.windows) + ... await server.acmd("kill-session", target="akill_concurrent") + ... return window_count + >>> asyncio.run(test_concurrent_cleanup()) + 1 + """ + flags: tuple[str, ...] = () + + if all_except: + flags += ("-a",) + + proc = await self.acmd( + "kill-window", + *flags, + ) + + if proc.stderr: + raise exc.LibTmuxException(proc.stderr) + """ Commands (tmux-like) """ @@ -597,6 +768,10 @@ def kill( ``$ tmux kill-window``. + See Also + -------- + :meth:`akill` : Async version of this method + Examples -------- Kill a window: diff --git a/tests/asyncio/README.md b/tests/asyncio/README.md new file mode 100644 index 000000000..2ad6d7c5d --- /dev/null +++ b/tests/asyncio/README.md @@ -0,0 +1,272 @@ +# Async Tests for libtmux + +This directory contains comprehensive async tests for libtmux's async API (`AsyncTmuxCmd` and `.acmd()` methods). + +## 📁 Test Organization + +Tests are organized by object type to mirror the sync test structure: + +``` +tests/asyncio/ +├── test_server.py - Server.acmd() and concurrent server operations +├── test_session.py - Session.acmd() and concurrent session operations +├── test_window.py - Window.acmd() and concurrent window operations +├── test_pane.py - Pane.acmd() and concurrent pane operations +└── test_integration.py - Complex multi-object async workflows +``` + +## 🔒 Test Safety + +**ALL tests use isolated test servers** that never affect developer tmux sessions: + +- Socket names: `libtmux_test{8_random_chars}` (e.g., `libtmux_testx7k4m9n2`) +- Unique per test via `server` fixture +- Automatic cleanup via `request.addfinalizer(server.kill)` +- No manual cleanup needed (relies on pytest fixture pattern) + +### Example: +```python +@pytest.mark.asyncio +async def test_my_async_feature(server: Server) -> None: + """Test description. + + Safety: All operations in isolated test server. + """ + result = await server.acmd("new-session", "-d", "-P", "-F#{session_id}") + # ... test logic ... + # No cleanup needed - fixture handles it! +``` + +## 🎯 Test Categories + +### 1. Basic `.acmd()` Tests +Tests for low-level async command execution: +- `test_server_acmd_basic` - Basic command execution +- `test_session_acmd_basic` - Session context +- `test_window_acmd_split_pane` - Window operations +- `test_pane_acmd_send_keys` - Pane operations + +### 2. Concurrent Operations +Tests showcasing async benefits (parallel execution): +- `test_concurrent_session_creation` - Create 3 sessions in parallel +- `test_concurrent_window_creation` - Create 4 windows concurrently +- `test_concurrent_pane_splits` - Create 2x2 pane grid efficiently +- `test_batch_session_operations` - Batch create and verify + +### 3. Real-World Automation +Tests demonstrating practical async use cases: +- `test_batch_pane_setup_automation` - Initialize dev environment (frontend/backend/database) +- `test_concurrent_send_keys_multiple_panes` - Execute commands across panes simultaneously +- `test_parallel_pane_monitoring` - Monitor logs from multiple services +- `test_multi_session_parallel_automation` - Set up multiple project environments + +### 4. Integration Workflows +Tests for complex multi-object scenarios: +- `test_async_full_workflow` - Complete workflow: session → window → pane → command +- `test_complex_pane_grid_automation` - Create 2x3 monitoring dashboard +- `test_multi_session_parallel_automation` - Automate multiple projects + +### 5. Error Handling & Edge Cases +Tests for robust error handling: +- `test_async_invalid_command` - Invalid command error capture +- `test_async_session_not_found` - Nonexistent session handling +- `test_concurrent_operations_with_partial_failure` - Handle partial failures gracefully +- `test_async_command_timeout_handling` - Timeout patterns with `asyncio.wait_for()` + +## 🚀 Running Tests + +```bash +# Run all async tests +pytest tests/asyncio/ -v + +# Run specific test file +pytest tests/asyncio/test_server.py -v + +# Run specific test +pytest tests/asyncio/test_server.py::test_concurrent_session_creation -v + +# Run with coverage +pytest tests/asyncio/ --cov=libtmux --cov-report=term-missing +``` + +## 📊 Test Statistics + +| File | Tests | Focus | +|------|-------|-------| +| test_server.py | 8 | Server operations, concurrency, error handling | +| test_session.py | 4 | Session operations, parallel window management | +| test_window.py | 3 | Window operations, concurrent pane creation | +| test_pane.py | 5 | Pane operations, real-world automation | +| test_integration.py | 5 | Complex workflows, error handling | +| **Total** | **25** | **Comprehensive async coverage** | + +## 💡 Key Patterns Demonstrated + +### Pattern 1: Concurrent Creation +```python +@pytest.mark.asyncio +async def test_concurrent_creation(server: Server) -> None: + """Create multiple objects concurrently.""" + async def create_session(name: str) -> Session: + result = await server.acmd("new-session", "-d", "-P", "-F#{session_id}", "-s", name) + return Session.from_session_id(result.stdout[0], server=server) + + # Create 3 sessions in parallel + sessions = await asyncio.gather( + create_session("session_1"), + create_session("session_2"), + create_session("session_3"), + ) +``` + +### Pattern 2: Parallel Queries +```python +@pytest.mark.asyncio +async def test_parallel_queries(server: Server) -> None: + """Query multiple objects concurrently.""" + async def get_info(session_id: str) -> dict: + result = await server.acmd("display-message", "-t", session_id, "-p", "#{session_name}") + return {"id": session_id, "name": result.stdout[0]} + + # Query all sessions in parallel + infos = await asyncio.gather(*[get_info(sid) for sid in session_ids]) +``` + +### Pattern 3: Batch Automation +```python +@pytest.mark.asyncio +async def test_batch_setup(session: Session) -> None: + """Set up multiple panes with commands.""" + configs = [ + {"cmd": "npm run dev", "name": "frontend"}, + {"cmd": "python manage.py runserver", "name": "backend"}, + {"cmd": "docker-compose up postgres", "name": "database"}, + ] + + async def setup_pane(pane_id: str, config: dict) -> bool: + pane = Pane.from_pane_id(pane_id, server=session.server) + await pane.acmd("send-keys", config["cmd"], "Enter") + return True + + # Set up all panes in parallel + await asyncio.gather(*[setup_pane(pid, cfg) for pid, cfg in zip(pane_ids, configs)]) +``` + +### Pattern 4: Error Handling +```python +@pytest.mark.asyncio +async def test_with_error_handling(server: Server) -> None: + """Handle errors in concurrent operations.""" + async def safe_create(name: str) -> tuple[str, bool]: + try: + result = await server.acmd("new-session", "-d", "-P", "-F#{session_id}", "-s", name) + return (name, result.returncode == 0) + except Exception: + return (name, False) + + # Some may fail, some succeed + results = await asyncio.gather(*[safe_create(name) for name in names]) + successes = [r for r in results if r[1]] + failures = [r for r in results if not r[1]] +``` + +### Pattern 5: Timeout Handling +```python +@pytest.mark.asyncio +async def test_with_timeout(server: Server) -> None: + """Use timeouts for async operations.""" + try: + result = await asyncio.wait_for( + server.acmd("new-session", "-d", "-P", "-F#{session_id}"), + timeout=5.0 + ) + except asyncio.TimeoutError: + # Handle timeout + pass +``` + +## 🔍 Why Async Matters for tmux + +Async provides **significant performance benefits** for tmux automation: + +### Sequential (Sync) - 3 seconds +```python +def setup_sync(server): + session1 = server.cmd("new-session", "-d") # 1s + session2 = server.cmd("new-session", "-d") # 1s + session3 = server.cmd("new-session", "-d") # 1s + # Total: 3 seconds +``` + +### Concurrent (Async) - 1 second +```python +async def setup_async(server): + sessions = await asyncio.gather( + server.acmd("new-session", "-d"), # ┐ + server.acmd("new-session", "-d"), # ├─ All run in parallel + server.acmd("new-session", "-d"), # ┘ + ) + # Total: 1 second +``` + +**3x faster** for this simple example. Real-world benefits increase with more operations! + +## 📚 Related Documentation + +- [Async API Documentation](../../docs/async_api.md) (if exists) +- [Pytest-asyncio Documentation](https://pytest-asyncio.readthedocs.io/) +- [Python asyncio Guide](https://docs.python.org/3/library/asyncio.html) + +## 🤝 Contributing + +When adding new async tests: + +1. **Use the `server` or `session` fixture** (already isolated and safe) +2. **Decorate with `@pytest.mark.asyncio`** +3. **Add docstring with safety note**: `Safety: All operations in isolated test server.` +4. **Follow existing patterns**: Look at similar tests for examples +5. **No manual cleanup needed**: Fixtures handle it via finalizers + +### Example Template: +```python +@pytest.mark.asyncio +async def test_your_feature(server: Server) -> None: + """Test description. + + Safety: All operations in isolated test server. + Demonstrates: [what pattern this test shows] + """ + # Your test code + result = await server.acmd(...) + assert result.returncode == 0 +``` + +## 🐛 Debugging Tips + +### Test Failures +```bash +# Run with verbose output +pytest tests/asyncio/test_server.py -vv + +# Run with print statements visible +pytest tests/asyncio/test_server.py -s + +# Run with debug on failure +pytest tests/asyncio/test_server.py --pdb +``` + +### Timing Issues +If tests are flaky due to timing: +- Increase `await asyncio.sleep()` duration +- Add explicit waits after `send-keys` before `capture-pane` +- Check if panes have finished executing commands + +### Isolation Issues +If tests affect each other: +- Verify using `server` fixture (not creating custom servers) +- Check socket names are unique (`libtmux_test{random}`) +- Ensure no global state between tests + +--- + +**Questions?** Check existing tests for examples or refer to the main libtmux documentation. diff --git a/tests/asyncio/__init__.py b/tests/asyncio/__init__.py new file mode 100644 index 000000000..fd76e7a9b --- /dev/null +++ b/tests/asyncio/__init__.py @@ -0,0 +1,22 @@ +"""Async tests for libtmux. + +This directory contains asynchronous tests for libtmux's async API. Tests are +organized by object type to mirror the sync test structure: + +- test_server.py: Server.acmd() and concurrent server operations +- test_session.py: Session.acmd() and concurrent session operations +- test_window.py: Window.acmd() and concurrent window operations +- test_pane.py: Pane.acmd() and concurrent pane operations +- test_integration.py: Complex multi-object async workflows + +All tests use isolated test servers via the `server` fixture with unique socket +names (libtmux_test{8_random_chars}) that never affect developer sessions. + +Key patterns demonstrated: +- Concurrent operations (parallel session/window/pane creation) +- Real-world automation (batch operations, multi-pane setup) +- Error handling (timeouts, command failures, race conditions) +- Integration workflows (complex multi-object scenarios) +""" + +from __future__ import annotations diff --git a/tests/asyncio/test_integration.py b/tests/asyncio/test_integration.py new file mode 100644 index 000000000..4cca9ac13 --- /dev/null +++ b/tests/asyncio/test_integration.py @@ -0,0 +1,488 @@ +"""Integration tests for complex async workflows. + +SAFETY: All tests use isolated test servers via fixtures. +Socket names: libtmux_test{8_random_chars} - never affects developer sessions. +""" + +from __future__ import annotations + +import asyncio +import logging +from dataclasses import dataclass +from typing import TYPE_CHECKING + +import pytest + +from libtmux.pane import Pane +from libtmux.session import Session +from libtmux.window import Window + + +@dataclass(slots=True) +class ProjectSessionResult: + """Summary of concurrently created project sessions.""" + + session_id: str + name: str + window_count: int + + +if TYPE_CHECKING: + from libtmux.server import Server + +logger = logging.getLogger(__name__) + + +# ============================================================================ +# Integration Tests +# ============================================================================ + + +@pytest.mark.asyncio +async def test_async_full_workflow(server: Server) -> None: + """Test complete async workflow: session -> window -> pane -> command. + + Safety: All objects created in isolated test server. + Demonstrates comprehensive async tmux manipulation. + Cleanup: Server fixture finalizer handles all resource destruction. + """ + # Create session + result = await server.acmd("new-session", "-d", "-P", "-F#{session_id}") + session_id = result.stdout[0] + session = Session.from_session_id(session_id=session_id, server=server) + + # Verify session created + assert session_id.startswith("$") + assert server.has_session(session_id) + + # Create window in session + result = await session.acmd("new-window", "-P", "-F#{window_id}") + window_id = result.stdout[0] + window = Window.from_window_id(window_id=window_id, server=server) + assert window_id.startswith("@") + + # Split pane in window + result = await window.acmd("split-window", "-P", "-F#{pane_id}") + pane_id = result.stdout[0] + pane = Pane.from_pane_id(pane_id=pane_id, server=server) + assert pane_id.startswith("%") + + # Send command to pane + await pane.acmd("send-keys", "echo 'integration_test_complete'", "Enter") + await asyncio.sleep(0.2) + + # Verify output + result = await pane.acmd("capture-pane", "-p") + assert any("integration_test_complete" in line for line in result.stdout) + + # Verify complete object hierarchy + session_obj_id = session.session_id + assert session_obj_id is not None + assert session_obj_id == session_id + window_obj_id = window.window_id + assert window_obj_id is not None + assert window_obj_id == window_id + pane_obj_id = pane.pane_id + assert pane_obj_id is not None + assert pane_obj_id == pane_id + + +@pytest.mark.asyncio +async def test_multi_session_parallel_automation(server: Server) -> None: + """Test automating multiple sessions concurrently. + + Safety: All sessions created in isolated test server. + Real-world pattern: Set up multiple project environments simultaneously. + """ + + async def setup_project_session( + name: str, num_windows: int + ) -> ProjectSessionResult: + """Create session with multiple windows.""" + # Create session + result = await server.acmd( + "new-session", + "-d", + "-P", + "-F#{session_id}", + "-s", + name, + ) + session_id = result.stdout[0] + session = Session.from_session_id(session_id=session_id, server=server) + + # Create additional windows concurrently + window_tasks = [ + session.acmd("new-window", "-P", "-F#{window_id}", "-n", f"win_{i}") + for i in range(num_windows - 1) # -1 because session starts with 1 window + ] + await asyncio.gather(*window_tasks) + + # Verify setup + result = await session.acmd("list-windows", "-F#{window_id}") + window_count = len(result.stdout) + + return ProjectSessionResult( + session_id=session_id, + name=name, + window_count=window_count, + ) + + # Set up 3 project sessions concurrently + results_tuple = await asyncio.gather( + setup_project_session("project_frontend", 3), + setup_project_session("project_backend", 4), + setup_project_session("project_infra", 2), + ) + results: list[ProjectSessionResult] = list(results_tuple) + + # Verify all sessions set up correctly + assert len(results) == 3 + assert results[0].window_count == 3 + assert results[1].window_count == 4 + assert results[2].window_count == 2 + + # Verify all sessions exist + for result in results: + assert server.has_session(result.name) + + +@pytest.mark.asyncio +async def test_complex_pane_grid_automation(server: Server) -> None: + """Test creating and configuring a complex pane grid. + + Safety: All operations in isolated test server. + Real-world pattern: Dashboard layout with multiple monitoring panes. + """ + # Create session + result = await server.acmd("new-session", "-d", "-P", "-F#{session_id}") + session_id = result.stdout[0] + session = Session.from_session_id(session_id=session_id, server=server) + + window = session.active_window + assert window is not None + + # Create a 2x3 grid of panes + # Split into 2 columns + await window.acmd("split-window", "-h") + + # Split each column into 3 rows concurrently + await asyncio.gather( + window.acmd("split-window", "-v"), + window.acmd("split-window", "-v"), + window.acmd("split-window", "-v", "-t", "{right}"), + window.acmd("split-window", "-v", "-t", "{right}"), + ) + + # Get all pane IDs + result = await window.acmd("list-panes", "-F#{pane_id}") + pane_ids = result.stdout + assert len(pane_ids) == 6 # 2x3 grid + + # Configure each pane with a different "monitoring" command concurrently + monitoring_commands = [ + "echo 'CPU Monitor'", + "echo 'Memory Monitor'", + "echo 'Disk Monitor'", + "echo 'Network Monitor'", + "echo 'Process Monitor'", + "echo 'Log Monitor'", + ] + + async def configure_pane(pane_id: str, command: str) -> str: + """Send command to pane.""" + pane = Pane.from_pane_id(pane_id=pane_id, server=server) + await pane.acmd("send-keys", command, "Enter") + return pane_id + + # Configure all panes concurrently + await asyncio.gather( + *[ + configure_pane(pid, cmd) + for pid, cmd in zip(pane_ids, monitoring_commands, strict=False) + ] + ) + + await asyncio.sleep(0.3) + + # Verify all panes configured + expected_texts = ["CPU", "Memory", "Disk", "Network", "Process", "Log"] + for pane_id, expected in zip(pane_ids, expected_texts, strict=False): + pane = Pane.from_pane_id(pane_id=pane_id, server=server) + result = await pane.acmd("capture-pane", "-p") + assert any(expected in line for line in result.stdout), f"{expected} not found" + + +# ============================================================================ +# Error Handling & Edge Cases +# ============================================================================ + + +@pytest.mark.asyncio +async def test_concurrent_operations_with_partial_failure(server: Server) -> None: + """Test handling partial failures in concurrent operations. + + Safety: All operations in isolated test server. + Demonstrates error handling: some operations succeed, some fail. + """ + + async def create_session_safe(name: str) -> tuple[str, bool, str]: + """Create session and return status.""" + try: + result = await server.acmd( + "new-session", + "-d", + "-P", + "-F#{session_id}", + "-s", + name, + ) + except Exception as e: + return (name, False, str(e)) + else: + session_id = result.stdout[0] if result.stdout else "" + success = result.returncode == 0 + return (name, success, session_id) + + # Create sessions, including one duplicate (will fail) + results = await asyncio.gather( + create_session_safe("valid_session_1"), + create_session_safe("valid_session_2"), + create_session_safe("valid_session_1"), # Duplicate - should fail + create_session_safe("valid_session_3"), + ) + + # Verify we got 4 results + assert len(results) == 4 + + # Check successes and failures + successes = [r for r in results if r[1]] + failures = [r for r in results if not r[1]] + + # Should have 3 successes and 1 failure (duplicate) + assert len(successes) == 3 + assert len(failures) == 1 + assert failures[0][0] == "valid_session_1" # Duplicate name + + +@pytest.mark.asyncio +async def test_async_command_timeout_handling(server: Server) -> None: + """Test handling slow/hanging commands. + + Safety: All operations in isolated test server. + Demonstrates: async timeout patterns for command execution. + """ + + async def create_session_with_timeout( + name: str, timeout: float + ) -> tuple[str, bool]: + """Create session with timeout.""" + try: + await asyncio.wait_for( + server.acmd("new-session", "-d", "-P", "-F#{session_id}", "-s", name), + timeout=timeout, + ) + except asyncio.TimeoutError: + return (name, False) + else: + return (name, True) + + # Create sessions with generous timeout (should all succeed) + results = await asyncio.gather( + create_session_with_timeout("session_1", 5.0), + create_session_with_timeout("session_2", 5.0), + create_session_with_timeout("session_3", 5.0), + ) + + # All should succeed with reasonable timeout + assert len(results) == 3 + assert all(success for _, success in results) + + +# ============================================================================ +# Async Pane Method Integration Tests +# ============================================================================ + + +@pytest.mark.asyncio +async def test_async_pane_workflow_complete(server: Server) -> None: + """Test complete pane lifecycle with new async methods. + + Safety: All operations in isolated test server. + Demonstrates: Full async workflow using asend_keys, acapture_pane, asplit. + Pattern: Create -> send -> capture -> split -> concurrent ops -> cleanup. + """ + # Create session + session = await server.anew_session("pane_workflow_test") + + # Get active pane + pane1 = session.active_pane + assert pane1 is not None + pane1_id = pane1.pane_id + assert pane1_id is not None + + # Send command using asend_keys + await pane1.asend_keys('echo "workflow_step_1"') + await asyncio.sleep(0.2) + + # Capture output using acapture_pane + output1 = await pane1.acapture_pane() + assert any("workflow_step_1" in line for line in output1) + + # Split pane using asplit + pane2 = await pane1.asplit() + assert pane2 is not None + assert pane2.pane_id is not None + assert pane2.pane_id != pane1_id + + # Verify both panes exist + window = session.active_window + assert window is not None + assert len(window.panes) == 2 + + # Send different commands to each pane concurrently + await asyncio.gather( + pane1.asend_keys('echo "pane1_data"'), + pane2.asend_keys('echo "pane2_data"'), + ) + await asyncio.sleep(0.3) + + # Capture outputs concurrently + outputs = await asyncio.gather( + pane1.acapture_pane(), + pane2.acapture_pane(), + ) + + # Verify both outputs + assert any("pane1_data" in line for line in outputs[0]) + assert any("pane2_data" in line for line in outputs[1]) + + +@pytest.mark.asyncio +async def test_multi_window_pane_automation(server: Server) -> None: + """Test complex multi-window, multi-pane async automation. + + Safety: All operations in isolated test server. + Demonstrates: Large-scale concurrent pane manipulation. + Pattern: 3 windows x 3 panes = 9 panes, all managed concurrently. + """ + # Create session + session = await server.anew_session("multi_window_automation") + + # Create 3 windows concurrently + windows_data = await asyncio.gather( + session.anew_window(window_name="window1"), + session.anew_window(window_name="window2"), + session.anew_window(window_name="window3"), + ) + + # Each window should have 1 pane initially + all_panes: list[Pane] = [] + + # For each window, split into 3 panes total + for _idx, window in enumerate(windows_data): + base_pane = window.active_pane + assert base_pane is not None + + # Create 2 more panes (total 3 per window) + from libtmux.pane import PaneDirection + + new_panes = await asyncio.gather( + base_pane.asplit(direction=PaneDirection.Right), + base_pane.asplit(direction=PaneDirection.Below), + ) + + # Collect all 3 panes from this window + all_panes.extend([base_pane, *new_panes]) + + # Verify we have 9 panes total (3 windows x 3 panes) + assert len(all_panes) == 9 + + # Send unique commands to all 9 panes concurrently + send_tasks = [ + pane.asend_keys(f'echo "pane_{i}_output"') for i, pane in enumerate(all_panes) + ] + await asyncio.gather(*send_tasks) + await asyncio.sleep(0.4) + + # Capture output from all 9 panes concurrently + outputs = await asyncio.gather(*[pane.acapture_pane() for pane in all_panes]) + + # Verify all panes have correct output + assert len(outputs) == 9 + for i, output in enumerate(outputs): + assert any(f"pane_{i}_output" in line for line in output) + + +@pytest.mark.asyncio +async def test_pane_monitoring_dashboard(server: Server) -> None: + """Test monitoring dashboard pattern with async pane methods. + + Safety: All operations in isolated test server. + Demonstrates: Real-world monitoring use case with periodic capture. + Pattern: 2x3 grid of panes, periodic concurrent monitoring. + """ + # Create session + session = await server.anew_session("monitoring_dashboard") + window = session.active_window + assert window is not None + + # Create 2x3 grid (6 panes total) + # Start with 1 pane, split to make 6 + base_pane = window.active_pane + assert base_pane is not None + + from libtmux.pane import PaneDirection + + # Create top row (3 panes) + pane2 = await base_pane.asplit(direction=PaneDirection.Right) + pane3 = await base_pane.asplit(direction=PaneDirection.Right) + + # Create bottom row (3 more panes) + pane4 = await base_pane.asplit(direction=PaneDirection.Below) + pane5 = await pane2.asplit(direction=PaneDirection.Below) + pane6 = await pane3.asplit(direction=PaneDirection.Below) + + all_panes = [base_pane, pane2, pane3, pane4, pane5, pane6] + + # Verify grid created + assert len(window.panes) == 6 + + # Send "monitoring" commands to each pane + monitor_commands = [ + 'echo "CPU: 45%"', + 'echo "Memory: 60%"', + 'echo "Disk: 30%"', + 'echo "Network: 100Mbps"', + 'echo "Processes: 150"', + 'echo "Uptime: 5 days"', + ] + + await asyncio.gather( + *[ + pane.asend_keys(cmd) + for pane, cmd in zip(all_panes, monitor_commands, strict=False) + ] + ) + await asyncio.sleep(0.3) + + # Periodically capture all panes (simulate 3 monitoring rounds) + for round_num in range(3): + # Capture all panes concurrently + outputs = await asyncio.gather(*[pane.acapture_pane() for pane in all_panes]) + + # Verify all panes have output + assert len(outputs) == 6 + + # Verify specific monitoring data appears + assert any("CPU:" in line for line in outputs[0]) + assert any("Memory:" in line for line in outputs[1]) + assert any("Disk:" in line for line in outputs[2]) + + # Wait before next monitoring round + if round_num < 2: + await asyncio.sleep(0.2) + + # Verify dashboard functional after 3 rounds + final_outputs = await asyncio.gather(*[pane.acapture_pane() for pane in all_panes]) + assert all(len(output) > 0 for output in final_outputs) diff --git a/tests/asyncio/test_pane.py b/tests/asyncio/test_pane.py new file mode 100644 index 000000000..c755a492a --- /dev/null +++ b/tests/asyncio/test_pane.py @@ -0,0 +1,690 @@ +"""Tests for Pane async operations. + +SAFETY: All tests use isolated test servers via fixtures. +Socket names: libtmux_test{8_random_chars} - never affects developer sessions. +""" + +from __future__ import annotations + +import asyncio +import logging +from dataclasses import dataclass +from typing import TYPE_CHECKING + +import pytest + +if TYPE_CHECKING: + from libtmux.session import Session + + +@dataclass(slots=True) +class PaneSetupConfig: + """Pane command and expected output snippet for setup tests.""" + + cmd: str + check: str + + +@dataclass(slots=True) +class PaneSetupResult: + """Outcome of pane setup verification.""" + + pane_id: str + command: str + success: bool + + +@dataclass(slots=True) +class PaneMonitorResult: + """Result of monitoring command execution per pane.""" + + pane_id: str + service: str + running: bool + + +logger = logging.getLogger(__name__) + + +# ============================================================================ +# Pane.acmd() Tests +# ============================================================================ + + +@pytest.mark.asyncio +async def test_pane_acmd_basic(session: Session) -> None: + """Test Pane.acmd() executes in pane context. + + Safety: Commands sent to isolated test pane only. + """ + pane = session.active_pane + assert pane is not None + pane_id = pane.pane_id + assert pane_id is not None + + # Display pane ID + result = await pane.acmd("display-message", "-p", "#{pane_id}") + assert result.stdout[0] == pane_id + + +@pytest.mark.asyncio +async def test_pane_acmd_send_keys(session: Session) -> None: + """Test sending keys via Pane.acmd(). + + Safety: Keys sent to isolated test pane only. + """ + pane = session.active_pane + assert pane is not None + + # Send echo command + await pane.acmd("send-keys", "echo 'test_async_pane'", "Enter") + + # Give command time to execute + await asyncio.sleep(0.2) + + # Capture output + result = await pane.acmd("capture-pane", "-p") + assert any("test_async_pane" in line for line in result.stdout) + + +# ============================================================================ +# Real-World Automation Tests +# ============================================================================ + + +@pytest.mark.asyncio +async def test_concurrent_send_keys_multiple_panes(session: Session) -> None: + """Test sending commands to multiple panes concurrently. + + Safety: All panes in isolated test session. + Real-world pattern: Execute commands across multiple panes simultaneously. + """ + from libtmux.pane import Pane + + window = session.active_window + assert window is not None + + # Create 3 panes + result1 = await window.acmd("split-window", "-h", "-P", "-F#{pane_id}") + result2 = await window.acmd("split-window", "-v", "-P", "-F#{pane_id}") + + active_pane = session.active_pane + assert active_pane is not None + active_pane_id = active_pane.pane_id + assert active_pane_id is not None + + pane_ids: list[str] = [ + active_pane_id, + result1.stdout[0], + result2.stdout[0], + ] + + async def send_command(pane_id: str, command: str) -> str: + """Send command to pane and return pane ID.""" + pane = Pane.from_pane_id(pane_id=pane_id, server=session.server) + await pane.acmd("send-keys", command, "Enter") + return pane_id + + # Send different commands to all panes concurrently + await asyncio.gather( + send_command(pane_ids[0], "echo 'pane_0_output'"), + send_command(pane_ids[1], "echo 'pane_1_output'"), + send_command(pane_ids[2], "echo 'pane_2_output'"), + ) + + # Wait for commands to execute + await asyncio.sleep(0.3) + + # Verify outputs from each pane + async def check_output(pane_id: str, expected: str) -> bool: + """Check if pane output contains expected string.""" + pane = Pane.from_pane_id(pane_id=pane_id, server=session.server) + result = await pane.acmd("capture-pane", "-p") + return any(expected in line for line in result.stdout) + + # Check all panes concurrently + results = await asyncio.gather( + check_output(pane_ids[0], "pane_0_output"), + check_output(pane_ids[1], "pane_1_output"), + check_output(pane_ids[2], "pane_2_output"), + ) + + assert all(results), "Not all panes executed commands successfully" + + +@pytest.mark.asyncio +async def test_batch_pane_setup_automation(session: Session) -> None: + """Test setting up multiple panes with different commands. + + Safety: All operations in isolated test session. + Real-world pattern: Initialize development environment with multiple services. + """ + from libtmux.pane import Pane + + window = session.active_window + assert window is not None + + # Define pane setup: command and check string + pane_configs: list[PaneSetupConfig] = [ + PaneSetupConfig(cmd="echo 'Frontend: localhost:3000'", check="Frontend"), + PaneSetupConfig(cmd="echo 'Backend: localhost:8000'", check="Backend"), + PaneSetupConfig(cmd="echo 'Database: localhost:5432'", check="Database"), + ] + + # Create panes + active_pane = session.active_pane + assert active_pane is not None + first_pane_id = active_pane.pane_id + assert first_pane_id is not None + + pane_ids: list[str] = [first_pane_id] + for _ in range(2): + split_result = await window.acmd("split-window", "-h", "-P", "-F#{pane_id}") + pane_ids.append(split_result.stdout[0]) + + async def setup_pane(pane_id: str, config: PaneSetupConfig) -> PaneSetupResult: + """Set up a pane with command and verify output.""" + pane = Pane.from_pane_id(pane_id=pane_id, server=session.server) + + # Send command + await pane.acmd("send-keys", config.cmd, "Enter") + await asyncio.sleep(0.2) + + # Capture and verify + result = await pane.acmd("capture-pane", "-p") + success = any(config.check in line for line in result.stdout) + + return PaneSetupResult(pane_id=pane_id, command=config.cmd, success=success) + + # Set up all panes concurrently + results: list[PaneSetupResult] = await asyncio.gather( + *[ + setup_pane(pid, config) + for pid, config in zip(pane_ids, pane_configs, strict=False) + ] + ) + + # Verify all setups succeeded + assert len(results) == 3 + for pane_result in results: + assert pane_result.success, f"Pane {pane_result.pane_id} setup failed" + + +@pytest.mark.asyncio +async def test_parallel_pane_monitoring(session: Session) -> None: + """Test monitoring output from multiple panes concurrently. + + Safety: All panes in isolated test session. + Real-world pattern: Monitor logs from multiple services simultaneously. + """ + from libtmux.pane import Pane + + window = session.active_window + assert window is not None + + # Create 3 panes (original + 2 splits) + result1 = await window.acmd("split-window", "-h", "-P", "-F#{pane_id}") + result2 = await window.acmd("split-window", "-v", "-P", "-F#{pane_id}") + + active_pane = session.active_pane + assert active_pane is not None + active_pane_id = active_pane.pane_id + assert active_pane_id is not None + + pane_ids: list[str] = [ + active_pane_id, + result1.stdout[0], + result2.stdout[0], + ] + + async def send_and_verify(pane_id: str, service_num: int) -> PaneMonitorResult: + """Send command to pane and verify output.""" + pane = Pane.from_pane_id(pane_id=pane_id, server=session.server) + + # Send command + await pane.acmd("send-keys", f"echo 'service_{service_num}_running'", "Enter") + await asyncio.sleep(0.3) + + # Capture and verify + result = await pane.acmd("capture-pane", "-p") + found = any(f"service_{service_num}_running" in line for line in result.stdout) + + return PaneMonitorResult( + pane_id=pane_id, + service=f"service_{service_num}", + running=found, + ) + + # Send commands and monitor all panes concurrently + monitor_results: list[PaneMonitorResult] = await asyncio.gather( + *[send_and_verify(pid, i) for i, pid in enumerate(pane_ids)] + ) + + # Verify all services detected + assert len(monitor_results) == 3 + for monitor_result in monitor_results: + assert monitor_result.running, f"Service {monitor_result.service} not detected" + + +# ============================================================================ +# Pane.asend_keys() Tests +# ============================================================================ + + +@pytest.mark.asyncio +async def test_asend_keys_basic_execution(session: Session) -> None: + """Test Pane.asend_keys() basic command execution with enter. + + Safety: Commands sent to isolated test pane only. + """ + pane = session.active_pane + assert pane is not None + + # Send command with enter + await pane.asend_keys('echo "test_asend_basic"', enter=True) + + # Wait for command to execute + await asyncio.sleep(0.2) + + # Verify output + output = pane.capture_pane() + assert any("test_asend_basic" in line for line in output) + + +@pytest.mark.asyncio +async def test_asend_keys_without_enter(session: Session) -> None: + """Test Pane.asend_keys() without enter - command visible but not executed. + + Safety: Commands sent to isolated test pane only. + """ + pane = session.active_pane + assert pane is not None + + # Send command without enter + await pane.asend_keys('echo "should_not_execute"', enter=False) + + # Wait briefly + await asyncio.sleep(0.1) + + # Verify command text is visible but not executed + output = pane.capture_pane() + # Command should be visible in the pane + assert any("echo" in line for line in output) + # But output should NOT appear (command not executed) + # Note: We can't test for absence of output directly as the prompt might vary + + +@pytest.mark.asyncio +async def test_asend_keys_literal_mode(session: Session) -> None: + """Test Pane.asend_keys() literal mode - special chars sent as text. + + Safety: Commands sent to isolated test pane only. + """ + pane = session.active_pane + assert pane is not None + + # Send literal special character (not a signal) + await pane.asend_keys("C-c", literal=True, enter=False) + + # Wait briefly + await asyncio.sleep(0.1) + + # Verify literal text "C-c" appears (not an interrupt signal) + output = pane.capture_pane() + assert any("C-c" in line for line in output) + + +@pytest.mark.asyncio +async def test_asend_keys_suppress_history(session: Session) -> None: + """Test Pane.asend_keys() with suppress_history prepends space. + + Safety: Commands sent to isolated test pane only. + """ + pane = session.active_pane + assert pane is not None + + # Send command with history suppression + await pane.asend_keys('echo "secret_command"', suppress_history=True, enter=True) + + # Wait for execution + await asyncio.sleep(0.2) + + # Verify output appears (command executed) + output = pane.capture_pane() + assert any("secret_command" in line for line in output) + # Note: Full history verification would require shell-specific setup + + +@pytest.mark.asyncio +async def test_asend_keys_concurrent_multiple_panes(session: Session) -> None: + """Test sending keys to multiple panes concurrently via asend_keys(). + + Safety: All panes in isolated test session. + Real-world pattern: Execute commands across multiple panes simultaneously. + """ + window = session.active_window + assert window is not None + + # Create 3 panes + pane1 = window.active_pane + assert pane1 is not None + pane2 = window.split() + pane3 = window.split() + + # Send different commands to all panes concurrently + await asyncio.gather( + pane1.asend_keys('echo "pane1_output"'), + pane2.asend_keys('echo "pane2_output"'), + pane3.asend_keys('echo "pane3_output"'), + ) + + # Wait for commands to execute + await asyncio.sleep(0.3) + + # Verify each pane has correct output + output1 = pane1.capture_pane() + output2 = pane2.capture_pane() + output3 = pane3.capture_pane() + + assert any("pane1_output" in line for line in output1) + assert any("pane2_output" in line for line in output2) + assert any("pane3_output" in line for line in output3) + + +# ============================================================================ +# Pane.acapture_pane() Tests +# ============================================================================ + + +@pytest.mark.asyncio +async def test_acapture_pane_basic(session: Session) -> None: + """Test Pane.acapture_pane() basic output capture. + + Safety: Capture from isolated test pane only. + """ + pane = session.active_pane + assert pane is not None + + # Send command + await pane.asend_keys('echo "capture_test_output"') + await asyncio.sleep(0.2) + + # Capture output + output = await pane.acapture_pane() + + # Verify output + assert isinstance(output, list) + assert any("capture_test_output" in line for line in output) + + +@pytest.mark.asyncio +async def test_acapture_pane_with_start_parameter(session: Session) -> None: + """Test Pane.acapture_pane() with start parameter to capture history. + + Safety: Capture from isolated test pane only. + """ + pane = session.active_pane + assert pane is not None + + # Send multiple commands to build history + await pane.asend_keys('echo "line1"') + await asyncio.sleep(0.1) + await pane.asend_keys('echo "line2"') + await asyncio.sleep(0.1) + await pane.asend_keys('echo "line3"') + await asyncio.sleep(0.2) + + # Capture with start parameter (last 10 lines including history) + output = await pane.acapture_pane(start=-10) + + # Verify output includes history + assert isinstance(output, list) + assert len(output) > 0 + + +@pytest.mark.asyncio +async def test_acapture_pane_with_end_parameter(session: Session) -> None: + """Test Pane.acapture_pane() with end parameter to limit output. + + Safety: Capture from isolated test pane only. + """ + pane = session.active_pane + assert pane is not None + + # Send commands + await pane.asend_keys('echo "test_line"') + await asyncio.sleep(0.2) + + # Capture with end parameter (first 5 lines) + output = await pane.acapture_pane(end=5) + + # Verify output is limited + assert isinstance(output, list) + assert len(output) <= 6 # end=5 means lines 0-5 inclusive + + +@pytest.mark.asyncio +async def test_acapture_pane_full_history(session: Session) -> None: + """Test Pane.acapture_pane() capturing complete scrollback history. + + Safety: Capture from isolated test pane only. + """ + pane = session.active_pane + assert pane is not None + + # Send multiple commands + for i in range(5): + await pane.asend_keys(f'echo "history_line_{i}"') + await asyncio.sleep(0.1) + + # Capture full history (from start to end) + output = await pane.acapture_pane(start="-", end="-") + + # Verify we got output + assert isinstance(output, list) + assert len(output) > 0 + + +@pytest.mark.asyncio +async def test_acapture_pane_concurrent_multiple_panes(session: Session) -> None: + """Test capturing from multiple panes concurrently via acapture_pane(). + + Safety: All panes in isolated test session. + Real-world pattern: Monitor outputs from multiple panes simultaneously. + """ + window = session.active_window + assert window is not None + + # Create 3 panes + pane1 = window.active_pane + assert pane1 is not None + pane2 = window.split() + pane3 = window.split() + + # Send different commands to each pane + await asyncio.gather( + pane1.asend_keys('echo "capture1"'), + pane2.asend_keys('echo "capture2"'), + pane3.asend_keys('echo "capture3"'), + ) + await asyncio.sleep(0.3) + + # Capture output from all panes concurrently + outputs = await asyncio.gather( + pane1.acapture_pane(), + pane2.acapture_pane(), + pane3.acapture_pane(), + ) + + # Verify all outputs + assert len(outputs) == 3 + assert any("capture1" in line for line in outputs[0]) + assert any("capture2" in line for line in outputs[1]) + assert any("capture3" in line for line in outputs[2]) + + +# ============================================================================ +# Pane.asplit() Tests +# ============================================================================ + + +@pytest.mark.asyncio +async def test_asplit_default_below(session: Session) -> None: + """Test Pane.asplit() default split direction (below). + + Safety: Pane split in isolated test session. + """ + window = session.active_window + assert window is not None + pane = window.active_pane + assert pane is not None + original_pane_id = pane.pane_id + assert original_pane_id is not None + + initial_pane_count = len(window.panes) + + # Split pane (default is below) + new_pane = await pane.asplit() + + # Verify new pane created + assert len(window.panes) == initial_pane_count + 1 + assert new_pane is not None + assert new_pane.pane_id != original_pane_id + + +@pytest.mark.asyncio +async def test_asplit_direction_right(session: Session) -> None: + """Test Pane.asplit() vertical split to the right. + + Safety: Pane split in isolated test session. + """ + from libtmux.pane import PaneDirection + + window = session.active_window + assert window is not None + pane = window.active_pane + assert pane is not None + source_pane_id = pane.pane_id + assert source_pane_id is not None + + initial_pane_count = len(window.panes) + + # Split pane to the right + new_pane = await pane.asplit(direction=PaneDirection.Right) + + # Verify new pane created + assert len(window.panes) == initial_pane_count + 1 + assert new_pane is not None + assert new_pane.pane_id != source_pane_id + + +@pytest.mark.asyncio +async def test_asplit_with_start_directory(session: Session) -> None: + """Test Pane.asplit() with custom start directory. + + Safety: Pane split in isolated test session. + """ + window = session.active_window + assert window is not None + pane = window.active_pane + assert pane is not None + + # Split with custom directory + new_pane = await pane.asplit(start_directory="/tmp") + + # Verify pane created + assert new_pane is not None + + # Send pwd command to verify directory + await new_pane.asend_keys("pwd") + await asyncio.sleep(0.3) + + # Check output + output = new_pane.capture_pane() + # Verify /tmp appears in output (pwd result) + has_tmp = any("/tmp" in line for line in output) + assert has_tmp, f"Expected /tmp in output, got: {output}" + + +@pytest.mark.asyncio +async def test_asplit_with_size(session: Session) -> None: + """Test Pane.asplit() with size parameter. + + Safety: Pane split in isolated test session. + """ + window = session.active_window + assert window is not None + pane = window.active_pane + assert pane is not None + + initial_pane_count = len(window.panes) + + # Split with size (30%) + new_pane = await pane.asplit(size="30%") + + # Verify pane created + assert len(window.panes) == initial_pane_count + 1 + assert new_pane is not None + # Note: Actual size verification would require dimension checks + + +@pytest.mark.asyncio +async def test_asplit_with_shell_command(session: Session) -> None: + """Test Pane.asplit() with shell command (auto-closes after execution). + + Safety: Pane split in isolated test session. + Note: Pane auto-closes when command completes, which is expected behavior. + """ + window = session.active_window + assert window is not None + pane = window.active_pane + assert pane is not None + + initial_pane_count = len(window.panes) + + # Split with shell command that runs longer before exiting + # Use sleep to keep pane alive briefly + new_pane = await pane.asplit(shell='sleep 0.3 && echo "done"') + + # Verify pane was created + assert new_pane is not None + assert new_pane.pane_id is not None + + # Verify pane exists initially (before command finishes) + immediate_pane_count = len(window.panes) + assert immediate_pane_count == initial_pane_count + 1 + + # Wait for command to complete and pane to auto-close + await asyncio.sleep(0.6) + + # Verify pane count reduced (pane auto-closed) + final_pane_count = len(window.panes) + assert final_pane_count == initial_pane_count + + +@pytest.mark.asyncio +async def test_asplit_concurrent_multiple_splits(session: Session) -> None: + """Test creating multiple panes concurrently via asplit(). + + Safety: All panes in isolated test session. + Real-world pattern: Rapidly create complex pane layouts. + """ + window = session.active_window + assert window is not None + base_pane = window.active_pane + assert base_pane is not None + + initial_pane_count = len(window.panes) + + # Create multiple panes concurrently + from libtmux.pane import PaneDirection + + new_panes = await asyncio.gather( + base_pane.asplit(direction=PaneDirection.Right), + base_pane.asplit(direction=PaneDirection.Below), + ) + + # Verify panes created + assert len(new_panes) == 2 + assert all(p is not None for p in new_panes) + assert len(window.panes) >= initial_pane_count + 2 diff --git a/tests/asyncio/test_server.py b/tests/asyncio/test_server.py new file mode 100644 index 000000000..f5c7acf0f --- /dev/null +++ b/tests/asyncio/test_server.py @@ -0,0 +1,398 @@ +"""Tests for Server async operations. + +SAFETY: All tests use isolated test servers via fixtures. +Socket names: libtmux_test{8_random_chars} - never affects developer sessions. +""" + +from __future__ import annotations + +import asyncio +import logging +from dataclasses import dataclass +from typing import TYPE_CHECKING + +import pytest + +from libtmux.session import Session + + +@dataclass(slots=True) +class SessionQueryInfo: + """Structured data returned from async session queries.""" + + id: str + name: str + windows: int + + +if TYPE_CHECKING: + from libtmux.server import Server + +logger = logging.getLogger(__name__) + + +# ============================================================================ +# Server.acmd() Basic Tests +# ============================================================================ + + +@pytest.mark.asyncio +async def test_server_acmd_basic(server: Server) -> None: + """Test Server.acmd() basic async command execution. + + Safety: Uses isolated test server from `server` fixture. + Server socket: libtmux_test{random} - isolated from developer sessions. + """ + # Test basic command execution + result = await server.acmd("list-sessions") + # returncode may be 0 or 1 depending on whether sessions exist + # The important thing is the command executes asynchronously + assert result.returncode in {0, 1} + assert isinstance(result.stdout, list) + assert isinstance(result.stderr, list) + + +@pytest.mark.asyncio +async def test_server_acmd_new_session(server: Server) -> None: + """Test creating session via Server.acmd(). + + Safety: Session created in isolated test server only. + Cleanup: Server fixture finalizer handles session destruction. + """ + result = await server.acmd("new-session", "-d", "-P", "-F#{session_id}") + session_id = result.stdout[0] + + # Verify session was created + assert session_id.startswith("$") + assert server.has_session(session_id) + + # Verify we can get the session object + session = Session.from_session_id(session_id=session_id, server=server) + assert isinstance(session, Session) + session_obj_id = session.session_id + assert session_obj_id is not None + assert session_obj_id == session_id + + +# ============================================================================ +# Error Handling Tests +# ============================================================================ + + +@pytest.mark.asyncio +async def test_async_invalid_command(server: Server) -> None: + """Test async error handling for invalid commands. + + Safety: Invalid commands executed in isolated server only. + """ + # AsyncTmuxCmd captures errors in stderr rather than raising + result = await server.acmd("nonexistent-command-xyz") + + # Invalid command should populate stderr + assert len(result.stderr) > 0 + assert result.returncode != 0 + + +@pytest.mark.asyncio +async def test_async_session_not_found(server: Server) -> None: + """Test error when targeting nonexistent session. + + Safety: Test only affects isolated server. + """ + # has-session returns non-zero when session doesn't exist + result = await server.acmd("has-session", "-t", "nonexistent_session_xyz_123") + + # has-session returns 1 when session doesn't exist + assert result.returncode != 0 + + +# ============================================================================ +# Concurrent Operations Tests +# ============================================================================ + + +@pytest.mark.asyncio +async def test_concurrent_session_creation(server: Server) -> None: + """Test creating multiple sessions concurrently. + + Safety: All sessions created in isolated test server. + Demonstrates async benefit: concurrent tmux operations. + Cleanup: Server fixture finalizer handles all session destruction. + """ + + async def create_session(index: int) -> Session: + """Create a session asynchronously.""" + result = await server.acmd( + "new-session", + "-d", + "-P", + "-F#{session_id}", + "-s", + f"concurrent_test_{index}", + ) + session_id = result.stdout[0] + return Session.from_session_id(session_id=session_id, server=server) + + # Create 3 sessions concurrently + sessions = await asyncio.gather( + create_session(1), + create_session(2), + create_session(3), + ) + + # Verify all sessions were created + assert len(sessions) == 3 + assert all(isinstance(s, Session) for s in sessions) + + # Verify all session IDs are unique + session_ids: set[str] = set() + for session in sessions: + assert session.session_id is not None + session_ids.add(session.session_id) + assert len(session_ids) == 3 + + # Verify all sessions exist in server + for session_id in session_ids: + assert server.has_session(session_id) + + +@pytest.mark.asyncio +async def test_concurrent_session_queries(server: Server) -> None: + """Test querying multiple sessions concurrently. + + Safety: All sessions created/queried in isolated test server. + Demonstrates async benefit: parallel queries faster than sequential. + """ + # Create 5 sessions first + session_ids = [] + for i in range(5): + result = await server.acmd( + "new-session", + "-d", + "-P", + "-F#{session_id}", + "-s", + f"query_test_{i}", + ) + session_ids.append(result.stdout[0]) + + async def query_session(session_id: str) -> SessionQueryInfo: + """Query session information asynchronously.""" + result = await server.acmd( + "display-message", + "-t", + session_id, + "-p", + "#{session_id}:#{session_name}:#{session_windows}", + ) + output = result.stdout[0] + parts = output.split(":") + return SessionQueryInfo(id=parts[0], name=parts[1], windows=int(parts[2])) + + # Query all sessions concurrently + results: list[SessionQueryInfo] = await asyncio.gather( + *[query_session(sid) for sid in session_ids] + ) + + # Verify all queries returned valid data + assert len(results) == 5 + for i, info in enumerate(results): + assert info.id == session_ids[i] + assert info.name == f"query_test_{i}" + assert info.windows >= 1 + + +@pytest.mark.asyncio +async def test_batch_session_operations(server: Server) -> None: + """Test batch create and verify pattern. + + Safety: All operations in isolated test server. + Real-world pattern: Set up multiple sessions efficiently. + """ + session_names = [ + "dev_frontend", + "dev_backend", + "dev_database", + "logs_monitoring", + ] + + async def create_and_verify(name: str) -> tuple[str, bool]: + """Create session and verify it exists.""" + result = await server.acmd( + "new-session", + "-d", + "-P", + "-F#{session_id}", + "-s", + name, + ) + session_id = result.stdout[0] + + # Verify via has-session + check_result = await server.acmd("has-session", "-t", name) + exists = check_result.returncode == 0 + + return (session_id, exists) + + # Create all sessions concurrently + results = await asyncio.gather(*[create_and_verify(name) for name in session_names]) + + # Verify all sessions were created and verified + assert len(results) == 4 + for (session_id, exists), name in zip(results, session_names, strict=False): + assert session_id.startswith("$") + assert exists, f"Session {name} not found after creation" + assert server.has_session(name) + + +# ============================================================================ +# Server.anew_session() Tests +# ============================================================================ + + +@pytest.mark.asyncio +async def test_anew_session_basic(server: Server) -> None: + """Test Server.anew_session() creates session. + + Safety: Session created in isolated test server. + Demonstrates: High-level async session creation API. + """ + session = await server.anew_session("test_anew_session") + + # Verify session created with correct properties + session_name = session.session_name + assert session_name is not None + assert session_name == "test_anew_session" + assert server.has_session("test_anew_session") + assert isinstance(session, Session) + session_id = session.session_id + assert session_id is not None + assert session_id.startswith("$") + + +@pytest.mark.asyncio +async def test_anew_session_with_environment(server: Server) -> None: + """Test Server.anew_session() with environment variables. + + Safety: Session with env vars created in isolated test server. + Real-world pattern: Pass environment configuration to session. + """ + env_vars = { + "TEST_VAR": "test_value", + "ANOTHER_VAR": "another_value", + } + + session = await server.anew_session( + "test_env_session", + environment=env_vars, + ) + + # Verify session created + env_session_name = session.session_name + assert env_session_name is not None + assert env_session_name == "test_env_session" + assert server.has_session("test_env_session") + + # Verify environment variables were set + # Query environment in the session's pane + result = await session.acmd( + "show-environment", + "-s", + "TEST_VAR", + ) + # tmux formats env vars as: TEST_VAR="test_value"; export TEST_VAR; + assert "TEST_VAR" in result.stdout[0] + assert "test_value" in result.stdout[0] + + +@pytest.mark.asyncio +async def test_anew_session_concurrent(server: Server) -> None: + """Test creating multiple sessions concurrently via anew_session(). + + Safety: All sessions created in isolated test server. + Demonstrates: Async benefit - concurrent high-level session creation. + """ + + async def create_session(name: str) -> Session: + """Create session using anew_session().""" + return await server.anew_session(name) + + # Create 4 sessions concurrently + sessions = await asyncio.gather( + create_session("concurrent_a"), + create_session("concurrent_b"), + create_session("concurrent_c"), + create_session("concurrent_d"), + ) + + # Verify all sessions created + assert len(sessions) == 4 + assert all(isinstance(s, Session) for s in sessions) + + # Verify all have unique IDs and correct names + expected_names = ["concurrent_a", "concurrent_b", "concurrent_c", "concurrent_d"] + actual_names: list[str] = [] + for session in sessions: + assert session.session_name is not None + actual_names.append(session.session_name) + assert sorted(actual_names) == sorted(expected_names) + + # Verify all exist in server + for name in actual_names: + assert server.has_session(name) + + +# ============================================================================ +# Server.ahas_session() Tests +# ============================================================================ + + +@pytest.mark.asyncio +async def test_ahas_session(server: Server) -> None: + """Test Server.ahas_session() checks session existence. + + Safety: All operations in isolated test server. + Demonstrates: Async session existence check. + """ + # Create a session first + session = await server.anew_session("test_has_session") + + # Verify ahas_session returns True for existing session + assert await server.ahas_session("test_has_session") is True + + # Verify ahas_session returns False for non-existent session + assert await server.ahas_session("nonexistent_session_xyz") is False + + # Verify exact=True works with session ID + session_id = session.session_id + assert session_id is not None + assert await server.ahas_session(session_id, exact=True) is True + + +@pytest.mark.asyncio +async def test_ahas_session_concurrent_checks(server: Server) -> None: + """Test checking multiple sessions concurrently via ahas_session(). + + Safety: All sessions created/checked in isolated test server. + Demonstrates: Async benefit - parallel existence checks. + """ + # Create 3 sessions + await asyncio.gather( + server.anew_session("check_a"), + server.anew_session("check_b"), + server.anew_session("check_c"), + ) + + # Check all sessions concurrently + results = await asyncio.gather( + server.ahas_session("check_a"), + server.ahas_session("check_b"), + server.ahas_session("check_c"), + server.ahas_session("nonexistent"), + ) + + # Verify results + assert results[0] is True # check_a exists + assert results[1] is True # check_b exists + assert results[2] is True # check_c exists + assert results[3] is False # nonexistent doesn't exist diff --git a/tests/asyncio/test_session.py b/tests/asyncio/test_session.py new file mode 100644 index 000000000..f6b5e6eda --- /dev/null +++ b/tests/asyncio/test_session.py @@ -0,0 +1,311 @@ +"""Tests for Session async operations. + +SAFETY: All tests use isolated test servers via fixtures. +Socket names: libtmux_test{8_random_chars} - never affects developer sessions. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass +from typing import TYPE_CHECKING + +import pytest + +if TYPE_CHECKING: + from libtmux.session import Session + + +@dataclass(slots=True) +class WindowInfo: + """Minimal window details fetched concurrently during tests.""" + + id: str + name: str + panes: int + + +@dataclass(slots=True) +class ProjectSessionStatus: + """Summary of session setup used for verification.""" + + session_id: str + name: str + window_count: int + + +logger = logging.getLogger(__name__) + + +# ============================================================================ +# Session.acmd() Tests +# ============================================================================ + + +@pytest.mark.asyncio +async def test_session_acmd_basic(session: Session) -> None: + """Test Session.acmd() executes in session context. + + Safety: Uses `session` fixture which depends on isolated `server`. + """ + # List windows in the session + result = await session.acmd("list-windows", "-F#{window_id}") + assert len(result.stdout) >= 1 + assert all(wid.startswith("@") for wid in result.stdout) + + +@pytest.mark.asyncio +async def test_session_acmd_new_window(session: Session) -> None: + """Test creating window via Session.acmd(). + + Safety: Window created in isolated test session only. + """ + # Get initial window count + initial_windows = session.windows + initial_count = len(initial_windows) + + # Create new window asynchronously + result = await session.acmd("new-window", "-P", "-F#{window_id}") + window_id = result.stdout[0] + assert window_id.startswith("@") + + # Refresh session and verify window was created + # Note: We need to re-query the session to see new window + result = await session.acmd("list-windows", "-F#{window_id}") + assert len(result.stdout) == initial_count + 1 + assert window_id in result.stdout + + +# ============================================================================ +# Concurrent Operations Tests +# ============================================================================ + + +@pytest.mark.asyncio +async def test_concurrent_window_creation(session: Session) -> None: + """Test creating multiple windows concurrently in same session. + + Safety: All windows created in isolated test session. + Demonstrates async benefit: parallel window creation. + """ + import asyncio + + async def create_window(name: str) -> str: + """Create a window and return its ID.""" + result = await session.acmd( + "new-window", + "-P", + "-F#{window_id}", + "-n", + name, + ) + return result.stdout[0] + + # Create 4 windows concurrently + window_ids = await asyncio.gather( + create_window("editor"), + create_window("terminal"), + create_window("logs"), + create_window("monitor"), + ) + + # Verify all windows were created + assert len(window_ids) == 4 + assert all(wid.startswith("@") for wid in window_ids) + assert len(set(window_ids)) == 4 # All unique + + # Verify windows exist in session + result = await session.acmd("list-windows", "-F#{window_id}") + for window_id in window_ids: + assert window_id in result.stdout + + +@pytest.mark.asyncio +async def test_parallel_window_queries(session: Session) -> None: + """Test querying window properties concurrently. + + Safety: All operations in isolated test session. + Real-world pattern: Gather window information efficiently. + """ + import asyncio + + # Create a few windows first + for i in range(3): + await session.acmd("new-window", "-n", f"win_{i}") + + # Get all window IDs + result = await session.acmd("list-windows", "-F#{window_id}") + window_ids = result.stdout + + async def get_window_info(window_id: str) -> WindowInfo: + """Get window name and pane count.""" + result = await session.acmd( + "display-message", + "-t", + window_id, + "-p", + "#{window_id}:#{window_name}:#{window_panes}", + ) + output = result.stdout[0] + parts = output.split(":") + return WindowInfo(id=parts[0], name=parts[1], panes=int(parts[2])) + + # Query all windows concurrently + window_infos: list[WindowInfo] = await asyncio.gather( + *[get_window_info(wid) for wid in window_ids] + ) + + # Verify all queries succeeded + assert len(window_infos) >= 3 + for info in window_infos: + assert info.id.startswith("@") + assert info.panes >= 1 + + +# ============================================================================ +# Session.anew_window() Tests +# ============================================================================ + + +@pytest.mark.asyncio +async def test_anew_window_basic(session: Session) -> None: + """Test Session.anew_window() creates window. + + Safety: Window created in isolated test session. + Demonstrates: High-level async window creation API. + """ + from libtmux.window import Window + + # Get initial window count + initial_result = await session.acmd("list-windows", "-F#{window_id}") + initial_count = len(initial_result.stdout) + + # Create new window using anew_window() + window = await session.anew_window("test_window") + + # Verify window created with correct properties + assert isinstance(window, Window) + window_id = window.window_id + assert window_id is not None + assert window_id.startswith("@") + + # Verify window was added to session + result = await session.acmd("list-windows", "-F#{window_id}") + assert len(result.stdout) == initial_count + 1 + assert window_id in result.stdout + + +@pytest.mark.asyncio +async def test_anew_window_with_directory(session: Session) -> None: + """Test Session.anew_window() with start_directory. + + Safety: Window created in isolated test session. + Real-world pattern: Create window in specific working directory. + """ + import asyncio + from pathlib import Path + + from libtmux.window import Window + + # Use /tmp as start directory + start_dir = Path("/tmp") + + window = await session.anew_window( + "dir_window", + start_directory=start_dir, + ) + + # Verify window created + assert isinstance(window, Window) + + # Verify working directory by sending pwd command + pane = window.active_pane + assert pane is not None + + # Clear pane first to ensure clean output + await pane.acmd("send-keys", "clear", "Enter") + await asyncio.sleep(0.1) + + # Send pwd command + await pane.acmd("send-keys", "pwd", "Enter") + await asyncio.sleep(0.3) + + # Capture output + result = await pane.acmd("capture-pane", "-p", "-S", "-") + # Check if /tmp appears in any line of output + output_text = "\n".join(result.stdout) + assert "/tmp" in output_text, f"Expected /tmp in output, got: {output_text}" + + +@pytest.mark.asyncio +async def test_anew_window_concurrent(session: Session) -> None: + """Test creating multiple windows concurrently via anew_window(). + + Safety: All windows created in isolated test session. + Demonstrates: Async benefit - concurrent high-level window creation. + """ + import asyncio + + from libtmux.window import Window + + async def create_window(name: str) -> Window: + """Create window using anew_window().""" + return await session.anew_window(name) + + # Create 4 windows concurrently + windows = await asyncio.gather( + create_window("window_1"), + create_window("window_2"), + create_window("window_3"), + create_window("window_4"), + ) + + # Verify all windows created + assert len(windows) == 4 + assert all(isinstance(w, Window) for w in windows) + + # Verify all have unique IDs + window_ids: set[str] = set() + for window in windows: + assert window.window_id is not None + window_ids.add(window.window_id) + assert len(window_ids) == 4 + + # Verify all exist in session + result = await session.acmd("list-windows", "-F#{window_id}") + for window_id in window_ids: + assert window_id in result.stdout + + +# ============================================================================ +# Session.arename_session() Tests +# ============================================================================ + + +@pytest.mark.asyncio +async def test_arename_session(session: Session) -> None: + """Test Session.arename_session() renames session. + + Safety: Session renamed in isolated test server. + Demonstrates: High-level async session rename API. + """ + # Get original name + original_name = session.session_name + assert original_name is not None + + # Rename session + new_name = "renamed_async_session" + result_session = await session.arename_session(new_name) + + # Verify return value is the session object + assert result_session is session + + # Verify session was renamed + session.refresh() + current_name = session.session_name + assert current_name is not None + assert current_name == new_name + + # Verify old name is gone, new name exists + assert not session.server.has_session(original_name) + assert session.server.has_session(new_name) diff --git a/tests/asyncio/test_window.py b/tests/asyncio/test_window.py new file mode 100644 index 000000000..bceb598ef --- /dev/null +++ b/tests/asyncio/test_window.py @@ -0,0 +1,205 @@ +"""Tests for Window async operations. + +SAFETY: All tests use isolated test servers via fixtures. +Socket names: libtmux_test{8_random_chars} - never affects developer sessions. +""" + +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING + +import pytest + +if TYPE_CHECKING: + from libtmux.session import Session + +logger = logging.getLogger(__name__) + + +# ============================================================================ +# Window.acmd() Tests +# ============================================================================ + + +@pytest.mark.asyncio +async def test_window_acmd_split_pane(session: Session) -> None: + """Test splitting pane via Window.acmd(). + + Safety: Pane created in isolated test window only. + """ + window = session.active_window + assert window is not None + + # Get initial pane count + result = await window.acmd("list-panes", "-F#{pane_id}") + initial_pane_count = len(result.stdout) + + # Split window to create new pane + result = await window.acmd("split-window", "-P", "-F#{pane_id}") + pane_id = result.stdout[0] + assert pane_id.startswith("%") + + # Verify new pane was created + result = await window.acmd("list-panes", "-F#{pane_id}") + assert len(result.stdout) == initial_pane_count + 1 + assert pane_id in result.stdout + + +# ============================================================================ +# Concurrent Operations Tests +# ============================================================================ + + +@pytest.mark.asyncio +async def test_concurrent_pane_splits(session: Session) -> None: + """Test splitting window into multiple panes concurrently. + + Safety: All panes created in isolated test window. + Demonstrates creating a multi-pane layout efficiently. + """ + import asyncio + + window = session.active_window + assert window is not None + + async def split_pane(direction: str) -> str: + """Split the window and return new pane ID.""" + result = await window.acmd( + "split-window", + direction, + "-P", + "-F#{pane_id}", + ) + return result.stdout[0] + + # Create a 2x2 grid: split horizontally then split each half vertically + # First split horizontally + pane1 = await split_pane("-h") + + # Now split both panes vertically in parallel + pane2, pane3 = await asyncio.gather( + split_pane("-v"), + split_pane("-v"), + ) + + # Verify we now have 4 panes (1 original + 3 created) + result = await window.acmd("list-panes", "-F#{pane_id}") + assert len(result.stdout) == 4 + + # Verify all created panes exist + pane_ids = result.stdout + assert pane1 in pane_ids + assert pane2 in pane_ids + assert pane3 in pane_ids + + +@pytest.mark.asyncio +async def test_parallel_pane_queries(session: Session) -> None: + """Test querying multiple panes concurrently. + + Safety: All operations in isolated test window. + Real-world pattern: Monitor multiple panes efficiently. + """ + import asyncio + + window = session.active_window + assert window is not None + + # Create 3 panes (1 original + 2 splits) + await window.acmd("split-window", "-h") + await window.acmd("split-window", "-v") + + # Get all pane IDs + result = await window.acmd("list-panes", "-F#{pane_id}") + pane_ids = result.stdout + assert len(pane_ids) == 3 + + async def get_pane_info(pane_id: str) -> dict[str, str]: + """Get pane dimensions and active status.""" + result = await window.acmd( + "display-message", + "-t", + pane_id, + "-p", + "#{pane_id}:#{pane_width}:#{pane_height}:#{pane_active}", + ) + output = result.stdout[0] + parts = output.split(":") + return { + "id": parts[0], + "width": parts[1], + "height": parts[2], + "active": parts[3], + } + + # Query all panes concurrently + pane_infos = await asyncio.gather(*[get_pane_info(pid) for pid in pane_ids]) + + # Verify all queries succeeded + assert len(pane_infos) == 3 + for info in pane_infos: + assert info["id"].startswith("%") + assert int(info["width"]) > 0 + assert int(info["height"]) > 0 + assert info["active"] in {"0", "1"} + + +# ============================================================================ +# Window.akill() Tests +# ============================================================================ + + +@pytest.mark.asyncio +async def test_akill_basic(session: Session) -> None: + """Test Window.akill() kills window. + + Safety: Windows created and killed in isolated test session. + Demonstrates: High-level async window destruction API. + """ + # Create 2 windows (session starts with 1) + window1 = await session.anew_window("window_to_kill") + window2 = await session.anew_window("window_to_keep") + + # Get window count before kill + result = await session.acmd("list-windows", "-F#{window_id}") + windows_before = len(result.stdout) + assert windows_before == 3 # original + 2 new + + # Kill window1 + await window1.akill() + + # Verify window1 is gone + result = await session.acmd("list-windows", "-F#{window_id}") + windows_after = len(result.stdout) + assert windows_after == windows_before - 1 + assert window1.window_id not in result.stdout + assert window2.window_id in result.stdout + + +@pytest.mark.asyncio +async def test_akill_all_except(session: Session) -> None: + """Test Window.akill() with all_except flag. + + Safety: Windows created and killed in isolated test session. + Real-world pattern: Clean up all windows except current one. + """ + # Create 4 additional windows (session starts with 1) + await session.anew_window("extra_1") + await session.anew_window("extra_2") + await session.anew_window("extra_3") + target_window = await session.anew_window("target_window") + + # Get window count before kill + result = await session.acmd("list-windows", "-F#{window_id}") + windows_before = len(result.stdout) + assert windows_before == 5 # original + 4 new + + # Kill all windows except target_window + await target_window.akill(all_except=True) + + # Verify only target_window remains + result = await session.acmd("list-windows", "-F#{window_id}") + windows_after = result.stdout + assert len(windows_after) == 1 + assert windows_after[0] == target_window.window_id diff --git a/uv.lock b/uv.lock index 1e8c29495..5f5df07f3 100644 --- a/uv.lock +++ b/uv.lock @@ -51,6 +51,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/b7/b8/3fe70c75fe32afc4bb507f75563d39bc5642255d1d94f1f23604725780bf/babel-2.17.0-py3-none-any.whl", hash = "sha256:4d0b53093fdfb4b21c92b5213dba5a1b23885afa8383709427046b21c366e5f2", size = 10182537, upload-time = "2025-02-01T15:17:37.39Z" }, ] +[[package]] +name = "backports-asyncio-runner" +version = "1.2.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/8e/ff/70dca7d7cb1cbc0edb2c6cc0c38b65cba36cccc491eca64cabd5fe7f8670/backports_asyncio_runner-1.2.0.tar.gz", hash = "sha256:a5aa7b2b7d8f8bfcaa2b57313f70792df84e32a2a746f585213373f900b42162", size = 69893, upload-time = "2025-07-02T02:27:15.685Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/a0/59/76ab57e3fe74484f48a53f8e337171b4a2349e506eabe136d7e01d059086/backports_asyncio_runner-1.2.0-py3-none-any.whl", hash = "sha256:0da0a936a8aeb554eccb426dc55af3ba63bcdc69fa1a600b5bb305413a4477b5", size = 12313, upload-time = "2025-07-02T02:27:14.263Z" }, +] + [[package]] name = "beautifulsoup4" version = "4.14.2" @@ -428,6 +437,7 @@ dev = [ { name = "mypy" }, { name = "myst-parser" }, { name = "pytest" }, + { name = "pytest-asyncio" }, { name = "pytest-cov" }, { name = "pytest-mock" }, { name = "pytest-rerunfailures" }, @@ -470,6 +480,7 @@ lint = [ testing = [ { name = "gp-libs" }, { name = "pytest" }, + { name = "pytest-asyncio" }, { name = "pytest-mock" }, { name = "pytest-rerunfailures" }, { name = "pytest-watcher" }, @@ -493,6 +504,7 @@ dev = [ { name = "mypy" }, { name = "myst-parser" }, { name = "pytest" }, + { name = "pytest-asyncio" }, { name = "pytest-cov" }, { name = "pytest-mock" }, { name = "pytest-rerunfailures" }, @@ -529,6 +541,7 @@ lint = [ testing = [ { name = "gp-libs" }, { name = "pytest" }, + { name = "pytest-asyncio" }, { name = "pytest-mock" }, { name = "pytest-rerunfailures" }, { name = "pytest-watcher" }, @@ -791,6 +804,20 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/a8/a4/20da314d277121d6534b3a980b29035dcd51e6744bd79075a6ce8fa4eb8d/pytest-8.4.2-py3-none-any.whl", hash = "sha256:872f880de3fc3a5bdc88a11b39c9710c3497a547cfa9320bc3c5e62fbf272e79", size = 365750, upload-time = "2025-09-04T14:34:20.226Z" }, ] +[[package]] +name = "pytest-asyncio" +version = "1.2.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "backports-asyncio-runner", marker = "python_full_version < '3.11'" }, + { name = "pytest" }, + { name = "typing-extensions", marker = "python_full_version < '3.13'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/42/86/9e3c5f48f7b7b638b216e4b9e645f54d199d7abbbab7a64a13b4e12ba10f/pytest_asyncio-1.2.0.tar.gz", hash = "sha256:c609a64a2a8768462d0c99811ddb8bd2583c33fd33cf7f21af1c142e824ffb57", size = 50119, upload-time = "2025-09-12T07:33:53.816Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/04/93/2fa34714b7a4ae72f2f8dad66ba17dd9a2c793220719e736dda28b7aec27/pytest_asyncio-1.2.0-py3-none-any.whl", hash = "sha256:8e17ae5e46d8e7efe51ab6494dd2010f4ca8dae51652aa3c8d55acf50bfb2e99", size = 15095, upload-time = "2025-09-12T07:33:52.639Z" }, +] + [[package]] name = "pytest-cov" version = "7.0.0"