Skip to content

Waiter v3.0: Requirements #584

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
1 task
tony opened this issue Feb 27, 2025 · 0 comments
Open
1 task

Waiter v3.0: Requirements #584

tony opened this issue Feb 27, 2025 · 0 comments

Comments

@tony
Copy link
Member

tony commented Feb 27, 2025

  • 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:

    • WaitTimeout and WaitResult needs to include a snapshot
    • Pytest comparions should be useful

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

  1. 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.
  2. Type Safety: Dataclass snapshots ensure fields are consistently named and typed.
  3. Maintainability: Easy to add or remove fields in one place without worrying about scattered dictionary references.
  4. 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

  1. Improved Debugging: Dataclass snapshots provide a consistent, structured view of tmux objects at timeout.
  2. Immutable States: Using frozen=True ensures the captured data remains a faithful “moment in time.”
  3. Type Safety: Your IDE or static analyzer knows exactly what fields exist on PaneSnapshot, etc.
  4. Centralized Logic: The from_* factory methods define how to copy fields once and for all, preventing drift across the codebase.
  5. 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.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
None yet
Projects
None yet
Development

No branches or pull requests

1 participant