codex-app-server/cloud/tests/test_thread_unarchive.py

"""unarchive 解析单测：已归档 thread 只能在 --archived 列表里。

背景（生产实测 2026-10-28，探针 probe9）：thread/archive
200 后 thread/unarchive 稳定 404。根因是 #103 时代的
逻辑 bug：unarchive 走普通解析，普通 thread-list 不传
--archived 时只回非归档 thread，刚归档的 thread 永远
解析不到。本批修法：解析加 archived 模式，unarchive
专走该模式；其余改操作（rename/archive/compact）不动，
仍走非归档解析。

不跑真实模型、不调真实 host.py：_run_host_cli 均 stub。
"""

import os
import sys
import tempfile
import unittest
from pathlib import Path
from unittest import mock

try:
    import cloud.server as server_mod
    from cloud.server import (
        CloudControlPlaneHandler,
        _pick_latest_thread,
    )
except ImportError:
    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
    import cloud.server as server_mod
    from cloud.server import (
        CloudControlPlaneHandler,
        _pick_latest_thread,
    )


def _handler(tmp):
    h = CloudControlPlaneHandler.__new__(CloudControlPlaneHandler)
    srv = mock.Mock()
    srv.tenants_root = os.path.join(tmp, "tenants")
    srv.host_worktrees_dir = os.path.join(tmp, "worktrees")
    h.server = srv
    return h


class ArchivedResolveTestCase(unittest.TestCase):
    def _stub(self, calls, plain, archived):
        def fake_cli(argv, home, model_provider=None, **kwargs):
            calls.append(list(argv))
            if "--archived" in argv:
                return True, {"data": archived}
            return True, {"data": plain}
        return fake_cli

    def test_archived_list_passes_flag(self):
        # 生产形状：archived 列表里条目的 archived 字段为 None。
        thread = {"id": "t-a", "updatedAt": 100, "archived": None}
        calls = []
        with tempfile.TemporaryDirectory() as tmp:
            h = _handler(tmp)
            task = {"id": "task-1", "project_id": "proj-1"}
            with mock.patch.object(
                    server_mod, "_run_host_cli",
                    self._stub(calls, [], [thread])), \
                    mock.patch.object(
                        server_mod, "_local_worker_codex_home",
                        return_value=None):
                threads, _h, _c, err, _p = \
                    h._list_task_threads(task, archived=True)
        self.assertIsNone(err)
        self.assertEqual(threads, [thread])
        self.assertIn("--archived", calls[0])

    def test_plain_resolve_404s_when_only_archived_exists(self):
        # 这就是探针里 404 的原样：普通列表空、归档列表有。
        thread = {"id": "t-a", "updatedAt": 100, "archived": None}
        with tempfile.TemporaryDirectory() as tmp:
            h = _handler(tmp)
            task = {"id": "task-1", "project_id": "proj-1"}
            with mock.patch.object(
                    server_mod, "_run_host_cli",
                    self._stub([], [], [thread])), \
                    mock.patch.object(
                        server_mod, "_local_worker_codex_home",
                        return_value=None):
                _t, _h, _c, err, _p = h._resolve_task_thread(task)
                self.assertEqual(err, (404, "thread not found"))
                got, _h, _c, err2, _p = \
                    h._resolve_task_thread(task, archived=True)
        self.assertIsNone(err2)
        self.assertEqual(got["id"], "t-a")

    def test_pick_latest_archived_flag(self):
        thread = {"id": "t-a", "updatedAt": 100, "archived": True}
        self.assertIsNone(_pick_latest_thread([thread]))
        self.assertEqual(
            _pick_latest_thread([thread], include_archived=True),
            thread)


class UnarchiveMutateTestCase(unittest.TestCase):
    def test_unarchive_resolves_archived_and_calls_cli(self):
        thread = {"id": "t-a"}
        task = {"id": "task-1", "project_id": "proj-1",
                "status": "completed"}
        seen = {}
        with tempfile.TemporaryDirectory() as tmp:
            h = _handler(tmp)
            h.server.db_path = os.path.join(tmp, "control.db")

            def fake_resolve(task_arg, archived=False):
                seen["archived"] = archived
                return thread, "/home/tenant", "/wt", None, "sub2api"

            def fake_cli(argv, home, timeout=60, model_provider=None,
                         **kwargs):
                seen["argv"] = argv
                seen["provider"] = model_provider
                return True, {}

            store = mock.Mock()
            ctx = mock.Mock()
            ctx.__enter__ = mock.Mock(return_value=store)
            ctx.__exit__ = mock.Mock(return_value=False)
            sent = {}

            def fake_send(status, body):
                sent["status"] = status
                sent["body"] = body

            with mock.patch.object(h, "_thread_task",
                                   return_value=(task, None)), \
                    mock.patch.object(h, "_thread_task_guard",
                                      return_value=None), \
                    mock.patch.object(h, "_resolve_task_thread",
                                      fake_resolve), \
                    mock.patch.object(server_mod, "_run_host_cli",
                                      fake_cli), \
                    mock.patch.object(server_mod, "_auth_store",
                                      return_value=ctx), \
                    mock.patch.object(h, "_send_json", fake_send):
                h._task_thread_unarchive("user-1", "task-1")
        self.assertTrue(seen["archived"])
        self.assertEqual(seen["argv"],
                         ["thread-unarchive", "--thread-id", "t-a"])
        self.assertEqual(seen["provider"], "sub2api")
        self.assertEqual(sent["status"], 200)
        self.assertEqual(sent["body"]["thread_id"], "t-a")


if __name__ == "__main__":
    unittest.main()