Allow deleting stopped sessions with stale tabs

tests/test_selector.py

@@ -112,6 +112,42 @@ class SelectorCatalogTests(unittest.TestCase):
         matches = filter_rows(failed, "offline")
         self.assertEqual([row.kind for row in matches], ["retry"])
 
+    def test_stopped_remote_with_a_dead_tab_remains_deletable(self) -> None:
+        window = CatalogWindow(
+            "@2",
+            "remote:orangepi:old",
+            "orangepi/old",
+            True,
+            False,
+            kind="remote",
+            target="orangepi",
+            session="old",
+        )
+        result = RemoteQueryResult(
+            RemoteEntry("orangepi"),
+            sessions=(
+                RemoteSession("default", False),
+                RemoteSession("old", False),
+            ),
+        )
+
+        rows = build_main_rows(
+            [window],
+            [RemoteEntry("orangepi")],
+            {"orangepi": result},
+            set(),
+        )
+
+        row = next(item for item in rows if item.key == "window:@2")
+        self.assertTrue(row.deletable)
+        self.assertEqual(row.window_id, "@2")
+        self.assertIn("同时关闭 Tab", row.detail)
+        default = next(
+            item for item in rows if item.key == "remote:orangepi:default"
+        )
+        self.assertFalse(default.deletable)
+        self.assertIn("Herdr 保留 default", default.detail)
+
     def test_active_remote_session_has_close_and_stop_action(self) -> None:
         local = CatalogWindow(
             "@1",
@@ -313,13 +349,13 @@ class SelectorActionTests(unittest.TestCase):
         with tempfile.TemporaryDirectory() as directory:
             registry = RemoteRegistry(Path(directory) / "remotes.json")
             registry.upsert(RemoteEntry("orangepi"))
-            deleted: list[tuple[str, str]] = []
+            deleted: list[tuple[str, str, str | None]] = []
             selector = SessionSelector(
                 [],
                 registry,
                 str(Path(directory) / "missing-ssh"),
-                delete_session_handler=lambda target, session: deleted.append(
-                    (target, session)
+                delete_session_handler=lambda target, session, window_id: deleted.append(
+                    (target, session, window_id)
                 ),
             )
             self.addCleanup(selector.pool.close)
@@ -341,7 +377,7 @@ class SelectorActionTests(unittest.TestCase):
             self.assertEqual(selector.view, "main")
             self.assertEqual(deleted, [])
             self.assertEqual(selector.query, "")
-            self.assertIn("只能删除", selector.message)
+            self.assertIn("default", selector.message)
 
             selector.query = ""
             old = next(row for row in rows if row.key == "remote:orangepi:old")
@@ -364,7 +400,7 @@ class SelectorActionTests(unittest.TestCase):
             self.assertEqual(selector.view, "working-delete-session")
             wait_for_action(selector)
 
-            self.assertEqual(deleted, [("orangepi", "old")])
+            self.assertEqual(deleted, [("orangepi", "old", None)])
             self.assertEqual(selector.view, "main")
             self.assertIn("已删除历史 Session", selector.message)
             self.assertNotIn(
@@ -372,6 +408,73 @@ class SelectorActionTests(unittest.TestCase):
                 [item.name for item in selector.results["orangepi"].sessions],
             )
 
+    def test_stopped_remote_session_with_tab_is_deleted_and_tab_is_closed(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            registry = RemoteRegistry(Path(directory) / "remotes.json")
+            registry.upsert(RemoteEntry("orangepi"))
+            local = CatalogWindow(
+                "@1",
+                "local:default",
+                "local",
+                False,
+                True,
+                kind="local",
+                session="default",
+            )
+            remote = CatalogWindow(
+                "@2",
+                "remote:orangepi:old",
+                "orangepi/old",
+                True,
+                False,
+                kind="remote",
+                target="orangepi",
+                session="old",
+            )
+            deleted: list[tuple[str, str, str | None]] = []
+
+            def delete_handler(
+                target: str,
+                session: str,
+                window_id: str | None,
+            ) -> DeleteSessionOutcome:
+                deleted.append((target, session, window_id))
+                return DeleteSessionOutcome(
+                    windows=(local,),
+                    tab_closed=True,
+                )
+
+            selector = SessionSelector(
+                [local, remote],
+                registry,
+                str(Path(directory) / "missing-ssh"),
+                delete_session_handler=delete_handler,
+            )
+            self.addCleanup(selector.pool.close)
+            self.addCleanup(selector.action_executor.shutdown, wait=True)
+            selector.results["orangepi"] = RemoteQueryResult(
+                RemoteEntry("orangepi"),
+                sessions=(RemoteSession("old", False),),
+            )
+
+            rows = selector._rows()  # type: ignore[attr-defined]
+            row = next(item for item in rows if item.key == "window:@2")
+            selector.selected_key = row.key
+            selector._handle_key("d", rows)  # type: ignore[attr-defined]
+            self.assertEqual(selector.view, "confirm-delete-session")
+            confirm = next(
+                item
+                for item in selector._rows()  # type: ignore[attr-defined]
+                if item.kind == "delete-session-yes"
+            )
+            self.assertEqual(confirm.window_id, "@2")
+            selector._activate(confirm)  # type: ignore[attr-defined]
+            wait_for_action(selector)
+
+            self.assertEqual(deleted, [("orangepi", "old", "@2")])
+            self.assertEqual(selector.windows, [local])
+            self.assertIn("已关闭 Tab 并删除", selector.message)
+
     def test_delete_partial_success_is_reported_as_warning_not_failure(self) -> None:
         with tempfile.TemporaryDirectory() as directory:
             registry = RemoteRegistry(Path(directory) / "remotes.json")
@@ -380,8 +483,8 @@ class SelectorActionTests(unittest.TestCase):
                 [],
                 registry,
                 "/missing/ssh",
-                delete_session_handler=lambda _target, _session: DeleteSessionOutcome(
-                    "显示名称清理失败"
+                delete_session_handler=lambda _target, _session, _window_id: (
+                    DeleteSessionOutcome("显示名称清理失败")
                 ),
             )
             self.addCleanup(selector.pool.close)