Description
-
Snapshot / frame dataclass #370
We need a way to snapshot a state, including pane contents and the full object graph, for use in debug outputs.
Problem: We need the most recent state at the time of
WaitTimeout
, as things may shift after the timeout, including the pane, window, session, and even server itself being active. Sometimes these errors can manifest remotely in CI pipelines. -
Waiter v3.0:
Waiter 3.0
Enhanced Waiter Proposal With Dataclass Snapshots
This proposal extends libtmux’s Waiter functionality so that when a wait operation times out, the resulting exceptions and WaitResult
objects contain immutable dataclass snapshots of tmux objects. These snapshots replace the old dictionary-based snapshots (e.g., self.pane_snapshot = { ... }
) with more robust, typed, and maintainable dataclasses like PaneSnapshot
, WindowSnapshot
, etc.
1. Rationale
- Improved Diagnostics: When a wait times out, having a complete, immutable record of the tmux object’s state in a well-defined dataclass makes debugging faster.
- Type Safety: Dataclass snapshots ensure fields are consistently named and typed.
- Maintainability: Easy to add or remove fields in one place without worrying about scattered dictionary references.
- Better Test Output: With a pytest plugin (using
pytest_assertrepr_compare
), the wait-timeout errors can print structured snapshots automatically.
2. Snapshot Dataclasses (Vanilla Python)
Below is an example of vanilla dataclass snapshots for the four main tmux objects (Server
, Session
, Window
, Pane
). Each snapshot is immutable (frozen=True
), capturing the object’s fields at a specific moment in time:
# libtmux/snapshot.py
from __future__ import annotations
import datetime
import typing as t
from dataclasses import dataclass, field, fields
import copy
# Reference imports to the original live tmux objects
from libtmux.server import Server
from libtmux.session import Session
from libtmux.window import Window
from libtmux.pane import Pane
@dataclass(frozen=True)
class PaneSnapshot(Pane):
"""Immutable snapshot of a Pane."""
pane_content: t.Optional[t.List[str]] = None
created_at: datetime.datetime = field(default_factory=lambda: datetime.datetime.now(datetime.timezone.utc))
# Link back to the parent window snapshot, if needed
window_snapshot: t.Optional[WindowSnapshot] = None
# Override tmux commands to make snapshot read-only
def cmd(self, *args, **kwargs):
raise NotImplementedError("PaneSnapshot cannot execute tmux commands")
def capture_pane(self, *args, **kwargs):
return self.pane_content or []
@classmethod
def from_pane(cls, pane: Pane, capture_content: bool = True) -> PaneSnapshot:
content = None
if capture_content:
try:
content = pane.capture_pane()
except Exception:
pass
# Gather fields from Pane
snapshot_kwargs = {
field_.name: copy.deepcopy(getattr(pane, field_.name, None))
for field_ in fields(Pane) if hasattr(pane, field_.name)
}
snapshot_kwargs["pane_content"] = content
return cls(**snapshot_kwargs)
@dataclass(frozen=True)
class WindowSnapshot(Window):
"""Immutable snapshot of a Window."""
panes_snapshot: t.List[PaneSnapshot] = field(default_factory=list)
created_at: datetime.datetime = field(default_factory=lambda: datetime.datetime.now(datetime.timezone.utc))
def cmd(self, *args, **kwargs):
raise NotImplementedError("WindowSnapshot cannot execute tmux commands")
@property
def panes(self):
return self.panes_snapshot
@classmethod
def from_window(cls, window: Window, include_panes: bool = True) -> WindowSnapshot:
snapshot_kwargs = {
field_.name: copy.deepcopy(getattr(window, field_.name, None))
for field_ in fields(Window) if hasattr(window, field_.name)
}
snapshot_kwargs["panes_snapshot"] = []
snapshot = cls(**snapshot_kwargs)
if include_panes:
# Build PaneSnapshot objects for each pane
pane_snaps = [PaneSnapshot.from_pane(p, capture_content=True) for p in window.panes]
object.__setattr__(snapshot, "panes_snapshot", pane_snaps)
return snapshot
@dataclass(frozen=True)
class SessionSnapshot(Session):
"""Immutable snapshot of a Session."""
windows_snapshot: t.List[WindowSnapshot] = field(default_factory=list)
created_at: datetime.datetime = field(default_factory=lambda: datetime.datetime.now(datetime.timezone.utc))
def cmd(self, *args, **kwargs):
raise NotImplementedError("SessionSnapshot cannot execute tmux commands")
@property
def windows(self):
return self.windows_snapshot
@classmethod
def from_session(cls, session: Session, include_windows: bool = True) -> SessionSnapshot:
snapshot_kwargs = {
field_.name: copy.deepcopy(getattr(session, field_.name, None))
for field_ in fields(Session) if hasattr(session, field_.name)
}
snapshot_kwargs["windows_snapshot"] = []
snapshot = cls(**snapshot_kwargs)
if include_windows:
w_snaps = [WindowSnapshot.from_window(w, include_panes=True) for w in session.windows]
object.__setattr__(snapshot, "windows_snapshot", w_snaps)
return snapshot
@dataclass(frozen=True)
class ServerSnapshot(Server):
"""Immutable snapshot of a Server."""
sessions_snapshot: t.List[SessionSnapshot] = field(default_factory=list)
created_at: datetime.datetime = field(default_factory=lambda: datetime.datetime.now(datetime.timezone.utc))
def cmd(self, *args, **kwargs):
raise NotImplementedError("ServerSnapshot cannot execute tmux commands")
@property
def sessions(self):
return self.sessions_snapshot
@classmethod
def from_server(cls, server: Server, include_sessions: bool = True) -> ServerSnapshot:
snapshot_kwargs = {
field_.name: copy.deepcopy(getattr(server, field_.name, None))
for field_ in fields(Server) if hasattr(server, field_.name)
}
snapshot_kwargs["sessions_snapshot"] = []
snapshot = cls(**snapshot_kwargs)
if include_sessions:
s_snaps = [SessionSnapshot.from_session(s, include_windows=True) for s in server.sessions]
object.__setattr__(snapshot, "sessions_snapshot", s_snaps)
return snapshot
Note: Each snapshot class is read-only, forbidding tmux commands. By default, we store a created_at
timestamp and optionally capture the entire parent-child hierarchy (e.g., session → windows → panes).
3. Specialized WaitTimeout
Exceptions
Whenever a wait operation times out, we raise specialized exceptions (PaneWaitTimeout
, WindowWaitTimeout
, etc.) that embed these snapshot objects, rather than dictionaries:
# libtmux/exc.py (or a new module for wait-related exceptions)
import time
from libtmux.exc import LibTmuxException, WaitTimeout
from libtmux.snapshot import (
PaneSnapshot, WindowSnapshot, SessionSnapshot, ServerSnapshot
)
from libtmux._internal.waiter_types import WaitResult
class PaneWaitTimeout(WaitTimeout):
def __init__(self, message: str, pane, wait_result: WaitResult | None = None):
super().__init__(message, wait_result)
self.pane = pane
# Build a snapshot for debugging
self.pane_snapshot = PaneSnapshot.from_pane(
pane, capture_content=not wait_result or not wait_result.content
)
# Store the final content array for convenience
if wait_result and wait_result.content:
self.pane_contents = wait_result.content
else:
self.pane_contents = self.pane_snapshot.pane_content
# ... Similarly for WindowWaitTimeout, SessionWaitTimeout, ServerWaitTimeout ...
When a PaneWaitTimeout
is raised, code (or pytest) can inspect exc.pane_snapshot
to see exactly what the pane state looked like at the moment of timeout.
4. Updating Wait Functions to Raise Specialized Exceptions
In waiter.py
(or wherever your wait logic lives), replace the generic WaitTimeout
with specialized versions. For example:
# In waiter.py
from libtmux.exc import PaneWaitTimeout
def wait_for_pane_content(...):
# ...
try:
success, exception = retry_until_extended(
check_content,
timeout,
interval=interval,
raises=raises,
)
if exception and raises:
raise PaneWaitTimeout(str(exception), pane=pane, wait_result=result) from exception
# ...
except WaitTimeout as e:
if raises:
raise PaneWaitTimeout(str(e), pane=pane, wait_result=result) from e
# ...
Likewise for functions like wait_for_session_condition
, raise a SessionWaitTimeout
; for wait_for_window_condition
, a WindowWaitTimeout
; and so on.
5. pytest Integration (Optional)
Finally, you can define a pytest_assertrepr_compare
hook that prints these snapshots in a user-friendly manner:
def pytest_assertrepr_compare(op, left, right):
if not (isinstance(left, WaitTimeout) or isinstance(right, WaitTimeout)):
return None
exc = left if isinstance(left, WaitTimeout) else right
lines = [f"WaitTimeout: {exc}", ""]
if isinstance(exc, PaneWaitTimeout):
snap = exc.pane_snapshot
lines.append(f"Pane ID: {snap.id}")
lines.append(f"Window ID: {snap.window_id}")
lines.append(f"Pane Content (first 5 lines):")
for i, line in enumerate(snap.pane_content[:5] if snap.pane_content else []):
lines.append(f" {i:3d} | {line}")
# etc.
return lines
So when a test fails on an assertion involving a PaneWaitTimeout
, pytest will show a structured snapshot of the pane.
6. Summary of Benefits
- Improved Debugging: Dataclass snapshots provide a consistent, structured view of tmux objects at timeout.
- Immutable States: Using
frozen=True
ensures the captured data remains a faithful “moment in time.” - Type Safety: Your IDE or static analyzer knows exactly what fields exist on
PaneSnapshot
, etc. - Centralized Logic: The
from_*
factory methods define how to copy fields once and for all, preventing drift across the codebase. - Better CI Diagnostics: The specialized exceptions store all relevant object details automatically, reducing guesswork in failing test logs.
By leveraging these dataclass snapshots, libtmux’s Waiter becomes far more transparent about why timeouts occur, simplifying test maintenance and debugging.