Skip to content

Commit 9da278a

Browse files
tonytony
committed
tests(race-conditions): Add xfail tests for create-query race conditions
why: Document known vulnerability where two-step "create → query" pattern in new_session(), new_window(), and split() can fail if tmux state changes between the creation command and the subsequent list-* query (#624). what: - Add tests/test_race_conditions.py with 6 strict xfail tests - 3 tests for server crash between create and query (Mode A) - 3 tests for stale empty list-* response (Mode B) - All tests are parallel-safe (unique sockets, function-scoped monkeypatch) - Tests will flip to XPASS when creation methods construct objects directly from -P output instead of re-querying
1 parent cb21427 commit 9da278a

1 file changed

Lines changed: 251 additions & 0 deletions

File tree

‎tests/test_race_conditions.py‎

Lines changed: 251 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,251 @@
1+
"""Race condition tests for object creation methods (#624).
2+
3+
Each test deterministically reproduces a failure mode where the two-step
4+
"create -> query" pattern in new_session(), new_window(), and split() fails.
5+
6+
All tests are strict xfail -- they document known vulnerabilities and will
7+
flip to XPASS when the creation methods are changed to construct objects
8+
directly from -P output.
9+
10+
Parallel-safe: each test uses the server/session fixture (unique socket)
11+
and function-scoped monkeypatch.
12+
"""
13+
14+
from __future__ import annotations
15+
16+
import typing as t
17+
18+
import pytest
19+
20+
from libtmux import exc
21+
from libtmux.server import Server
22+
23+
if t.TYPE_CHECKING:
24+
from libtmux.neo import ListCmd, ListExtraArgs
25+
from libtmux.session import Session
26+
27+
28+
# --- Failure Mode A: Server crash between create and query ---
29+
30+
31+
@pytest.mark.xfail(
32+
raises=exc.TmuxObjectDoesNotExist,
33+
reason="new_session() re-queries via list-sessions after new-session returns. "
34+
"Server crash between steps loses the session. "
35+
"See https://github.com/tmux-python/libtmux/issues/624",
36+
strict=True,
37+
)
38+
def test_new_session_server_crash(
39+
server: Server,
40+
monkeypatch: pytest.MonkeyPatch,
41+
) -> None:
42+
"""Server crash between new-session and list-sessions (#624)."""
43+
from libtmux import neo
44+
45+
original_fetch_objs = neo.fetch_objs
46+
server_crashed = False
47+
48+
def fetch_objs_with_crash(
49+
server: Server,
50+
list_cmd: ListCmd,
51+
list_extra_args: ListExtraArgs = None,
52+
) -> list[dict[str, t.Any]]:
53+
nonlocal server_crashed
54+
if list_cmd == "list-sessions" and not server_crashed:
55+
server_crashed = True
56+
server.cmd("kill-server")
57+
server.cmd("new-session", "-d", "-s", "_replacement")
58+
return original_fetch_objs(
59+
server=server,
60+
list_cmd=list_cmd,
61+
list_extra_args=list_extra_args,
62+
)
63+
64+
# Bumper session ensures race_test gets $1+, replacement server gets $0
65+
server.cmd("new-session", "-d", "-s", "_bumper")
66+
monkeypatch.setattr(neo, "fetch_objs", fetch_objs_with_crash)
67+
server.new_session(session_name="race_test")
68+
69+
70+
@pytest.mark.xfail(
71+
raises=exc.TmuxObjectDoesNotExist,
72+
reason="new_window() re-queries via list-windows after new-window returns. "
73+
"Server crash between steps loses the window.",
74+
strict=True,
75+
)
76+
def test_new_window_server_crash(
77+
session: Session,
78+
monkeypatch: pytest.MonkeyPatch,
79+
) -> None:
80+
"""Server crash between new-window and list-windows."""
81+
from libtmux import neo
82+
83+
original_fetch_objs = neo.fetch_objs
84+
server_crashed = False
85+
86+
def fetch_objs_with_crash(
87+
server: Server,
88+
list_cmd: ListCmd,
89+
list_extra_args: ListExtraArgs = None,
90+
) -> list[dict[str, t.Any]]:
91+
nonlocal server_crashed
92+
if list_cmd == "list-windows" and not server_crashed:
93+
server_crashed = True
94+
server.cmd("kill-server")
95+
server.cmd("new-session", "-d", "-s", "_replacement")
96+
return original_fetch_objs(
97+
server=server,
98+
list_cmd=list_cmd,
99+
list_extra_args=list_extra_args,
100+
)
101+
102+
monkeypatch.setattr(neo, "fetch_objs", fetch_objs_with_crash)
103+
session.new_window(window_name="race_test")
104+
105+
106+
@pytest.mark.xfail(
107+
raises=exc.TmuxObjectDoesNotExist,
108+
reason="split() re-queries via list-panes after split-window returns. "
109+
"Server crash between steps loses the pane.",
110+
strict=True,
111+
)
112+
def test_split_server_crash(
113+
session: Session,
114+
monkeypatch: pytest.MonkeyPatch,
115+
) -> None:
116+
"""Server crash between split-window and list-panes."""
117+
from libtmux import neo
118+
119+
original_fetch_objs = neo.fetch_objs
120+
server_crashed = False
121+
122+
def fetch_objs_with_crash(
123+
server: Server,
124+
list_cmd: ListCmd,
125+
list_extra_args: ListExtraArgs = None,
126+
) -> list[dict[str, t.Any]]:
127+
nonlocal server_crashed
128+
if list_cmd == "list-panes" and not server_crashed:
129+
server_crashed = True
130+
server.cmd("kill-server")
131+
server.cmd("new-session", "-d", "-s", "_replacement")
132+
return original_fetch_objs(
133+
server=server,
134+
list_cmd=list_cmd,
135+
list_extra_args=list_extra_args,
136+
)
137+
138+
monkeypatch.setattr(neo, "fetch_objs", fetch_objs_with_crash)
139+
pane = session.active_pane
140+
assert pane is not None
141+
pane.split()
142+
143+
144+
# --- Failure Mode B: Stale empty query response ---
145+
146+
147+
@pytest.mark.xfail(
148+
raises=exc.TmuxObjectDoesNotExist,
149+
reason="new_session() re-queries via list-sessions. "
150+
"Stale empty response causes TmuxObjectDoesNotExist.",
151+
strict=True,
152+
)
153+
def test_new_session_stale_list(
154+
server: Server,
155+
monkeypatch: pytest.MonkeyPatch,
156+
) -> None:
157+
"""Empty list-sessions after new-session (#624)."""
158+
from libtmux import neo
159+
160+
original_fetch_objs = neo.fetch_objs
161+
intercepted = False
162+
163+
def fetch_objs_stale(
164+
server: Server,
165+
list_cmd: ListCmd,
166+
list_extra_args: ListExtraArgs = None,
167+
) -> list[dict[str, t.Any]]:
168+
nonlocal intercepted
169+
if list_cmd == "list-sessions" and not intercepted:
170+
intercepted = True
171+
return []
172+
return original_fetch_objs(
173+
server=server,
174+
list_cmd=list_cmd,
175+
list_extra_args=list_extra_args,
176+
)
177+
178+
monkeypatch.setattr(neo, "fetch_objs", fetch_objs_stale)
179+
server.new_session(session_name="race_test")
180+
181+
182+
@pytest.mark.xfail(
183+
raises=exc.TmuxObjectDoesNotExist,
184+
reason="new_window() re-queries via list-windows. "
185+
"Stale empty response causes TmuxObjectDoesNotExist.",
186+
strict=True,
187+
)
188+
def test_new_window_stale_list(
189+
session: Session,
190+
monkeypatch: pytest.MonkeyPatch,
191+
) -> None:
192+
"""Empty list-windows after new-window."""
193+
from libtmux import neo
194+
195+
original_fetch_objs = neo.fetch_objs
196+
intercepted = False
197+
198+
def fetch_objs_stale(
199+
server: Server,
200+
list_cmd: ListCmd,
201+
list_extra_args: ListExtraArgs = None,
202+
) -> list[dict[str, t.Any]]:
203+
nonlocal intercepted
204+
if list_cmd == "list-windows" and not intercepted:
205+
intercepted = True
206+
return []
207+
return original_fetch_objs(
208+
server=server,
209+
list_cmd=list_cmd,
210+
list_extra_args=list_extra_args,
211+
)
212+
213+
monkeypatch.setattr(neo, "fetch_objs", fetch_objs_stale)
214+
session.new_window(window_name="race_test")
215+
216+
217+
@pytest.mark.xfail(
218+
raises=exc.TmuxObjectDoesNotExist,
219+
reason="split() re-queries via list-panes. "
220+
"Stale empty response causes TmuxObjectDoesNotExist.",
221+
strict=True,
222+
)
223+
def test_split_stale_list(
224+
session: Session,
225+
monkeypatch: pytest.MonkeyPatch,
226+
) -> None:
227+
"""Empty list-panes after split-window."""
228+
from libtmux import neo
229+
230+
original_fetch_objs = neo.fetch_objs
231+
intercepted = False
232+
233+
def fetch_objs_stale(
234+
server: Server,
235+
list_cmd: ListCmd,
236+
list_extra_args: ListExtraArgs = None,
237+
) -> list[dict[str, t.Any]]:
238+
nonlocal intercepted
239+
if list_cmd == "list-panes" and not intercepted:
240+
intercepted = True
241+
return []
242+
return original_fetch_objs(
243+
server=server,
244+
list_cmd=list_cmd,
245+
list_extra_args=list_extra_args,
246+
)
247+
248+
monkeypatch.setattr(neo, "fetch_objs", fetch_objs_stale)
249+
pane = session.active_pane
250+
assert pane is not None
251+
pane.split()

0 commit comments

Comments
 (0)