feat: Add test_snapshot

Author: tony

src/libtmux/pane.py

@@ -22,6 +22,7 @@
 )
 from libtmux.formats import FORMAT_SEPARATOR
 from libtmux.neo import Obj, fetch_obj
+from libtmux.snapshot import PaneRecording, PaneSnapshot
 
 from . import exc
 
@@ -342,6 +343,43 @@ def capture_pane(
             cmd.extend(["-E", str(end)])
         return self.cmd(*cmd).stdout
 
+    def snapshot(
+        self,
+        start: t.Literal["-"] | int | None = None,
+        end: t.Literal["-"] | int | None = None,
+    ) -> PaneSnapshot:
+        """Create a snapshot of the pane's current state.
+
+        This is a convenience method that creates a :class:`PaneSnapshot` instance
+        from the current pane state.
+
+        Parameters
+        ----------
+        start : int | "-" | None
+            Start line for capture_pane
+        end : int | "-" | None
+            End line for capture_pane
+
+        Returns
+        -------
+        PaneSnapshot
+            A frozen snapshot of the pane's current state
+        """
+        return PaneSnapshot.from_pane(self, start=start, end=end)
+
+    def record(self) -> PaneRecording:
+        """Create a new recording for this pane.
+
+        This is a convenience method that creates a :class:`PaneRecording` instance
+        for recording snapshots of this pane.
+
+        Returns
+        -------
+        PaneRecording
+            A new recording instance for this pane
+        """
+        return PaneRecording()
+
     def send_keys(
         self,
         cmd: str,

src/libtmux/snapshot.py

@@ -0,0 +1,225 @@
+"""Snapshot and recording functionality for tmux panes."""
+
+from __future__ import annotations
+
+import dataclasses
+import datetime
+import typing as t
+
+from typing_extensions import Self
+
+from libtmux.formats import PANE_FORMATS
+
+if t.TYPE_CHECKING:
+    from collections.abc import Iterator, Sequence
+
+    from libtmux.pane import Pane
+
+
+@dataclasses.dataclass(frozen=True)
+class PaneSnapshot:
+    """A frozen snapshot of a pane's state at a point in time.
+
+    This class captures both the content and metadata of a tmux pane,
+    making it suitable for testing and debugging purposes.
+
+    Attributes
+    ----------
+    content : list[str]
+        The captured content of the pane
+    timestamp : datetime.datetime
+        When the snapshot was taken (in UTC)
+    pane_id : str
+        The ID of the pane
+    window_id : str
+        The ID of the window containing the pane
+    session_id : str
+        The ID of the session containing the window
+    server_name : str
+        The name of the tmux server
+    metadata : dict[str, str]
+        Additional pane metadata from tmux formats
+    """
+
+    content: list[str]
+    timestamp: datetime.datetime
+    pane_id: str
+    window_id: str
+    session_id: str
+    server_name: str
+    metadata: dict[str, str]
+
+    @classmethod
+    def from_pane(
+        cls,
+        pane: Pane,
+        start: t.Literal["-"] | int | None = None,
+        end: t.Literal["-"] | int | None = None,
+    ) -> Self:
+        """Create a snapshot from a pane.
+
+        Parameters
+        ----------
+        pane : Pane
+            The pane to snapshot
+        start : int | "-" | None
+            Start line for capture_pane
+        end : int | "-" | None
+            End line for capture_pane
+
+        Returns
+        -------
+        PaneSnapshot
+            A frozen snapshot of the pane's state
+        """
+        metadata = {
+            fmt: getattr(pane, fmt)
+            for fmt in PANE_FORMATS
+            if hasattr(pane, fmt) and getattr(pane, fmt) is not None
+        }
+
+        content = pane.capture_pane(start=start, end=end)
+        if isinstance(content, str):
+            content = [content]
+
+        return cls(
+            content=content,
+            timestamp=datetime.datetime.now(datetime.timezone.utc),
+            pane_id=str(pane.pane_id),
+            window_id=str(pane.window.window_id),
+            session_id=str(pane.session.session_id),
+            server_name=str(pane.server.socket_name),
+            metadata=metadata,
+        )
+
+    def __str__(self) -> str:
+        """Return a string representation of the snapshot.
+
+        Returns
+        -------
+        str
+            A formatted string showing the snapshot content and metadata
+        """
+        return (
+            f"PaneSnapshot(pane={self.pane_id}, window={self.window_id}, "
+            f"session={self.session_id}, server={self.server_name}, "
+            f"timestamp={self.timestamp.isoformat()}, "
+            f"content=\n{self.content_str})"
+        )
+
+    @property
+    def content_str(self) -> str:
+        """Get the pane content as a single string.
+
+        Returns
+        -------
+        str
+            The pane content with lines joined by newlines
+        """
+        return "\n".join(self.content)
+
+
+@dataclasses.dataclass
+class PaneRecording:
+    """A time-series recording of pane snapshots.
+
+    This class maintains an ordered sequence of pane snapshots,
+    allowing for analysis of how a pane's content changes over time.
+
+    Attributes
+    ----------
+    snapshots : list[PaneSnapshot]
+        The sequence of snapshots in chronological order
+    """
+
+    snapshots: list[PaneSnapshot] = dataclasses.field(default_factory=list)
+
+    def add_snapshot(
+        self,
+        pane: Pane,
+        start: t.Literal["-"] | int | None = None,
+        end: t.Literal["-"] | int | None = None,
+    ) -> None:
+        """Add a new snapshot to the recording.
+
+        Parameters
+        ----------
+        pane : Pane
+            The pane to snapshot
+        start : int | "-" | None
+            Start line for capture_pane
+        end : int | "-" | None
+            End line for capture_pane
+        """
+        self.snapshots.append(PaneSnapshot.from_pane(pane, start=start, end=end))
+
+    def __len__(self) -> int:
+        """Get the number of snapshots in the recording.
+
+        Returns
+        -------
+        int
+            The number of snapshots
+        """
+        return len(self.snapshots)
+
+    def __iter__(self) -> Iterator[PaneSnapshot]:
+        """Iterate through snapshots in chronological order.
+
+        Returns
+        -------
+        Iterator[PaneSnapshot]
+            Iterator over the snapshots
+        """
+        return iter(self.snapshots)
+
+    def __getitem__(self, index: int) -> PaneSnapshot:
+        """Get a snapshot by index.
+
+        Parameters
+        ----------
+        index : int
+            The index of the snapshot to retrieve
+
+        Returns
+        -------
+        PaneSnapshot
+            The snapshot at the specified index
+        """
+        return self.snapshots[index]
+
+    @property
+    def latest(self) -> PaneSnapshot | None:
+        """Get the most recent snapshot.
+
+        Returns
+        -------
+        PaneSnapshot | None
+            The most recent snapshot, or None if no snapshots exist
+        """
+        return self.snapshots[-1] if self.snapshots else None
+
+    def get_snapshots_between(
+        self,
+        start_time: datetime.datetime,
+        end_time: datetime.datetime,
+    ) -> Sequence[PaneSnapshot]:
+        """Get snapshots between two points in time.
+
+        Parameters
+        ----------
+        start_time : datetime.datetime
+            The start of the time range
+        end_time : datetime.datetime
+            The end of the time range
+
+        Returns
+        -------
+        Sequence[PaneSnapshot]
+            Snapshots within the specified time range
+        """
+        return [
+            snapshot
+            for snapshot in self.snapshots
+            if start_time <= snapshot.timestamp <= end_time
+        ]

tests/test_snapshot.py

@@ -0,0 +1,107 @@
+"""Tests for libtmux snapshot functionality."""
+
+from __future__ import annotations
+
+import datetime
+import shutil
+import time
+import typing as t
+
+from libtmux.snapshot import PaneRecording, PaneSnapshot
+
+if t.TYPE_CHECKING:
+    from libtmux.session import Session
+
+
+def test_pane_snapshot(session: Session) -> None:
+    """Test creating a PaneSnapshot from a pane."""
+    env = shutil.which("env")
+    assert env is not None, "Cannot find usable `env` in PATH."
+
+    session.new_window(
+        attach=True,
+        window_name="snapshot_test",
+        window_shell=f"{env} PS1='$ ' sh",
+    )
+    pane = session.active_window.active_pane
+    assert pane is not None
+
+    # Take initial snapshot
+    snapshot = PaneSnapshot.from_pane(pane)
+    assert snapshot.content == ["$"]
+    assert snapshot.pane_id == pane.pane_id
+    assert snapshot.window_id == pane.window.window_id
+    assert snapshot.session_id == pane.session.session_id
+    assert snapshot.server_name == pane.server.socket_name
+    assert isinstance(snapshot.timestamp, datetime.datetime)
+    assert snapshot.timestamp.tzinfo == datetime.timezone.utc
+
+    # Verify metadata
+    assert "pane_id" in snapshot.metadata
+    assert "pane_width" in snapshot.metadata
+    assert "pane_height" in snapshot.metadata
+
+    # Test string representation
+    str_repr = str(snapshot)
+    assert "PaneSnapshot" in str_repr
+    assert snapshot.pane_id in str_repr
+    assert snapshot.window_id in str_repr
+    assert snapshot.session_id in str_repr
+    assert snapshot.server_name in str_repr
+    assert snapshot.timestamp.isoformat() in str_repr
+    assert "$" in str_repr
+
+
+def test_pane_recording(session: Session) -> None:
+    """Test creating and managing a PaneRecording."""
+    env = shutil.which("env")
+    assert env is not None, "Cannot find usable `env` in PATH."
+
+    session.new_window(
+        attach=True,
+        window_name="recording_test",
+        window_shell=f"{env} PS1='$ ' sh",
+    )
+    pane = session.active_window.active_pane
+    assert pane is not None
+
+    recording = PaneRecording()
+    assert len(recording) == 0
+    assert recording.latest is None
+
+    # Take initial snapshot
+    recording.add_snapshot(pane)
+    assert len(recording) == 1
+    assert recording.latest is not None
+    assert recording.latest.content == ["$"]
+
+    # Send some commands and take more snapshots
+    pane.send_keys("echo 'Hello'")
+    time.sleep(0.1)  # Give tmux time to update
+    recording.add_snapshot(pane)
+
+    pane.send_keys("echo 'World'")
+    time.sleep(0.1)  # Give tmux time to update
+    recording.add_snapshot(pane)
+
+    assert len(recording) == 3
+
+    # Test iteration
+    snapshots = list(recording)
+    assert len(snapshots) == 3
+    assert snapshots[0].content == ["$"]
+    assert "Hello" in snapshots[1].content_str
+    assert "World" in snapshots[2].content_str
+
+    # Test indexing
+    assert recording[0].content == ["$"]
+    assert "Hello" in recording[1].content_str
+    assert "World" in recording[2].content_str
+
+    # Test time-based filtering
+    start_time = snapshots[0].timestamp
+    mid_time = snapshots[1].timestamp
+    end_time = snapshots[2].timestamp
+
+    assert len(recording.get_snapshots_between(start_time, end_time)) == 3
+    assert len(recording.get_snapshots_between(mid_time, end_time)) == 2