fix(tests): clean up gptme_N sessions in tmux test fixture (#926)

TimeToBuildBob · web-flow · commit dbe1676e4465 · 2025-12-04T20:58:56.000+01:00
* fix(tests): clean up gptme_N sessions in tmux test fixture The cleanup fixture only cleaned up sessions matching gptme_{worker_id}_* but new_session() creates sessions named gptme_N (simple numeric suffix). These sessions were not cleaned up between tests, causing failures when sessions from previous tests still existed. This fix extends the cleanup to also remove gptme_N pattern sessions (where N is just digits) in addition to the worker-prefixed ones. Fixes flaky test: TestNewSession::test_increments_session_id * fix(tests): use worker-isolated session in test_lists_created_sessions The test was using new_session() which creates gptme_N format sessions. In parallel testing with pytest-xdist, other workers' cleanup could remove these sessions before the test verifies them, causing race condition failures. Now uses _create_test_session() which creates worker-prefixed sessions (gptme_gw0_1, etc.) that are isolated from other workers' cleanup. * fix(tests): add tmux session cleanup for test_cli::test_tmux Add cleanup_tmux_sessions fixture to prevent cross-test contamination when running tests in parallel. The test_cli::test_tmux test runs the actual gptme CLI which creates gptme_N sessions internally, and these can leak between test runs. Changes: - Add cleanup_tmux_sessions fixture in conftest.py that cleans up gptme_N sessions before and after tests - Apply fixture to test_tmux in test_cli.py - Add xdist_group marker to ensure tmux tests run on same worker Fixes the CI failure where test_tmux was capturing output from a different test run's tmux session. * fix(tests): add xdist_group to all tmux tests for serialization All tmux tests now run on the same worker to prevent race conditions where tests creating gptme_N sessions interfere with each other. The test_cli::test_tmux was still failing because test_tools_tmux.py tests (like test_increments_session_id) create gptme_N sessions without the xdist_group marker, allowing parallel execution that causes session name collisions. By adding xdist_group('tmux') at the module level to test_tools_tmux.py, all tmux-related tests now run serially on the same worker. * Revert "fix(tests): add xdist_group to all tmux tests for serialization" This reverts commit ffc35da. * style(tests): use regex for tmux session pattern matching Use re.compile(r'^gptme_\d+$') for cleaner session name matching instead of string splitting. Improves readability as suggested in code review. Addresses review feedback on PR #926. * fix(tests): serialize tests using new_session() to avoid race conditions Tests that call new_session() directly create gptme_N sessions that can conflict with other workers' cleanup. This commit: 1. Removes automatic cleanup of gptme_N sessions from the autouse fixture (only keeps worker-specific gptme_{worker_id}_* cleanup) 2. Creates a new cleanup_new_session_sessions fixture for tests that need gptme_N session cleanup (non-autouse, must be explicitly requested) 3. Marks TestNewSession, TestKillSession, and test_auto_prefixes_session_id with @pytest.mark.xdist_group('tmux_new_session') to ensure they run serially on the same worker, preventing race conditions This approach isolates the problematic tests without affecting the performance of other tmux tests that use worker-isolated sessions. * fix(tests): align xdist_group name for test_cli::test_tmux Changed xdist_group from 'tmux' to 'tmux_new_session' to match the group used in test_tools_tmux.py. This ensures ALL tests that create gptme_N sessions run on the same worker, preventing race conditions. * fix(tests): use worker-isolated sessions in TestKillSession The test_kills_session test was failing in CI because new_session() creates gptme_N sessions which can be killed by other workers' cleanup. Fix: Use _create_test_session() which creates worker-isolated sessions (gptme_{worker_id}_N) that aren't affected by other workers' cleanup. This removes the need for xdist_group on TestKillSession since the sessions are now worker-isolated by design.
diff --git a/tests/conftest.py b/tests/conftest.py
@@ -151,6 +151,51 @@ def init_():
     init(None, interactive=False, tool_allowlist=None, tool_format="markdown")
 
 
+@pytest.fixture
+def cleanup_tmux_sessions():
+    """Clean up gptme_* tmux sessions before and after a test.
+
+    This prevents cross-test contamination when tests run the gptme CLI
+    which creates gptme_N sessions internally.
+    """
+    import subprocess
+
+    def _cleanup():
+        """Kill all gptme_* sessions except worker-specific ones."""
+        import re
+
+        # Match simple gptme_N sessions (N = digits only)
+        # but NOT worker-specific ones like gptme_gw0_test_*
+        simple_session_pattern = re.compile(r"^gptme_\d+$")
+
+        try:
+            result = subprocess.run(
+                ["tmux", "list-sessions", "-F", "#{session_name}"],
+                capture_output=True,
+                text=True,
+                timeout=5,
+            )
+            if result.returncode == 0:
+                for session in result.stdout.strip().split("\n"):
+                    session = session.strip()
+                    if not session:
+                        continue
+                    if simple_session_pattern.match(session):
+                        subprocess.run(
+                            ["tmux", "kill-session", "-t", session],
+                            capture_output=True,
+                            timeout=5,
+                        )
+        except (subprocess.TimeoutExpired, FileNotFoundError):
+            pass  # tmux not available or timed out
+
+    # Cleanup before test
+    _cleanup()
+    yield
+    # Cleanup after test
+    _cleanup()
+
+
 @pytest.fixture
 def server_thread():
     """Start a server in a thread for testing."""
diff --git a/tests/test_cli.py b/tests/test_cli.py
@@ -344,7 +344,8 @@ def test_chain(args: list[str], runner: CliRunner):
 # TODO: move elsewhere
 @pytest.mark.slow
 @pytest.mark.requires_api
-def test_tmux(args: list[str], runner: CliRunner):
+@pytest.mark.xdist_group("tmux_new_session")
+def test_tmux(args: list[str], runner: CliRunner, cleanup_tmux_sessions):
     """
     $ gptme '/impersonate lets find out the current load
     ```tmux
diff --git a/tests/test_tools_tmux.py b/tests/test_tools_tmux.py
@@ -67,11 +67,16 @@ def worker_id():
 
 @pytest.fixture(autouse=True)
 def cleanup_sessions(worker_id):
-    """Clean up any test sessions before and after each test."""
+    """Clean up worker-specific test sessions before and after each test.
+    
+    Only cleans up sessions with worker prefix (gptme_{worker_id}_*) to avoid
+    race conditions in parallel test runs. Tests that use new_session() directly
+    should use the cleanup_new_session_sessions fixture instead.
+    """
 
     def cleanup():
         for session in get_sessions():
-            # Only clean up sessions with our worker prefix to avoid killing other workers' sessions
+            # Clean up sessions with our worker prefix (for parallel test isolation)
             if session.startswith(f"gptme_{worker_id}_"):
                 subprocess.run(
                     ["tmux", "kill-session", "-t", session],
@@ -83,6 +88,31 @@ def cleanup():
     cleanup()
 
 
+@pytest.fixture
+def cleanup_new_session_sessions():
+    """Clean up gptme_N sessions for tests that use new_session() directly.
+    
+    This fixture is NOT autouse - only apply to specific tests that need it.
+    Tests using this should also use @pytest.mark.xdist_group("tmux_new_session")
+    to ensure they run serially and avoid race conditions with other workers.
+    """
+    import re
+
+    def cleanup():
+        for session in get_sessions():
+            if session.startswith("gptme_"):
+                # Match gptme_N where N is just digits (not gptme_gw0_*, etc.)
+                if re.fullmatch(r"gptme_\d+", session):
+                    subprocess.run(
+                        ["tmux", "kill-session", "-t", session],
+                        capture_output=True,
+                    )
+
+    cleanup()
+    yield
+    cleanup()
+
+
 class TestGetSessions:
     """Tests for get_sessions function."""
 
@@ -94,16 +124,21 @@ def test_empty_when_no_sessions(self, worker_id):
         assert len(worker_sessions) == 0
 
 
+@pytest.mark.xdist_group("tmux_new_session")
 class TestNewSession:
-    """Tests for new_session function."""
+    """Tests for new_session function.
+    
+    These tests use new_session() directly which creates gptme_N sessions.
+    They are grouped to run serially to avoid race conditions.
+    """
 
-    def test_creates_session(self):
+    def test_creates_session(self, cleanup_new_session_sessions):
         """Should create a new tmux session."""
         msg = new_session("echo 'hello world'")
         assert "gptme_" in msg.content
         assert "hello world" in msg.content or "Running" in msg.content
 
-    def test_increments_session_id(self):
+    def test_increments_session_id(self, cleanup_new_session_sessions):
         """Should create sessions with incrementing IDs."""
         msg1 = new_session("echo 'first'")
         msg2 = new_session("echo 'second'")
@@ -122,32 +157,29 @@ def test_increments_session_id(self):
 class TestListSessions:
     """Tests for list_sessions function."""
 
-    def test_lists_created_sessions(self):
-        """Should list sessions that were created."""
-        msg = new_session("echo 'test'")
-        # Extract the session ID from the creation message
-        import re
+    def test_lists_created_sessions(self, worker_id):
+        """Should list sessions that were created.
 
-        match = re.search(r"gptme_(\d+)", msg.content)
-        assert match
-        session_id = f"gptme_{match.group(1)}"
+        Uses worker-isolated session to avoid race conditions in parallel tests.
+        """
+        # Create a worker-isolated session to avoid race conditions
+        # where other workers' cleanup might remove our session
+        session_id = _create_test_session("echo 'test'", worker_id)
 
         msg = list_sessions()
         assert session_id in msg.content
 
 
 class TestKillSession:
-    """Tests for kill_session function."""
+    """Tests for kill_session function.
+    
+    Uses worker-isolated sessions to avoid race conditions in parallel tests.
+    """
 
-    def test_kills_session(self):
+    def test_kills_session(self, worker_id):
         """Should kill an existing session."""
-        msg = new_session("echo 'to kill'")
-        # Extract the session ID from the creation message
-        import re
-
-        match = re.search(r"gptme_(\d+)", msg.content)
-        assert match
-        session_id = f"gptme_{match.group(1)}"
+        # Use worker-isolated session to avoid race conditions
+        session_id = _create_test_session("echo 'to kill'", worker_id, 1)
 
         msg = kill_session(session_id)
         assert "Killed" in msg.content
@@ -187,8 +219,12 @@ def test_timeout_for_ongoing_command(self, worker_id):
         assert "timed out" in msg.content
         assert elapsed >= 4  # Should have waited for timeout
 
-    def test_auto_prefixes_session_id(self):
-        """Should automatically add gptme_ prefix if missing."""
+    @pytest.mark.xdist_group("tmux_new_session")
+    def test_auto_prefixes_session_id(self, cleanup_new_session_sessions):
+        """Should automatically add gptme_ prefix if missing.
+        
+        Uses new_session() directly, so grouped with other new_session tests.
+        """
         # This test verifies the prefix behavior of wait_for_output
         # We use new_session here because we need a gptme_N style session
         msg = new_session("echo 'prefix test'")
