inquiry_robot/inquiry-agent/tests/test_collab_watch_reconcile.py

# -*- coding: utf-8 -*-
"""协同群监听对账：补挂与清孤儿。"""

from __future__ import annotations

import unittest
from types import SimpleNamespace
from unittest.mock import MagicMock, patch


class CollabWatchReconcileTests(unittest.TestCase):
    def test_rewatch_open_missing_from_watch(self) -> None:
        import agent.jobs.collab_watch_reconcile as mod

        book = MagicMock()
        book.list_open_collab.return_value = [
            {
                "work_order_no": "WO1",
                "collab_chat_id": "wr_open",
                "collab_ended": False,
                "collab_member_ids": ["A", "B"],
            }
        ]
        with (
            patch.object(mod, "_try_lock", return_value=True),
            patch("agent.channel.wshoto.rooms.list_watched_rooms", return_value=[]),
            patch("agent.channel.wshoto.rooms.watch_collab_room") as watch,
            patch("agent.channel.wshoto.rooms.unwatch_collab_room") as unwatch,
            patch("agent.channel.wshoto.rooms.ticket_no_for_room", return_value=""),
            patch.object(mod, "_group_client", return_value=MagicMock()),
        ):
            stats = mod.reconcile_collab_watch(
                ledger=book,
                settings=SimpleNamespace(ledger_backend="http"),
                dismiss=False,
            )
        self.assertEqual(stats["rewatched"], 1)
        watch.assert_called_with("wr_open", work_order_no="WO1")
        unwatch.assert_not_called()

    def test_dismiss_ended_in_watch(self) -> None:
        import agent.jobs.collab_watch_reconcile as mod

        book = MagicMock()
        book.list_open_collab.return_value = [
            {
                "work_order_no": "WO2",
                "collab_chat_id": "wr_end",
                "collab_ended": True,
                "collab_member_ids": ["A", "owner"],
            }
        ]
        client = MagicMock()
        client._appchat_info.return_value = {
            "owner": "owner",
            "userlist": ["A", "owner"],
        }
        with (
            patch.object(mod, "_try_lock", return_value=True),
            patch("agent.channel.wshoto.rooms.list_watched_rooms", return_value=["wr_end"]),
            patch("agent.channel.wshoto.rooms.watch_collab_room"),
            patch("agent.channel.wshoto.rooms.unwatch_collab_room") as unwatch,
            patch("agent.channel.wshoto.rooms.ticket_no_for_room", return_value="WO2"),
            patch.object(mod, "_group_client", return_value=client),
            patch("agent.policy.inquiry_copy.group_dissolved", return_value="群已解散"),
        ):
            stats = mod.reconcile_collab_watch(
                ledger=book,
                settings=SimpleNamespace(ledger_backend="http"),
                dismiss=True,
            )
        self.assertEqual(stats["dismissed"], 1)
        client.dismiss_group.assert_called()
        unwatch.assert_called_with("wr_end")

    def test_list_empty_does_not_orphan_dismiss_live_redis_wo(self) -> None:
        """listOpen 空 + byChat 失败时，Redis 仍有活单工单号 → 禁止清人。"""
        import agent.jobs.collab_watch_reconcile as mod

        book = MagicMock()
        book.list_open_collab.return_value = []
        book.find_by_collab_chat.return_value = None
        book.get_ticket.return_value = {
            "work_order_no": "WO202610030039",
            "collab_ended": False,
            "collab_chat_id": "wr_live",
            "collab_member_ids": ["WuJiLin", "LinMianXun"],
        }
        client = MagicMock()
        client._appchat_info.return_value = {
            "owner": "WuJiLin",
            "userlist": ["WuJiLin", "LinMianXun", "AI001"],
        }
        with (
            patch.object(mod, "_try_lock", return_value=True),
            patch("agent.channel.wshoto.rooms.list_watched_rooms", return_value=["wr_live"]),
            patch("agent.channel.wshoto.rooms.watch_collab_room"),
            patch("agent.channel.wshoto.rooms.unwatch_collab_room") as unwatch,
            patch(
                "agent.channel.wshoto.rooms.ticket_no_for_room",
                return_value="WO202610030039",
            ),
            patch.object(mod, "_group_client", return_value=client),
        ):
            stats = mod.reconcile_collab_watch(
                ledger=book,
                settings=SimpleNamespace(ledger_backend="http"),
                dismiss=True,
            )
        self.assertEqual(stats["dismissed"], 0)
        client.dismiss_group.assert_not_called()
        unwatch.assert_not_called()
        self.assertGreaterEqual(stats["skipped"], 1)

    def test_list_none_blocks_orphan_without_redis_wo(self) -> None:
        """listOpen 失败(None) 且无 Redis 工单映射时，也不做孤儿清人。"""
        import agent.jobs.collab_watch_reconcile as mod

        book = MagicMock()
        book.list_open_collab.return_value = None
        book.find_by_collab_chat.return_value = None
        client = MagicMock()
        client._appchat_info.return_value = {
            "owner": "A",
            "userlist": ["A", "B"],
        }
        with (
            patch.object(mod, "_try_lock", return_value=True),
            patch("agent.channel.wshoto.rooms.list_watched_rooms", return_value=["wr_x"]),
            patch("agent.channel.wshoto.rooms.watch_collab_room"),
            patch("agent.channel.wshoto.rooms.unwatch_collab_room") as unwatch,
            patch("agent.channel.wshoto.rooms.ticket_no_for_room", return_value=""),
            patch.object(mod, "_group_client", return_value=client),
        ):
            stats = mod.reconcile_collab_watch(
                ledger=book,
                settings=SimpleNamespace(ledger_backend="http"),
                dismiss=True,
            )
        self.assertEqual(stats["dismissed"], 0)
        client.dismiss_group.assert_not_called()
        unwatch.assert_not_called()
        self.assertGreaterEqual(stats["skipped"], 1)


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