diff --git a/CHANGES b/CHANGES index 115d0bc43..dae2824d4 100644 --- a/CHANGES +++ b/CHANGES @@ -45,6 +45,47 @@ $ uvx --from 'libtmux' --prerelease allow python _Notes on the upcoming release will go here._ +### Breaking changes + +#### Query errors join the libtmux exception hierarchy (#718) + +{exc}`~libtmux.exc.ObjectDoesNotExist` and +{exc}`~libtmux.exc.MultipleObjectsReturned` now subclass +{exc}`~libtmux.exc.LibTmuxException`. Because +{exc}`~libtmux.exc.TmuxObjectDoesNotExist` inherits from +{exc}`~libtmux.exc.ObjectDoesNotExist`, it now falls under +{exc}`~libtmux.exc.LibTmuxException` as well. Previously, a lookup-specific +handler could follow the common libtmux handler: + +```python +from libtmux import exc + +try: + pane = server.panes.get(pane_id="%0") +except exc.LibTmuxException: + pane = None +except exc.ObjectDoesNotExist: + pane = None +``` + +Put the more specific handler first: + +```python +from libtmux import exc + +try: + pane = server.panes.get(pane_id="%0") +except exc.ObjectDoesNotExist: + pane = None +except exc.LibTmuxException: + pane = None +``` + +Blanket retry policies for {exc}`~libtmux.exc.LibTmuxException` should exclude +{exc}`~libtmux.exc.ObjectDoesNotExist` and +{exc}`~libtmux.exc.MultipleObjectsReturned`: they report deterministic missing +or ambiguous lookups, not transient tmux failures. + ### What's new #### Find where you are running (#714) @@ -67,6 +108,21 @@ See {ref}`self-location` for the whole story — why the session id tmux exports goes stale, the window that *contains* you versus the one in front of you, and how to test code that locates itself. +#### Ask a window which sessions hold it (#718) + +{attr}`~libtmux.Window.linked_sessions` lists every +{class}`~libtmux.Session` from which a window is reachable. A typical window +returns one session; a linked or grouped window can return several, with each +session appearing once even when it holds the same window at multiple indexes. + +Use {attr}`~libtmux.Window.linked_sessions` when you need every holder. +{attr}`~libtmux.Window.session` follows the `session_id` recorded on that +{class}`~libtmux.Window` instance; use +{meth}`~libtmux.Window.from_window_id` when you need tmux to resolve a window id +afresh. +See {ref}`winlinks` for the relationship between sessions, indexes, and shared +windows. + ### Fixes #### Linked windows resolve to the session tmux would use (#713) @@ -74,8 +130,10 @@ how to test code that locates itself. A window can be linked into several sessions at once, and libtmux used to pick one of them by accident — whichever sorted last by name. So {meth}`Window.from_window_id() `, -{meth}`Pane.from_pane_id() ` and both `refresh()` -methods could change their answer when you renamed a session. +{meth}`Pane.from_pane_id() `, +{meth}`Window.refresh() ` and +{meth}`Pane.refresh() ` could change their answer when you +renamed a session. libtmux now lets tmux resolve the object, so a window reports the session tmux itself would act on. For a window in a single session — the ordinary case — @@ -85,6 +143,40 @@ sessions holding it, where before it was stable but arbitrary. The lookups also got cheaper: they list one window's panes or one session's windows instead of scanning the whole server. +#### A window linked twice now reports tmux's index (#718) + +tmux lets one session link the same window at multiple indexes. +{meth}`Window.from_window_id() ` and +{meth}`Window.refresh() ` now give +{attr}`Window.window_index ` the index tmux would +choose: the current link when that window is current, otherwise the +lowest-indexed link. Windows held at one index are unaffected. + +#### Missing and ambiguous lookups explain what happened (#718) + +{meth}`QueryList.get() ` now names +both the lookup and its missing or ambiguous outcome. A missing +`pane_id="%99"` raises {exc}`~libtmux.exc.ObjectDoesNotExist` with +`No objects found: pane_id='%99'`; two matches for `pane_id="%0"` raise +{exc}`~libtmux.exc.MultipleObjectsReturned` with +`Multiple objects returned (2): pane_id='%0'`. + +### Documentation + +#### Understand winlinks in server-wide listings (#718) + +{ref}`winlinks` explains that +{attr}`Server.windows ` and +{attr}`Server.panes ` enumerate {term}`winlinks ` +— the session, index, and window relationships tmux stores — rather than +deduplicating windows. A shared window can therefore appear once for each path +by which a session reaches it. + +Read those rows when you need their session and index context. For a unique +lookup by id, use {meth}`Pane.from_pane_id() ` or +{meth}`Window.from_window_id() `; use +{attr}`~libtmux.Window.linked_sessions` when you need every holding session. + ## libtmux 0.61.0 (2026-07-04) libtmux 0.61.0 hardens support for the tmux 3.7 patch line. It fixes diff --git a/MIGRATION b/MIGRATION index f185c6021..71f598170 100644 --- a/MIGRATION +++ b/MIGRATION @@ -113,6 +113,71 @@ sections below for detailed migration examples and code samples. _Detailed migration steps for the next version will be posted here._ +## libtmux 0.62.x: Query exceptions join the hierarchy (#718) + +{exc}`~libtmux.exc.ObjectDoesNotExist` and +{exc}`~libtmux.exc.MultipleObjectsReturned` now subclass +{exc}`~libtmux.exc.LibTmuxException`. +{exc}`~libtmux.exc.TmuxObjectDoesNotExist` already subclasses +{exc}`~libtmux.exc.ObjectDoesNotExist`, so it follows the same hierarchy. + +Before 0.62, a broad libtmux handler could precede lookup-specific handlers +because the exception families were separate: + +```python +from libtmux import exc + +try: + pane = server.panes.get(pane_id="%0") + pane.refresh() +except exc.LibTmuxException: + retry_tmux_operation() +except exc.TmuxObjectDoesNotExist: + handle_disappeared_pane() +except exc.ObjectDoesNotExist: + handle_missing_pane() +except exc.MultipleObjectsReturned: + handle_ambiguous_pane() +``` + +Starting with 0.62, order the handlers from most specific to most general so +the broad clause does not intercept lookup outcomes: + +```python +from libtmux import exc + +try: + pane = server.panes.get(pane_id="%0") + pane.refresh() +except exc.TmuxObjectDoesNotExist: + handle_disappeared_pane() +except exc.ObjectDoesNotExist: + handle_missing_pane() +except exc.MultipleObjectsReturned: + handle_ambiguous_pane() +except exc.LibTmuxException: + retry_tmux_operation() +``` + +Blanket retry predicates for {exc}`~libtmux.exc.LibTmuxException` now also +match deterministic missing and ambiguous lookups. Exclude both lookup +exceptions explicitly: + +```python +def should_retry(error: Exception) -> bool: + lookup_errors = ( + exc.ObjectDoesNotExist, + exc.MultipleObjectsReturned, + ) + return isinstance(error, exc.LibTmuxException) and not isinstance( + error, + lookup_errors, + ) +``` + +The {exc}`~libtmux.exc.ObjectDoesNotExist` exclusion also covers +{exc}`~libtmux.exc.TmuxObjectDoesNotExist`. + ## libtmux 0.57.0: Subcommand-tagged exceptions (#672) ### `LibTmuxException` `str()` gains a subcommand prefix diff --git a/docs/glossary.md b/docs/glossary.md index 64fce2b36..08fe11218 100644 --- a/docs/glossary.md +++ b/docs/glossary.md @@ -58,6 +58,18 @@ Pane a pseudoterminal. +winlink + The link between a {term}`session` and a {term}`window`: the triple + ``(session, index, window)``. + + A window does not live *in* one session; a session holds *links* to + windows, each at an index. ``link-window`` adds another link to the same + window, so one window can be reachable from several sessions -- and even + from one session at two indexes. + + This is what tmux enumerates: a row of ``list-windows`` or + ``list-panes -a`` names a winlink, not a window. See {ref}`winlinks`. + Target A target, cited in the manual as ``[-t target]`` can be a session, window or pane. diff --git a/docs/topics/filtering.md b/docs/topics/filtering.md index 71d56c674..02cbbf4b2 100644 --- a/docs/topics/filtering.md +++ b/docs/topics/filtering.md @@ -105,8 +105,20 @@ Window(@... ..., Session($... ...)) `get()` insists on exactly one result. If the query matches the wrong number of objects, it raises: -- {exc}`~libtmux._internal.query_list.ObjectDoesNotExist` - no matching object found -- {exc}`~libtmux._internal.query_list.MultipleObjectsReturned` - more than one object matches +- {exc}`~libtmux.exc.ObjectDoesNotExist` - no matching object found +- {exc}`~libtmux.exc.MultipleObjectsReturned` - more than one object matches + +Both are {exc}`~libtmux.exc.LibTmuxException`s, so one `except` clause catches +either, and both say what they went looking for: + +```python +>>> from libtmux import exc +>>> try: +... session.windows.get(window_name="nonexistent") +... except exc.LibTmuxException as e: +... print(e) +No objects found: window_name='nonexistent' +``` Pass a `default` to get a fallback value back instead of an exception: @@ -115,6 +127,106 @@ Pass a `default` to get a fallback value back instead of an exception: True ``` +A `default` stands in for an object that is *absent*, so it does not apply when +a query is merely ambiguous — {exc}`~libtmux.exc.MultipleObjectsReturned` is +raised whether or not you passed one. Handing back one of several equally valid +matches is how you end up driving the wrong pane. The next section is about the +one case where an ambiguous match is routine rather than a mistake. + +(winlinks)= + +## When one window is in two sessions + +A server-wide collection — {attr}`server.windows ` and +{attr}`server.panes ` — does not enumerate windows. It +enumerates {term}`winlinks `: the `(session, index, window)` edges tmux +actually stores. Nearly always there is exactly one edge per window and you never +notice the difference. + +Sharing a window adds edges. `link-window` does it explicitly, and a grouped +session (`tmux new-session -t existing`, the mechanism behind [tmuxp](https://tmuxp.git-pull.com/)'s session +groups) does it for every window at once. The window is then genuinely reachable +from each session that links it, and a server-wide listing reports it once per +edge: + +```python +>>> home = server.new_session(session_name="home") +>>> shared = home.new_window(window_name="shared", attach=False) +>>> guest = server.new_session(session_name="guest") +>>> _ = server.cmd( +... "link-window", "-d", "-s", shared.window_id, "-t", f"{guest.session_id}:" +... ) + +>>> len(server.windows.filter(window_id=shared.window_id)) +2 +``` + +Two rows for one `window_id` is not a miscount — it is the shape of the data. +The window is in both sessions, and a listing that collapsed the rows would be +throwing away the very fact you need. + +The consequence is that a *point lookup* against a server-wide collection can be +ambiguous, and says so rather than guessing: + +```python +>>> from libtmux import exc +>>> home = server.new_session(session_name="home") +>>> shared = home.new_window(window_name="shared", attach=False) +>>> guest = server.new_session(session_name="guest") +>>> _ = server.cmd( +... "link-window", "-d", "-s", shared.window_id, "-t", f"{guest.session_id}:" +... ) + +>>> try: +... server.windows.get(window_id=shared.window_id) +... except exc.MultipleObjectsReturned as e: +... print(e) # doctest: +ELLIPSIS +Multiple objects returned (2): window_id='@...' +``` + +### Which sessions hold it? + +{attr}`Window.linked_sessions ` answers directly, +listing each holding session once however many indexes it links the window at: + +```python +>>> home = server.new_session(session_name="home") +>>> shared = home.new_window(window_name="shared", attach=False) +>>> guest = server.new_session(session_name="guest") +>>> _ = server.cmd( +... "link-window", "-d", "-s", shared.window_id, "-t", f"{guest.session_id}:" +... ) + +>>> sorted(s.session_name for s in shared.linked_sessions) +['guest', 'home'] +``` + +### Just fetch the object + +When you have an id and want the object, don't scan a listing for it — name it. +{meth}`Pane.from_pane_id ` and +{meth}`Window.from_window_id ` hand the id to tmux +with a `-t` target, and tmux always resolves it to exactly one object — the same +one it would act on if you typed the command yourself. They cannot be ambiguous, +so they are the right tool for a lookup by id: + +```python +>>> from libtmux.window import Window +>>> home = server.new_session(session_name="home") +>>> shared = home.new_window(window_name="shared", attach=False) +>>> guest = server.new_session(session_name="guest") +>>> _ = server.cmd( +... "link-window", "-d", "-s", shared.window_id, "-t", f"{guest.session_id}:" +... ) + +>>> Window.from_window_id(server, shared.window_id).window_id == shared.window_id +True +``` + +Reserve the server-wide collections for what they are good at — sweeping the +whole server — and reach for them with {meth}`~libtmux._internal.query_list.QueryList.filter`, +which is happy to return two rows, rather than `get()`, which is not. + ## Chaining filters You can stack conditions two ways, and both narrow with AND. Pass several diff --git a/src/libtmux/_internal/query_list.py b/src/libtmux/_internal/query_list.py index 7eeb1ec7d..e28ea42e0 100644 --- a/src/libtmux/_internal/query_list.py +++ b/src/libtmux/_internal/query_list.py @@ -12,6 +12,11 @@ import typing as t from collections.abc import Callable, Iterable, Mapping, Sequence +from libtmux.exc import ( + MultipleObjectsReturned as MultipleObjectsReturned, + ObjectDoesNotExist as ObjectDoesNotExist, +) + logger = logging.getLogger(__name__) if t.TYPE_CHECKING: @@ -33,14 +38,6 @@ def __call__( no_arg = object() -class MultipleObjectsReturned(Exception): - """The query returned multiple objects when only one was expected.""" - - -class ObjectDoesNotExist(Exception): - """The requested object does not exist.""" - - def keygetter( obj: Mapping[str, t.Any], path: str, @@ -558,17 +555,65 @@ def get( default: t.Any | None = no_arg, **kwargs: t.Any, ) -> T | None: - """Retrieve one object. - - Raises :exc:`MultipleObjectsReturned` if multiple objects found. - - Raises :exc:`ObjectDoesNotExist` if no object found, unless ``default`` stated. + """Retrieve exactly one object. + + Parameters + ---------- + matcher : :class:`~collections.abc.Callable` or object, optional + Same matcher :meth:`filter` takes. + default : object, optional + Returned instead of raising when *nothing* matches. + **kwargs : object + Lookups, as passed to :meth:`filter`. + + Returns + ------- + object + The single match, or *default* when there is no match and one was + given. + + Raises + ------ + :exc:`~libtmux.exc.ObjectDoesNotExist` + When nothing matches and no *default* was passed. + :exc:`~libtmux.exc.MultipleObjectsReturned` + When more than one object matches. A *default* does not suppress + this: it stands in for an object that is absent, and an ambiguous + lookup is not an absent one. + + Examples + -------- + >>> from libtmux._internal.query_list import QueryList + >>> from libtmux import exc + >>> qs = QueryList([{"pane_id": "%0"}, {"pane_id": "%0"}, {"pane_id": "%1"}]) + + >>> qs.get(pane_id="%1") + {'pane_id': '%1'} + + A lookup that matches nothing says what it went looking for: + + >>> try: + ... qs.get(pane_id="%9") + ... except exc.ObjectDoesNotExist as e: + ... print(e) + No objects found: pane_id='%9' + + >>> qs.get(pane_id="%9", default=None) is None + True + + A lookup that matches several says how many, and for what: + + >>> try: + ... qs.get(pane_id="%0") + ... except exc.MultipleObjectsReturned as e: + ... print(e) + Multiple objects returned (2): pane_id='%0' """ objs = self.filter(matcher=matcher, **kwargs) if len(objs) > 1: - raise MultipleObjectsReturned + raise MultipleObjectsReturned(count=len(objs), query=kwargs) if len(objs) == 0: - if default == no_arg: - raise ObjectDoesNotExist + if default is no_arg: + raise ObjectDoesNotExist(query=kwargs) return default return objs[0] diff --git a/src/libtmux/exc.py b/src/libtmux/exc.py index e4337c63c..4d34a2d35 100644 --- a/src/libtmux/exc.py +++ b/src/libtmux/exc.py @@ -9,12 +9,28 @@ import typing as t -from libtmux._internal.query_list import ObjectDoesNotExist - if t.TYPE_CHECKING: from libtmux.neo import ListExtraArgs +def _format_query(query: t.Mapping[str, t.Any]) -> str: + """Render a :meth:`QueryList.get` lookup back as ``key=value`` text. + + Examples + -------- + >>> from libtmux.exc import _format_query + >>> _format_query({"pane_id": "%0"}) + "pane_id='%0'" + + >>> _format_query({"window_name": "shared", "window_index": "1"}) + "window_name='shared', window_index='1'" + + >>> _format_query({}) + '' + """ + return ", ".join(f"{key}={value!r}" for key, value in query.items()) + + class LibTmuxException(Exception): """Base Exception for libtmux Errors. @@ -134,8 +150,141 @@ def __init__( ) +class ObjectDoesNotExist(LibTmuxException): + """A lookup expected one object and matched none. + + Raised by :meth:`~libtmux._internal.query_list.QueryList.get` when nothing + matches and no ``default`` was passed. + + Parameters + ---------- + *args : object + A ready-made message, forwarded to :class:`LibTmuxException`. When + omitted, the message is built from *query*. + query : :class:`~collections.abc.Mapping`, optional + The lookup that matched nothing, e.g. ``{"pane_id": "%99"}``. + + Examples + -------- + >>> from libtmux import exc + >>> str(exc.ObjectDoesNotExist()) + 'No objects found' + + A lookup that named what it wanted says so: + + >>> str(exc.ObjectDoesNotExist(query={"pane_id": "%99"})) + "No objects found: pane_id='%99'" + + It is part of the :exc:`LibTmuxException` hierarchy, so + ``except LibTmuxException`` catches it: + + >>> issubclass(exc.ObjectDoesNotExist, exc.LibTmuxException) + True + + .. versionchanged:: 0.62 + + Re-based on :exc:`LibTmuxException` and given a message. + """ + + def __init__( + self, + *args: object, + query: t.Mapping[str, t.Any] | None = None, + ) -> None: + self.query: t.Mapping[str, t.Any] | None = query + if args: + super().__init__(*args) + return + msg = "No objects found" + if query: + msg += f": {_format_query(query)}" + super().__init__(msg) + + +class MultipleObjectsReturned(LibTmuxException): + """A lookup expected one object and matched several. + + Raised by :meth:`~libtmux._internal.query_list.QueryList.get`. Unlike + :exc:`ObjectDoesNotExist`, a ``default`` does **not** suppress it: a + ``default`` is a stand-in for an object that is *absent*, and an ambiguous + lookup is not an absent one. Silently answering with one of several equally + valid matches is how you end up driving the wrong pane. + + On a server-wide collection, several matches for a single id is ordinary + and means the window is linked into more than one session. See + :ref:`winlinks` for what to do about it. + + Parameters + ---------- + *args : object + A ready-made message, forwarded to :class:`LibTmuxException`. When + omitted, the message is built from *count* and *query*. + count : int, optional + How many objects the lookup matched. + query : :class:`~collections.abc.Mapping`, optional + The lookup that matched them, e.g. ``{"pane_id": "%0"}``. + + Examples + -------- + >>> from libtmux import exc + >>> str(exc.MultipleObjectsReturned()) + 'Multiple objects returned' + + A lookup that matched too much reports how much, and for what: + + >>> str(exc.MultipleObjectsReturned(count=2, query={"pane_id": "%0"})) + "Multiple objects returned (2): pane_id='%0'" + + It is part of the :exc:`LibTmuxException` hierarchy, so + ``except LibTmuxException`` catches it: + + >>> issubclass(exc.MultipleObjectsReturned, exc.LibTmuxException) + True + + .. versionadded:: 0.62 + + Added to :mod:`libtmux.exc` as a :exc:`LibTmuxException` subclass with + a message. + """ + + def __init__( + self, + *args: object, + count: int | None = None, + query: t.Mapping[str, t.Any] | None = None, + ) -> None: + self.count: int | None = count + self.query: t.Mapping[str, t.Any] | None = query + if args: + super().__init__(*args) + return + msg = "Multiple objects returned" + if count is not None: + msg += f" ({count})" + if query: + msg += f": {_format_query(query)}" + super().__init__(msg) + + class TmuxObjectDoesNotExist(ObjectDoesNotExist): - """The query returned multiple objects when only one was expected.""" + """tmux has no object with the id that was asked for. + + Examples + -------- + >>> from libtmux import exc + >>> str(exc.TmuxObjectDoesNotExist()) + 'Could not find object' + + >>> str( + ... exc.TmuxObjectDoesNotExist( + ... obj_key="pane_id", + ... obj_id="%99", + ... list_cmd="list-panes", + ... list_extra_args=("-t", "%99"), + ... ) + ... ) + "Could not find pane_id=%99 for list-panes ('-t', '%99')" + """ def __init__( self, diff --git a/src/libtmux/neo.py b/src/libtmux/neo.py index 080a95669..7a7d1a46a 100644 --- a/src/libtmux/neo.py +++ b/src/libtmux/neo.py @@ -823,6 +823,77 @@ def _is_target_not_found_error(stderr_text: str) -> bool: return "can't find " in stderr_text +def _best_winlink(rows: OutputsRaw) -> OutputRaw: + """Pick the winlink row tmux would select. + + A ``list-windows`` listing enumerates :term:`winlinks ` -- + ``(session, index, window)`` edges -- not windows. ``link-window`` can attach + one window to a session at several indexes at once, so the same + ``window_id`` may appear on several rows, each with a different + ``window_index``. + + tmux selects the current winlink when it contains the window, otherwise the + first. ``#{window_active}`` identifies the current row, and the lowest + ``window_index`` is tmux's first -- chosen explicitly here, so the caller + need not pre-sort the rows. + + Parameters + ---------- + rows : OutputsRaw + Non-empty rows for one object id in one session. Order does not + matter: the current winlink wins, otherwise the lowest + ``window_index``. + + Returns + ------- + OutputRaw + The row naming the winlink tmux would act on. + + Examples + -------- + One row is the whole answer: + + >>> from libtmux.neo import _best_winlink + >>> _best_winlink([{"window_id": "@0", "window_index": "1"}])["window_index"] + '1' + + A window linked into one session twice gives two rows. When the session is + sitting on the higher-indexed link, that is the one tmux acts on: + + >>> _best_winlink([ + ... {"window_id": "@0", "window_index": "1", "window_active": "0"}, + ... {"window_id": "@0", "window_index": "5", "window_active": "1"}, + ... ])["window_index"] + '5' + + When the session is sitting on some *other* window, neither link is current, + and tmux falls back to the first: + + >>> _best_winlink([ + ... {"window_id": "@0", "window_index": "1", "window_active": "0"}, + ... {"window_id": "@0", "window_index": "5", "window_active": "0"}, + ... ])["window_index"] + '1' + + The fallback reads the lowest index, not the first row, so a listing that + happened to arrive high-index-first still answers tmux's first: + + >>> _best_winlink([ + ... {"window_id": "@0", "window_index": "5", "window_active": "0"}, + ... {"window_id": "@0", "window_index": "1", "window_active": "0"}, + ... ])["window_index"] + '1' + """ + if len(rows) == 1: + return rows[0] + + for row in rows: + if row.get("window_active") == "1": + return row + + return min(rows, key=lambda row: int(row["window_index"])) + + def fetch_obj( server: Server, obj_key: str, @@ -832,6 +903,10 @@ def fetch_obj( ) -> OutputRaw: """Fetch the single ``list-*`` row whose *obj_key* equals *obj_id*. + A listing enumerates :term:`winlinks `, so a window linked into one + session at two indexes matches twice. :func:`_best_winlink` then picks the + row tmux itself would act on, rather than whichever sorted last. + Parameters ---------- server : :class:`~libtmux.server.Server` @@ -908,12 +983,9 @@ def fetch_obj( list_extra_args=list_extra_args, ) from e - obj = None - for _obj in obj_formatters_filtered: - if _obj.get(obj_key) == obj_id: - obj = _obj + matches = [row for row in obj_formatters_filtered if row.get(obj_key) == obj_id] - if obj is None: + if not matches: raise exc.TmuxObjectDoesNotExist( obj_key=obj_key, obj_id=obj_id, @@ -921,4 +993,4 @@ def fetch_obj( list_extra_args=list_extra_args, ) - return obj + return _best_winlink(matches) diff --git a/src/libtmux/window.py b/src/libtmux/window.py index 2a7e624a1..8a7980d98 100644 --- a/src/libtmux/window.py +++ b/src/libtmux/window.py @@ -283,6 +283,82 @@ def session(self) -> Session: return Session.from_session_id(server=self.server, session_id=self.session_id) + @property + def linked_sessions(self) -> QueryList[Session]: + """Every session this window is reachable from. + + Usually one, and then this is just :attr:`Window.session` in a list. + ``link-window`` and grouped sessions (``tmux new-session -t existing``) + make it more: the window is genuinely in each of them at once, and + :attr:`Window.session` returns the session recorded on this + :class:`Window` instance. + + Each session is listed once, however many indexes it links the window + at, and they come back in the order tmux lists them. Fetching the + holders takes two list commands total, independent of how many there + are. + If either listing fails, the result is empty. + + Returns + ------- + :class:`~libtmux._internal.query_list.QueryList` of :class:`~libtmux.Session` + + See Also + -------- + :attr:`Window.session` : the session recorded on this window instance. + + Examples + -------- + A window you just made belongs to the session you made it in: + + >>> window = session.new_window(window_name="solo", attach=False) + >>> [s.session_name for s in window.linked_sessions] == [session.session_name] + True + + Link it into a second session and it belongs to both: + + >>> guest = server.new_session(session_name="guest") + >>> target = f"{guest.session_id}:" + >>> _ = server.cmd("link-window", "-d", "-s", window.window_id, "-t", target) + >>> sorted(s.session_name for s in window.linked_sessions) == sorted( + ... [session.session_name, "guest"] + ... ) + True + + .. versionadded:: 0.62 + """ + from libtmux.session import Session + + try: + window_rows = fetch_objs( + server=self.server, + list_cmd="list-windows", + list_extra_args=("-a",), + ) + session_ids = dict.fromkeys( + row["session_id"] + for row in window_rows + if row.get("window_id") == self.window_id and row.get("session_id") + ) + session_rows = fetch_objs( + server=self.server, + list_cmd="list-sessions", + ) + except exc.LibTmuxException: + return QueryList([]) + + sessions_by_id = { + row["session_id"]: row for row in session_rows if row.get("session_id") + } + + return QueryList( + [ + Session(server=self.server, **sessions_by_id[session_id]) + for session_id in session_ids + if session_id in sessions_by_id + ], + ) + @property def panes(self) -> QueryList[Pane]: """Panes contained by window. diff --git a/tests/_internal/test_query_list.py b/tests/_internal/test_query_list.py index 9559be963..7d4a306c4 100644 --- a/tests/_internal/test_query_list.py +++ b/tests/_internal/test_query_list.py @@ -249,8 +249,10 @@ class Obj: ([1, 2, 3, 4, 5], None, QueryList([1, 2, 3, 4, 5])), ([1, 2, 3, 4, 5], [1], QueryList([1])), ([1, 2, 3, 4, 5], [1, 4], QueryList([1, 4])), + ([1, 2, 3, 4, 5], [9], QueryList([])), ([1, 2, 3, 4, 5], lambda val: val == 1, QueryList([1])), ([1, 2, 3, 4, 5], lambda val: val == 2, QueryList([2])), + ([1, 2, 3, 4, 5], lambda val: val == 9, QueryList([])), ], ) def test_filter( @@ -267,27 +269,77 @@ def test_filter( else: assert qs.filter() == expected_result - if ( - isinstance(expected_result, list) - and len(expected_result) > 0 - and not isinstance(expected_result[0], dict) - ): - if len(expected_result) == 1: + if not isinstance(expected_result, list): + return + + # ``get()`` insists on exactly one match. Assert the message too: an + # exception nobody can read is the defect, not just the raise. + if len(expected_result) == 0: + with pytest.raises(ObjectDoesNotExist) as missing: + if isinstance(filter_expr, dict): + qs.get(**filter_expr) + else: + qs.get(filter_expr) + assert missing.match("No objects found") + return + + if isinstance(expected_result[0], dict): + return + + if len(expected_result) == 1: + if isinstance(filter_expr, dict): + assert qs.get(**filter_expr) == expected_result[0] + else: + assert qs.get(filter_expr) == expected_result[0] + else: + with pytest.raises(MultipleObjectsReturned) as multiple: if isinstance(filter_expr, dict): - assert qs.get(**filter_expr) == expected_result[0] + qs.get(**filter_expr) else: - assert qs.get(filter_expr) == expected_result[0] - elif len(expected_result) > 1: - with pytest.raises(MultipleObjectsReturned) as e: - if isinstance(filter_expr, dict): - assert qs.get(**filter_expr) == expected_result - else: - assert qs.get(filter_expr) == expected_result - assert e.match("Multiple objects returned") - elif len(expected_result) == 0: - with pytest.raises(ObjectDoesNotExist) as exc: - if isinstance(filter_expr, dict): - assert qs.get(**filter_expr) == expected_result - else: - assert qs.get(filter_expr) == expected_result - assert exc.match("No objects found") + qs.get(filter_expr) + assert multiple.match("Multiple objects returned") + + +def test_get_multiple_objects_exposes_ambiguity_data() -> None: + query: dict[str, t.Any] = {"fruit": "apple"} + qs = QueryList([query, query]) + + with pytest.raises(MultipleObjectsReturned) as multiple: + qs.get(**query) + + assert multiple.value.count == 2 + assert multiple.value.query == query + + +def test_query_exception_annotations_resolve() -> None: + """Runtime introspection resolves the query annotations.""" + for initializer in ( + ObjectDoesNotExist.__init__, + MultipleObjectsReturned.__init__, + ): + assert "query" in t.get_type_hints(initializer) + + +def test_get_default_with_broad_eq_is_returned() -> None: + """A default whose ``__eq__`` is non-identity is returned, not raised. + + ``get`` guards the "nothing matched" branch with the ``no_arg`` sentinel. + Comparing the default to it with ``==`` rather than ``is`` misfires when the + default answers ``__eq__`` truthily to an unrelated object. + """ + + class BroadEq: + def __eq__(self, other: object) -> bool: + return True + + __hash__ = None # type: ignore[assignment] + + default = BroadEq() + assert QueryList([]).get(missing="x", default=default) is default + + +def test_object_does_not_exist_exposes_query() -> None: + """``ObjectDoesNotExist`` exposes its lookup, mirroring its sibling.""" + query: dict[str, t.Any] = {"pane_id": "%9"} + assert ObjectDoesNotExist(query=query).query == query + assert ObjectDoesNotExist().query is None diff --git a/tests/test_resolution.py b/tests/test_resolution.py index 0b7ecb3fd..28604d2da 100644 --- a/tests/test_resolution.py +++ b/tests/test_resolution.py @@ -17,7 +17,7 @@ import pytest -from libtmux import exc +from libtmux import exc, neo, window as window_module from libtmux._internal.query_list import ObjectDoesNotExist from libtmux.pane import Pane from libtmux.window import Window @@ -177,3 +177,388 @@ def test_dead_server_raises_libtmux_exception( Pane.from_pane_id(server=server, pane_id=pane_id) assert not isinstance(excinfo.value, ObjectDoesNotExist) + + +def tmux_resolves_index_to(server: Server, target: str) -> str: + """Ask tmux which window index it would act on for *target*.""" + result = server.cmd("display-message", "-p", "-t", target, "#{window_index}") + return result.stdout[0] + + +def link_window_twice_into_one_session(server: Server) -> tuple[Window, Session]: + """Link one window into a *single* session at two indexes. + + ``dup`` ends up holding ``@0`` at index 1 and again at index 5, plus an + unrelated window at index 2 to park the cursor on. Two winlinks, one window, + one session -- so scoping the listing with ``-t`` does not narrow it to one + row, and a resolver reading the last row answers 5. + """ + dup = server.new_session(session_name="dup") + + window = dup.active_window + assert window.window_id is not None + + dup.new_window(window_name="other", attach=False) + server.cmd("link-window", "-d", "-s", window.window_id, "-t", "dup:5") + + return window, dup + + +class WinlinkFixture(t.NamedTuple): + """A window reachable at more than one winlink.""" + + test_id: str + shape: str + #: Window index to make current before resolving, if any. + select: str | None + #: Sessions the window should be reachable from. + expected_holders: int + + +WINLINK_FIXTURES: list[WinlinkFixture] = [ + WinlinkFixture( + test_id="same_session_low_link_current", + shape="same-session", + select="dup:1", + expected_holders=1, + ), + WinlinkFixture( + test_id="same_session_high_link_current", + shape="same-session", + select="dup:5", + expected_holders=1, + ), + WinlinkFixture( + test_id="same_session_other_window_current", + shape="same-session", + select="dup:2", + expected_holders=1, + ), + WinlinkFixture( + test_id="cross_session", + shape="cross-session", + select=None, + expected_holders=2, + ), + WinlinkFixture( + test_id="grouped_session", + shape="grouped", + select=None, + expected_holders=2, + ), +] + + +def build_winlink_shape(server: Server, shape: str) -> Window: + """Create *shape* and hand back the multiply-linked window.""" + if shape == "same-session": + window, _dup = link_window_twice_into_one_session(server) + return window + if shape == "cross-session": + window, _old, _recent = link_window_into_second_session(server) + return window + if shape == "grouped": + # ``new-session -t`` is a grouped session: it shares the origin's + # windows. This is what tmuxp's session groups are built on. + origin = server.new_session(session_name="origin") + server.cmd("new-session", "-d", "-t", "origin", "-s", "grouped") + return origin.active_window + msg = f"unknown shape: {shape}" + raise AssertionError(msg) + + +@pytest.mark.parametrize( + list(WinlinkFixture._fields), + WINLINK_FIXTURES, + ids=[test.test_id for test in WINLINK_FIXTURES], +) +def test_multiply_linked_window_resolves_like_tmux( + server: Server, + test_id: str, + shape: str, + select: str | None, + expected_holders: int, +) -> None: + """A window at several winlinks resolves to the one tmux would act on. + + tmux's rule, from ``cmd_find_best_winlink_with_window`` in ``cmd-find.c``: + "the current if it contains the window, otherwise the first". Reading the + last row of the listing instead answers with the highest index, which is + what libtmux did. + """ + window = build_winlink_shape(server, shape) + assert window.window_id is not None + + if select is not None: + server.cmd("select-window", "-t", select) + + canonical_session = tmux_resolves_to(server, window.window_id) + canonical_index = tmux_resolves_index_to(server, window.window_id) + + resolved = Window.from_window_id(server=server, window_id=window.window_id) + assert resolved.window_index == canonical_index + assert resolved.session_id == canonical_session + + window.refresh() + assert window.window_index == canonical_index + assert window.session_id == canonical_session + + assert len(window.linked_sessions) == expected_holders + + +def test_same_session_double_link_answers_the_low_index(server: Server) -> None: + """The regression in its narrowest form: tmux says 1, libtmux said 5.""" + window, _dup = link_window_twice_into_one_session(server) + assert window.window_id is not None + + server.cmd("select-window", "-t", "dup:1") + + assert tmux_resolves_index_to(server, window.window_id) == "1" + assert Window.from_window_id(server, window.window_id).window_index == "1" + + +def test_pane_of_multiply_linked_window_agrees_with_its_window(server: Server) -> None: + """:meth:`Pane.from_pane_id` and :meth:`Window.from_window_id` do not disagree. + + ``list-panes`` emits each pane once, so the pane side was already right. The + point is that the window side now matches it. + """ + window, _dup = link_window_twice_into_one_session(server) + pane = window.active_pane + assert pane is not None + assert pane.pane_id is not None + assert window.window_id is not None + + server.cmd("select-window", "-t", "dup:1") + + resolved_pane = Pane.from_pane_id(server=server, pane_id=pane.pane_id) + resolved_window = Window.from_window_id(server=server, window_id=window.window_id) + + assert resolved_pane.window_index == resolved_window.window_index + assert resolved_pane.window_index == tmux_resolves_index_to(server, pane.pane_id) + + +@pytest.mark.parametrize( + list(WinlinkFixture._fields), + WINLINK_FIXTURES, + ids=[test.test_id for test in WINLINK_FIXTURES], +) +def test_server_collections_enumerate_winlinks( + server: Server, + test_id: str, + shape: str, + select: str | None, + expected_holders: int, +) -> None: + """A server-wide listing yields one row per winlink, and that is the truth. + + :attr:`Server.panes` and :attr:`Server.windows` list with ``-a``, which tmux + answers with one row per ``(session, index, window)`` edge. A window at two + winlinks therefore appears twice -- it really is reachable two ways. + """ + window = build_winlink_shape(server, shape) + assert window.window_id is not None + pane = window.active_pane + assert pane is not None + assert pane.pane_id is not None + + window_rows = server.windows.filter(window_id=window.window_id) + pane_rows = server.panes.filter(pane_id=pane.pane_id) + + assert len(window_rows) == 2 + assert len(pane_rows) == 2 + + # Every row names the same window, at a different winlink. + assert {row.window_id for row in window_rows} == {window.window_id} + winlinks = {(row.session_id, row.window_index) for row in window_rows} + assert len(winlinks) == 2 + + +@pytest.mark.parametrize( + list(WinlinkFixture._fields), + WINLINK_FIXTURES, + ids=[test.test_id for test in WINLINK_FIXTURES], +) +def test_ambiguous_point_lookup_is_catchable_and_legible( + server: Server, + test_id: str, + shape: str, + select: str | None, + expected_holders: int, +) -> None: + """An ambiguous ``get()`` explains itself, and ``except LibTmuxException`` sees it. + + Two rows for one pane id is a real ambiguity, and a ``default`` does not + make it go away -- it stands in for an object that is *absent*. So the + lookup raises, but it raises something the caller can catch and read. + """ + window = build_winlink_shape(server, shape) + pane = window.active_pane + assert pane is not None + assert pane.pane_id is not None + + with pytest.raises(exc.LibTmuxException) as excinfo: + server.panes.get(pane_id=pane.pane_id, default=None) + + assert isinstance(excinfo.value, exc.MultipleObjectsReturned) + assert str(excinfo.value) == ( + f"Multiple objects returned (2): pane_id={pane.pane_id!r}" + ) + + +def test_point_lookup_by_id_is_the_way_out(server: Server) -> None: + """The documented escape from an ambiguous scan: name the id with ``-t``. + + :meth:`Pane.from_pane_id` and :meth:`Window.from_window_id` hand the question + to tmux, which always has exactly one answer. + """ + window, _old, _recent = link_window_into_second_session(server) + pane = window.active_pane + assert pane is not None + assert pane.pane_id is not None + assert window.window_id is not None + + assert Pane.from_pane_id(server, pane.pane_id).pane_id == pane.pane_id + assert Window.from_window_id(server, window.window_id).window_id == window.window_id + + +def test_linked_sessions_lists_each_holder_once(server: Server) -> None: + """:attr:`Window.linked_sessions` dedupes winlinks down to sessions. + + A window linked into one session twice is in *one* session, not two -- so + the two winlinks collapse to one holder. + """ + window, dup = link_window_twice_into_one_session(server) + + assert [s.session_id for s in window.linked_sessions] == [dup.session_id] + + guest = server.new_session(session_name="guest") + server.cmd( + "link-window", "-d", "-s", window.window_id, "-t", f"{guest.session_id}:" + ) + + holders = {s.session_id for s in window.linked_sessions} + assert holders == {dup.session_id, guest.session_id} + + +def test_linked_sessions_returns_empty_after_server_kill( + server: Server, + session: Session, +) -> None: + """A retained window has no knowable holders after its server dies.""" + window = session.active_window + + server.kill() + + assert list(window.linked_sessions) == [] + + +def test_linked_sessions_returns_empty_when_session_snapshot_fails( + server: Server, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The second listing shares the accessor's lenient error boundary. + + A live server cannot naturally fail only its second back-to-back listing, + so this exceptional boundary needs a narrow wrapper around the real first + snapshot. + """ + window = server.new_session(session_name="holder").active_window + real_fetch_objs = neo.fetch_objs + + def fail_session_snapshot(**kwargs: t.Any) -> list[dict[str, str]]: + if kwargs["list_cmd"] == "list-sessions": + msg = "session snapshot failed" + raise exc.LibTmuxException(msg) + return real_fetch_objs(**kwargs) + + monkeypatch.setattr(window_module, "fetch_objs", fail_session_snapshot) + + assert list(window.linked_sessions) == [] + + +def test_linked_sessions_skips_holder_lost_between_snapshots( + server: Server, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A holder disappearing between real snapshots is skipped safely. + + The race cannot be scheduled deterministically without wrapping the first + listing, so the standard server fixture supplies both live snapshots while + the wrapper removes one holder between them. + """ + home = server.new_session(session_name="home") + window = home.active_window + guest = server.new_session(session_name="guest") + server.cmd( + "link-window", + "-d", + "-s", + window.window_id, + "-t", + f"{guest.session_id}:", + ) + real_fetch_objs = neo.fetch_objs + + def remove_guest_after_window_snapshot(**kwargs: t.Any) -> list[dict[str, str]]: + rows = real_fetch_objs(**kwargs) + if kwargs["list_cmd"] == "list-windows": + guest.kill() + return rows + + monkeypatch.setattr( + window_module, + "fetch_objs", + remove_guest_after_window_snapshot, + ) + + assert [holder.session_id for holder in window.linked_sessions] == [ + home.session_id, + ] + + +def test_linked_sessions_uses_two_list_calls( + server: Server, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Resolving several holders uses two real tmux listings in total. + + ``fetch_objs`` is wrapped only to record call shape; the real function and + tmux server still provide every row used by the assertion. + """ + window, _dup = link_window_twice_into_one_session(server) + for name in ("guest-one", "guest-two"): + guest = server.new_session(session_name=name) + server.cmd( + "link-window", + "-d", + "-s", + window.window_id, + "-t", + f"{guest.session_id}:", + ) + + real_fetch_objs = neo.fetch_objs + rows = real_fetch_objs( + server=server, + list_cmd="list-windows", + list_extra_args=("-a",), + ) + expected_session_ids = list( + dict.fromkeys( + row["session_id"] + for row in rows + if row.get("window_id") == window.window_id and row.get("session_id") + ), + ) + list_calls: list[str] = [] + + def recording_fetch_objs(**kwargs: t.Any) -> list[dict[str, str]]: + list_calls.append(kwargs["list_cmd"]) + return real_fetch_objs(**kwargs) + + monkeypatch.setattr(window_module, "fetch_objs", recording_fetch_objs) + monkeypatch.setattr(neo, "fetch_objs", recording_fetch_objs) + + assert [item.session_id for item in window.linked_sessions] == expected_session_ids + assert list_calls == ["list-windows", "list-sessions"]