Agent/examples/auto_put_ad_mini/test_roi_agency_delivery.py

import json
import logging
import os
import tempfile
import unittest
from contextlib import contextmanager
from datetime import datetime
from pathlib import Path
from unittest.mock import Mock, patch

import httpx

from roi_control.agency_delivery import (
    AgencyWebhookNotifier,
    agency_route_key,
    publish_agency_reports,
)
from roi_control.config import AgencyWebhookConfig, RoiConfig
from roi_control.feishu import RoiFeishuPublisher
from roi_control.service import (
    _build_internal_reports,
    _internal_source_revision,
    _internal_webhook_config,
    _redact_agency_webhooks,
)


WEBHOOK = "https://open.feishu.cn/open-apis/bot/v2/hook/test-route"


class FakePublisher:
    def __init__(self):
        self.paths = []

    def upload_workbook(self, path):
        self.paths.append(path)
        return {"url": "https://example.test/sheet", "sheet_token": "sheet-1"}


class FakeNotifier:
    def __init__(self):
        self.calls = []

    def send(self, webhook_url, **values):
        self.calls.append((webhook_url, values))
        return "0"


class AgencyDeliveryTest(unittest.TestCase):
    def test_formal_and_internal_schedule_defaults(self):
        with patch.dict(os.environ, {}, clear=True):
            config = RoiConfig.from_env()
        self.assertEqual((config.report_hour, config.report_minute), (9, 0))
        self.assertEqual(
            (config.internal_test_hour, config.internal_test_minute),
            (15, 30),
        )
        self.assertTrue(config.internal_test_dedup_enabled)

    def test_internal_test_dedup_switch_controls_run_revision(self):
        current = datetime(2026, 8, 4, 14, 0, 1, 123456)
        dedup = RoiConfig(internal_test_dedup_enabled=True)
        no_dedup = RoiConfig(internal_test_dedup_enabled=False)
        self.assertEqual(_internal_source_revision(dedup, current), "internal_test")
        self.assertEqual(
            _internal_source_revision(no_dedup, current),
            "it_140001123456",
        )

    def test_internal_report_set_contains_only_automation_report(self):
        reports = _build_internal_reports(
            agency_reports=[
                {
                    "agency_name": "小程序-代投-棱镜",
                    "report_version": "v7",
                    "report": "mirror.xlsx",
                    "creative_rows": 4,
                    "ad_rows": 2,
                },
                {
                    "agency_name": "自动化投放",
                    "report_version": "v7",
                    "report": "auto.xlsx",
                    "creative_rows": 1,
                    "ad_rows": 1,
                },
            ],
        )
        self.assertEqual(
            [row["agency_name"] for row in reports],
            ["自动化投放"],
        )
        self.assertTrue(all(str(row["title"]).startswith("内部测试_") for row in reports))
        config = _internal_webhook_config(reports, WEBHOOK)
        self.assertEqual(set(config.webhooks.values()), {WEBHOOK})
        self.assertEqual(set(config.webhooks), {"自动化投放"})

    def test_internal_report_set_rejects_missing_automation_report(self):
        with self.assertRaisesRegex(RuntimeError, "自动化投放"):
            _build_internal_reports(
                agency_reports=[
                    {
                        "agency_name": "小程序-代投-棱镜",
                        "report_version": "v7",
                        "report": "mirror.xlsx",
                        "creative_rows": 4,
                        "ad_rows": 2,
                    }
                ]
            )

    def test_config_snapshot_never_contains_webhook_secret(self):
        env = {
            "ROI_AGENCY_WEBHOOK_ENABLED": "1",
            "ROI_AGENCY_WEBHOOKS_JSON": json.dumps({"棱镜": WEBHOOK}),
        }
        with patch.dict(os.environ, env, clear=False):
            config = AgencyWebhookConfig.from_env()
        snapshot = config.snapshot()
        self.assertEqual(snapshot["agencies"], ["棱镜"])
        self.assertNotIn(WEBHOOK, json.dumps(snapshot, ensure_ascii=False))

    def test_disabled_config_does_not_parse_secret_json(self):
        with patch.dict(
            os.environ,
            {
                "ROI_AGENCY_WEBHOOK_ENABLED": "0",
                "ROI_AGENCY_WEBHOOKS_JSON": "not-json",
            },
            clear=False,
        ):
            config = AgencyWebhookConfig.from_env()
        self.assertFalse(config.enabled)
        self.assertEqual(config.webhooks, {})

    def test_route_key_is_exact_and_only_removes_known_prefix(self):
        self.assertEqual(agency_route_key("小程序-代投-棱镜"), "棱镜")
        self.assertEqual(agency_route_key("小程序 -代投-贝湉"), "贝湉")
        self.assertEqual(agency_route_key("自动化投放"), "自动化投放")

    def test_service_error_redaction_removes_webhook_secret(self):
        config = AgencyWebhookConfig(enabled=True, webhooks={"棱镜": WEBHOOK})
        error = RuntimeError(f"POST {WEBHOOK} failed")
        redacted = _redact_agency_webhooks(error, config)
        self.assertNotIn(WEBHOOK, redacted)
        self.assertIn("<redacted>", redacted)

    @patch("roi_control.agency_delivery.update_agency_delivery")
    @patch("roi_control.agency_delivery.upsert_agency_delivery")
    def test_only_configured_agency_uploads_and_sends(
        self,
        upsert,
        update,
    ):
        upsert.side_effect = [
            {"id": 1, "status": "PENDING"},
            {"id": 2, "status": "PENDING"},
        ]
        publisher = FakePublisher()
        notifier = FakeNotifier()
        with tempfile.TemporaryDirectory() as directory:
            route_path = Path(directory) / "20260804_棱镜_调控建议.xlsx"
            route_path.write_bytes(b"xlsx")
            skipped_path = Path(directory) / "20260804_自动化投放_调控建议.xlsx"
            skipped_path.write_bytes(b"xlsx")
            reports = [
                {
                    "agency_name": "小程序-代投-棱镜",
                    "report_version": "roi_agency_advice_v7",
                    "report": str(route_path),
                    "creative_rows": 10,
                    "ad_rows": 3,
                },
                {
                    "agency_name": "自动化投放",
                    "report_version": "roi_agency_advice_v7",
                    "report": str(skipped_path),
                    "creative_rows": 1,
                    "ad_rows": 1,
                },
            ]
            outcomes = publish_agency_reports(
                run_id="run-1",
                reports=reports,
                config=AgencyWebhookConfig(enabled=True, webhooks={"棱镜": WEBHOOK}),
                publisher=publisher,
                notifier=notifier,
            )

        self.assertEqual([row["status"] for row in outcomes], ["SENT", "SKIPPED"])
        self.assertEqual(len(publisher.paths), 1)
        self.assertEqual(len(notifier.calls), 1)
        self.assertEqual(notifier.calls[0][0], WEBHOOK)
        self.assertEqual(notifier.calls[0][1]["creative_rows"], 10)
        statuses = [call.kwargs["status"] for call in update.call_args_list]
        self.assertEqual(statuses, ["UPLOADED", "SENT", "SKIPPED"])
        self.assertNotIn(WEBHOOK, str(upsert.call_args_list))

    @patch("roi_control.agency_delivery.update_agency_delivery")
    @patch("roi_control.agency_delivery.upsert_agency_delivery")
    def test_uploaded_sheet_is_reused_when_notification_retries(
        self,
        upsert,
        update,
    ):
        upsert.return_value = {
            "id": 3,
            "status": "FAILED",
            "sheet_url": "https://example.test/existing-sheet",
            "sheet_token": "existing-token",
        }
        publisher = FakePublisher()
        notifier = FakeNotifier()
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory) / "20260804_棱镜_调控建议.xlsx"
            path.write_bytes(b"xlsx")
            outcomes = publish_agency_reports(
                run_id="run-2",
                reports=[
                    {
                        "agency_name": "小程序-代投-棱镜",
                        "report_version": "roi_agency_advice_v7",
                        "report": str(path),
                        "creative_rows": 10,
                        "ad_rows": 3,
                    }
                ],
                config=AgencyWebhookConfig(
                    enabled=True,
                    webhooks={"棱镜": WEBHOOK},
                ),
                publisher=publisher,
                notifier=notifier,
            )
        self.assertEqual(outcomes[0]["status"], "SENT")
        self.assertEqual(publisher.paths, [])
        self.assertEqual(
            notifier.calls[0][1]["sheet_url"],
            "https://example.test/existing-sheet",
        )
        self.assertEqual(update.call_args.kwargs["status"], "SENT")

    def test_webhook_card_contains_sheet_button(self):
        captured = {}

        def handler(request):
            captured.update(json.loads(request.content))
            return httpx.Response(200, json={"StatusCode": 0, "StatusMessage": "success"})

        notifier = AgencyWebhookNotifier()
        notifier.client.close()
        notifier.client = httpx.Client(transport=httpx.MockTransport(handler))
        try:
            code = notifier.send(
                WEBHOOK,
                title="20260804_棱镜_调控建议",
                sheet_url="https://example.test/sheet",
                creative_rows=10,
                ad_rows=3,
            )
        finally:
            notifier.close()
        self.assertEqual(code, "0")
        self.assertEqual(captured["msg_type"], "interactive")
        button = captured["card"]["elements"][1]["actions"][0]
        self.assertEqual(button["url"], "https://example.test/sheet")

    def test_main_chat_publish_still_uploads_once_and_notifies_each_chat(self):
        publisher = RoiFeishuPublisher.__new__(RoiFeishuPublisher)
        publisher.chat_ids = ["chat-1", "chat-2"]
        publisher.upload_workbook = Mock(
            return_value={
                "url": "https://example.test/main-sheet",
                "sheet_token": "main-token",
            }
        )
        publisher._token = Mock(return_value="tenant-token")
        publisher._send_card = Mock(side_effect=["message-1", "message-2"])

        result = publisher.publish(
            Path("main.xlsx"),
            run_id="run-main",
            batch_name="batch-main",
            summary="summary",
            requires_approval=True,
        )

        publisher.upload_workbook.assert_called_once_with(Path("main.xlsx"))
        self.assertEqual(publisher._send_card.call_count, 2)
        self.assertEqual(result["message_id"], "message-1")
        self.assertEqual(result["sheet_token"], "main-token")

    def test_service_failure_alert_uses_explicit_chat_and_red_card(self):
        publisher = RoiFeishuPublisher.__new__(RoiFeishuPublisher)
        publisher.client = Mock()
        publisher._token = Mock(return_value="tenant-token")
        publisher._json = Mock(return_value={"data": {"message_id": "alert-1"}})

        message_id = publisher.send_service_alert(
            title="日级ROI任务失败",
            content="任务执行失败",
            chat_id="chat-internal",
        )

        self.assertEqual(message_id, "alert-1")
        request = publisher.client.post.call_args
        self.assertEqual(request.kwargs["json"]["receive_id"], "chat-internal")
        self.assertEqual(request.kwargs["json"]["msg_type"], "interactive")
        card = json.loads(request.kwargs["json"]["content"])
        self.assertEqual(card["header"]["template"], "red")
        self.assertEqual(card["header"]["title"]["content"], "日级ROI任务失败")

    def test_service_failure_alert_falls_back_to_operator_chat(self):
        publisher = RoiFeishuPublisher.__new__(RoiFeishuPublisher)
        publisher.client = Mock()
        publisher._token = Mock(return_value="tenant-token")
        publisher._json = Mock(return_value={"data": {"message_id": "alert-2"}})

        with patch.dict(
            os.environ,
            {
                "ROI_FAILURE_FEISHU_CHAT_ID": "",
                "FEISHU_OPERATOR_CHAT_ID": "chat-operator",
            },
            clear=False,
        ):
            message_id = publisher.send_service_alert(
                title="日级ROI任务失败",
                content="任务执行失败",
            )

        self.assertEqual(message_id, "alert-2")
        request = publisher.client.post.call_args
        self.assertEqual(request.kwargs["json"]["receive_id"], "chat-operator")

    def test_alert_only_publisher_does_not_require_report_chat(self):
        with patch.dict(
            os.environ,
            {
                "FEISHU_APP_ID": "app-id",
                "FEISHU_APP_SECRET": "app-secret",
                "ROI_FEISHU_CHAT_ID": "",
                "FEISHU_AD_PROJECT_CHAT_ID": "",
                "RTC_COMMAND_CHAT_ID": "",
                "FEISHU_OPERATOR_CHAT_ID": "",
                "ROI_FAILURE_FEISHU_CHAT_ID": "chat-failure",
            },
            clear=True,
        ):
            publisher = RoiFeishuPublisher(require_chat_ids=False)
            try:
                self.assertEqual(publisher.chat_ids, [])
            finally:
                publisher.close()

    def test_formal_publisher_still_requires_report_chat(self):
        with patch.dict(
            os.environ,
            {
                "FEISHU_APP_ID": "app-id",
                "FEISHU_APP_SECRET": "app-secret",
            },
            clear=True,
        ):
            with self.assertRaisesRegex(RuntimeError, "ROI_FEISHU_CHAT_ID"):
                RoiFeishuPublisher()

    def test_daily_roi_nonzero_exit_triggers_failure_alert(self):
        previous_logging_disable = logging.root.manager.disable
        logging.disable(logging.CRITICAL)
        try:
            import run_daily_service as daily_service
        finally:
            logging.disable(previous_logging_disable)

        @contextmanager
        def acquired_lock():
            yield True

        internal_args = ["--internal-test"]
        with (
            patch.object(
                daily_service,
                "advisory_lock",
                return_value=acquired_lock(),
            ),
            patch.object(
                daily_service.subprocess,
                "run",
                return_value=Mock(returncode=7),
            ),
            patch.object(daily_service, "_notify_roi_failure") as notify,
        ):
            with self.assertRaisesRegex(RuntimeError, "exited with code 7"):
                daily_service._run_script(
                    "run_daily_roi.py",
                    "test-roi-lock",
                    internal_args,
                )

        notify.assert_called_once_with(
            extra_args=internal_args,
            error="process exited with code 7",
        )

    def test_disabled_internal_job_does_not_parse_agency_routes_at_startup(self):
        import run_daily_service as daily_service

        scheduler = Mock()
        scheduler.get_jobs.return_value = []
        with (
            patch.object(
                daily_service.RoiConfig,
                "from_env",
                return_value=RoiConfig(
                    daily_enabled=False,
                    internal_test_enabled=False,
                ),
            ),
            patch.object(
                daily_service.AgencyWebhookConfig,
                "from_env",
            ) as agency_from_env,
            patch.object(daily_service, "initialize_schema"),
            patch.object(daily_service, "_env_flag", return_value=False),
            patch.object(
                daily_service,
                "BlockingScheduler",
                return_value=scheduler,
            ),
        ):
            daily_service.main()

        agency_from_env.assert_not_called()
        scheduler.start.assert_called_once_with()


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