test(process_registry): trim handle-release tests to invariants, fix prune fixture

Drop the no-op and still-running change-detectors (4 invariant tests remain:
pipe closed on finish, PTY closed on finish, poll still serves buffered
output, prune releases handles). The accretion-caps fake session now
carries process/_pty like the real dataclass, since prune reads them.
This commit is contained in:
teknium1
2026-09-13 19:55:47 -07:00
committed by Teknium
parent 29b981c846
commit dcf17633de
2 changed files with 2 additions and 31 deletions
+2
View File
@@ -84,6 +84,8 @@ class TestCompletionConsumedPrune:
self.id = sid
self.started_at = time.time() - (FINISHED_TTL_SECONDS + 100)
self.exited = True
self.process = None # handle release reads the real dataclass fields
self._pty = None
reg._finished["stale-1"] = _FakeSess("stale-1")
reg._completion_consumed.add("stale-1")
-31
View File
@@ -756,14 +756,6 @@ class TestFinishedHandleRelease:
registry._move_to_finished(session)
assert pty_closed["closed"]
def test_move_to_finished_safe_without_handles(self, registry):
"""Env-backed / detached sessions have no local Popen or PTY; the
release must be a no-op, not a crash."""
session = _make_session(sid="proc_no_handles", exited=True)
registry._finished[session.id] = session
registry._move_to_finished(session) # must not raise
assert session.id in registry._finished
def test_poll_still_serves_output_after_handle_release(self, registry):
"""Output remains queryable after the pipes close — poll() reads the
buffered output, never the (now-closed) pipe."""
@@ -820,29 +812,6 @@ class TestFinishedHandleRelease:
assert session.id not in registry._finished
assert pty_closed["closed"], "pruned session must release its PTY handle"
def test_release_does_not_touch_running_sessions(self, registry):
"""A still-running session's handles must remain open: _move_to_finished
is only ever invoked with exited sessions, and prune only walks
_finished — a live session in _running keeps its pipe."""
proc = subprocess.Popen(
[sys.executable, "-c", "import time; time.sleep(5)"],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
stdin=subprocess.DEVNULL,
)
try:
session = _make_session(sid="proc_still_running", exited=False)
session.process = proc
registry._running[session.id] = session
with registry._lock:
registry._prune_if_needed()
assert session.id in registry._running
assert proc.stdout is not None and not proc.stdout.closed
finally:
proc.kill()
proc.wait(timeout=5)
# =========================================================================