Agent/examples/tencent_realtime_control/test_feishu_natural_commands.py

from __future__ import annotations

import sys
import unittest
from datetime import datetime
from pathlib import Path
from unittest.mock import Mock, patch
from zoneinfo import ZoneInfo


HERE = Path(__file__).resolve().parent
ROOT = HERE.parents[1]
for path in (ROOT, HERE):
    if str(path) not in sys.path:
        sys.path.insert(0, str(path))

from agent.tools.builtin.feishu.feishu_client import (
    ChatType,
    FeishuClient,
    FeishuMessageEvent,
)
from command_intent_parser import CommandIntentParser, IntentParserConfig
from feishu_command_service import FeishuCommandService
from operator_commands import (
    ACTION_DAY_PAUSE,
    ACTION_STOP,
    ACTION_TODAY_SPEND,
    SCOPE_ACCOUNTS,
    SCOPE_ALL,
    SCOPE_AUTOMATION,
    SCOPE_MISSING,
    CommandIntent,
    parse_deterministic_intent,
)
from operator_control import _execute_pause, pause_status_summary, preview_write_command
from realtime_config import RealtimeControlConfig
from run_scheduler import next_wake
from tencent_client import TencentClient, TencentWriteRejectedError
from today_spend_query import query_today_spend


SHANGHAI = ZoneInfo("Asia/Shanghai")


class NaturalCommandParsingTest(unittest.TestCase):
    def test_bare_pause_requires_scope(self) -> None:
        intent = parse_deterministic_intent("暂停")
        self.assertIsNotNone(intent)
        self.assertEqual(ACTION_DAY_PAUSE, intent.action)
        self.assertFalse(intent.complete)
        self.assertEqual(("scope",), intent.missing_fields)

    def test_stop_today_is_temporary_and_all_managed(self) -> None:
        intent = parse_deterministic_intent("停止今天所有账户的投放")
        self.assertEqual(ACTION_DAY_PAUSE, intent.action)
        self.assertEqual(SCOPE_ALL, intent.scope_type)
        self.assertTrue(intent.complete)

    def test_automation_accounts_means_all_managed(self) -> None:
        intent = parse_deterministic_intent("停止自动化账户")
        self.assertEqual(ACTION_STOP, intent.action)
        self.assertEqual(SCOPE_ALL, intent.scope_type)

    def test_model_cannot_invent_account_ids(self) -> None:
        parser = CommandIntentParser(IntentParserConfig(True, "test", 1, 0.8))
        intent = parser._validate_model_intent(
            {
                "action": "STOP",
                "scope_type": "ACCOUNTS",
                "missing_fields": [],
                "confidence": 0.99,
            },
            "把那个账户停掉",
        )
        self.assertEqual(SCOPE_MISSING, intent.scope_type)
        self.assertFalse(intent.complete)
        self.assertEqual((), intent.account_ids)

    def test_source_account_id_overrides_model_all_scope(self) -> None:
        parser = CommandIntentParser(IntentParserConfig(True, "test", 1, 0.8))
        intent = parser._validate_model_intent(
            {
                "action": "STOP",
                "scope_type": "ALL",
                "missing_fields": [],
                "confidence": 0.99,
            },
            "把86748335停掉",
        )
        self.assertEqual(SCOPE_ACCOUNTS, intent.scope_type)
        self.assertEqual((86748335,), intent.account_ids)

    def test_deterministic_action_is_not_overridden_by_model(self) -> None:
        parser = CommandIntentParser(IntentParserConfig(True, "test", 1, 0.8))
        with patch.object(
            parser,
            "_parse_with_model",
            return_value=CommandIntent(
                action=ACTION_STOP,
                scope_type=SCOPE_ALL,
                parse_source="llm",
            ),
        ):
            intent = parser.understand("今天让这批自动投放账户先歇一下")
        self.assertEqual(ACTION_DAY_PAUSE, intent.action)
        self.assertEqual(SCOPE_ALL, intent.scope_type)

    def test_incomplete_standard_command_survives_model_failure(self) -> None:
        parser = CommandIntentParser(IntentParserConfig(True, "test", 1, 0.8))
        with patch.object(
            parser,
            "_parse_with_model",
            side_effect=RuntimeError("model unavailable"),
        ):
            intent = parser.understand("暂停")
        self.assertEqual(ACTION_DAY_PAUSE, intent.action)
        self.assertEqual(("scope",), intent.missing_fields)

    def test_model_selects_automation_scope_for_today_spend(self) -> None:
        parser = CommandIntentParser(IntentParserConfig(True, "test", 1, 0.8))
        intent = parser._validate_model_intent(
            {
                "action": "TODAY_SPEND",
                "scope_type": "AUTOMATION",
                "missing_fields": [],
                "confidence": 0.99,
            },
            "查询今天自动化账户消耗",
        )
        self.assertEqual(ACTION_TODAY_SPEND, intent.action)
        self.assertEqual(SCOPE_AUTOMATION, intent.scope_type)
        self.assertTrue(intent.complete)


class TodaySpendQueryTest(unittest.TestCase):
    class Tencent:
        def get_today_account_metrics(self, account_id, data_date):
            return {
                "cost_fen": account_id,
                "impressions": 100,
                "clicks": 10,
                "conversions": 2,
            }

    def test_all_scope_returns_summary_only(self) -> None:
        now = datetime(2026, 7, 30, 12, 0, tzinfo=SHANGHAI)
        with patch(
            "today_spend_query.load_all_spend_accounts",
            return_value=[{"account_id": 10000001}, {"account_id": 10000002}],
        ):
            summary = query_today_spend(
                CommandIntent(
                    action=ACTION_TODAY_SPEND,
                    scope_type=SCOPE_ALL,
                ).to_parsed_command(),
                now=now,
                tencent=self.Tencent(),
            )
        self.assertEqual(2, summary["account_count"])
        self.assertEqual(20000003, summary["cost_fen"])
        self.assertEqual(200, summary["impressions"])
        self.assertTrue(summary["complete"])


class FeishuAuthorizationTest(unittest.TestCase):
    def setUp(self) -> None:
        self.service = FeishuCommandService.__new__(FeishuCommandService)
        self.service.allowed_chat_id = "oc_allowed"
        self.service.allowed_open_ids = {"ou_allowed"}

    def _event(self, *, mentioned: bool, chat_type: ChatType = ChatType.GROUP):
        return FeishuMessageEvent(
            message_id="om_1",
            chat_id="oc_allowed",
            chat_type=chat_type,
            content="暂停全部",
            content_type="text",
            sender_open_id="ou_allowed",
            mentioned_bot=mentioned,
        )

    def test_group_message_without_mention_is_ignored(self) -> None:
        self.assertFalse(self.service._authorized(self._event(mentioned=False)))

    def test_private_message_is_ignored(self) -> None:
        self.assertFalse(
            self.service._authorized(
                self._event(mentioned=True, chat_type=ChatType.P2P)
            )
        )


class FeishuMentionDetectionTest(unittest.TestCase):
    def setUp(self) -> None:
        self.client = FeishuClient.__new__(FeishuClient)
        self.client._bot_open_id = "ou_bot"

    @staticmethod
    def _mention(open_id: str) -> Mock:
        mention = Mock()
        mention.id.open_id = open_id
        return mention

    def test_mentioning_another_user_is_not_bot_mention(self) -> None:
        self.assertFalse(
            self.client._check_bot_mentioned([self._mention("ou_other")])
        )

    def test_mentioning_bot_is_bot_mention(self) -> None:
        self.assertTrue(
            self.client._check_bot_mentioned([self._mention("ou_bot")])
        )

    def test_missing_bot_identity_fails_closed(self) -> None:
        self.client._bot_open_id = None
        with patch.object(
            self.client,
            "_load_bot_open_id",
            side_effect=RuntimeError("unavailable"),
        ):
            self.assertFalse(
                self.client._check_bot_mentioned([self._mention("ou_other")])
            )


class PreviewSnapshotTest(unittest.TestCase):
    class FakeTencent:
        def get_ads(self, account_id: int):
            if account_id == 10000001:
                return [
                    {
                        "adgroup_id": 101,
                        "adgroup_name": "active",
                        "configured_status": "AD_STATUS_NORMAL",
                        "begin_date": "2026-07-01",
                        "end_date": "0",
                    },
                    {
                        "adgroup_id": 102,
                        "adgroup_name": "manual-off",
                        "configured_status": "AD_STATUS_SUSPEND",
                    },
                    {
                        "adgroup_id": 103,
                        "adgroup_name": "already-deferred",
                        "configured_status": "AD_STATUS_NORMAL",
                        "begin_date": "2026-07-31",
                    },
                ]
            return []

        def get_today_ad_metrics(self, account_id, adgroup_ids, data_date):
            return {
                101: {
                    "cost_fen": 12345,
                    "impressions": 1000,
                    "clicks": 20,
                    "conversions": 3,
                }
            }

    def test_preview_uses_impacted_accounts_and_freezes_metrics(self) -> None:
        captured = {}

        def persist(record, items):
            captured["record"] = record
            captured["items"] = items
            return {**record}

        now = datetime(2026, 7, 30, 12, 0, tzinfo=SHANGHAI)
        intent = parse_deterministic_intent("暂停全部")
        with (
            patch(
                "operator_control.load_realtime_accounts",
                return_value=[
                    {"account_id": 10000001, "audience_name": "auto"},
                    {"account_id": 10000002, "audience_name": "pause-only"},
                ],
            ),
            patch("operator_control.load_operator_pauses", return_value=[]),
            patch("operator_control.find_pending_command_conflict", return_value=None),
            patch("operator_control.create_operator_command_with_items", side_effect=persist),
            patch("operator_control.advisory_lock") as preview_lock,
        ):
            preview_lock.return_value.__enter__.return_value = True
            command = preview_write_command(
                intent.to_parsed_command(),
                now=now,
                source_message_id="om_1",
                chat_id="oc_1",
                sender_open_id="ou_1",
                sender_name="operator",
                confirmation_ttl_minutes=10,
                start_hour=6,
                tencent=self.FakeTencent(),
            )

        self.assertEqual(1, command["preview_account_count"])
        self.assertEqual([10000001], command["target_account_ids"])
        self.assertEqual(1, command["preview_ad_count"])
        self.assertEqual(12345, command["preview_cost_fen"])
        self.assertEqual(101, captured["items"][0]["adgroup_id"])
        self.assertEqual(3, captured["items"][0]["preview_conversions"])
        self.assertEqual(
            "AD_STATUS_NORMAL",
            captured["items"][0]["target_status"],
        )


class DayPauseExecutionTest(unittest.TestCase):
    class FakeTencent:
        def __init__(self) -> None:
            self.date_updates = []

        def get_ads(self, account_id: int):
            return [{
                "adgroup_id": 101,
                "adgroup_name": "active",
                "configured_status": "AD_STATUS_NORMAL",
                "begin_date": "2026-07-01",
                "end_date": "0",
            }]

        def update_ad_begin_dates(self, account_id, adgroup_ids, begin_date):
            self.date_updates.append((account_id, adgroup_ids, begin_date))

        def update_ad(self, *args, **kwargs):
            raise AssertionError("DAY_PAUSE must not update configured_status")

    def test_day_pause_only_moves_begin_date_to_tomorrow(self) -> None:
        client = self.FakeTencent()
        now = datetime(2026, 7, 30, 12, 0, tzinfo=SHANGHAI)
        with (
            patch(
                "operator_control.load_ad_states",
                return_value={101: {"operator_pause_mode": None}},
            ),
            patch("operator_control.set_operator_pause") as set_pause,
            patch("operator_control._record_item") as record_item,
        ):
            successes, failures, skipped = _execute_pause(
                {"command_id": "cmd_1", "action": ACTION_DAY_PAUSE},
                {"account_id": 10000001, "audience_name": "auto"},
                now=now,
                start_hour=6,
                client=client,
                items=[{"id": 1, "adgroup_id": 101}],
            )

        self.assertEqual((1, 0, 0), (successes, failures, skipped))
        self.assertEqual(
            [(10000001, [101], "2026-07-31")],
            client.date_updates,
        )
        self.assertEqual("UNTIL_NEXT_DELIVERY", set_pause.call_args.kwargs["mode"])
        self.assertEqual(
            "AD_STATUS_NORMAL",
            record_item.call_args.kwargs["target_status"],
        )

    def test_rejected_day_pause_clears_pending_pause_state(self) -> None:
        client = self.FakeTencent()
        now = datetime(2026, 7, 30, 12, 0, tzinfo=SHANGHAI)
        with (
            patch.object(
                client,
                "update_ad_begin_dates",
                side_effect=TencentWriteRejectedError("rejected"),
            ),
            patch(
                "operator_control.load_ad_states",
                return_value={101: {"operator_pause_mode": None}},
            ),
            patch("operator_control.set_operator_pause"),
            patch("operator_control.clear_operator_pause") as clear_pause,
            patch("operator_control._record_item"),
        ):
            result = _execute_pause(
                {"command_id": "cmd_1", "action": ACTION_DAY_PAUSE},
                {"account_id": 10000001, "audience_name": "auto"},
                now=now,
                start_hour=6,
                client=client,
                items=[{"id": 1, "adgroup_id": 101}],
            )

        self.assertEqual((0, 1, 0), result)
        clear_pause.assert_called_once_with(
            10000001,
            101,
            action="OPERATOR_PAUSE_FAILED",
            action_at=now,
        )

    def test_scheduler_waits_for_cpm_start_not_delivery_start(self) -> None:
        config = RealtimeControlConfig(start_hour=12, next_delivery_hour=6)
        now = datetime(2026, 7, 30, 1, 0, tzinfo=SHANGHAI)
        self.assertEqual(12, next_wake(now, config).hour)

    def test_expired_day_pause_is_not_reported_as_active(self) -> None:
        now = datetime(2026, 7, 31, 8, 0, tzinfo=SHANGHAI)
        with (
            patch(
                "operator_control.load_realtime_accounts",
                return_value=[{"account_id": 10000001}],
            ),
            patch(
                "operator_control.load_operator_pauses",
                return_value=[{
                    "account_id": 10000001,
                    "adgroup_id": 101,
                    "operator_pause_mode": "UNTIL_NEXT_DELIVERY",
                    "operator_resume_at": datetime(2026, 7, 31, 6, 0),
                }],
            ),
        ):
            summary = pause_status_summary(now)
        self.assertEqual(0, summary["total"])


class TencentDateUpdateClassificationTest(unittest.TestCase):
    def test_tencent_rejection_uses_rejected_error(self) -> None:
        client = TencentClient()
        response = Mock(status_code=200, text='{"code": 1900001}')
        response.json.return_value = {
            "code": 1900001,
            "message": "rejected",
        }
        with (
            patch.object(client, "_common_params", return_value={}),
            patch.object(client, "_user_token", return_value="token"),
            patch.object(client.session, "post", return_value=response),
            self.assertRaises(TencentWriteRejectedError),
        ):
            client.update_ad_begin_dates(10000001, [101], "2026-07-31")


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