import os import unittest from datetime import date, datetime from decimal import Decimal from unittest.mock import Mock, patch import pandas as pd from operator_commands import ACTION_CONFIRM, ACTION_REJECT, parse_command from roi_control.config import RoiConfig from roi_control.execution import _execute_pause_ad, plan_scale_bid from roi_control.metrics import ENTITY_GZH, ENTITY_SELF, ENTITY_SELF_AD from roi_control.policy import ( ACTION_PAUSE_AD, ACTION_PAUSE_CREATIVE, ACTION_SCALE_BID, MODE_ACTIONABLE, MODE_NOTIFY_ONLY, annotate_execution, ) from roi_control.repository import _is_expired from roi_control.sheet_approval import RoiSheetClient, parse_approval_rows from tencent_client import ACTIVE_STATUS, SUSPEND_STATUS class RoiControlPolicyTest(unittest.TestCase): def test_mysql_naive_expiry_compares_with_aware_service_time(self): expires_at = datetime(2026, 7, 28, 12, 0) now = datetime.fromisoformat("2026-07-28T12:01:00+08:00") self.assertTrue(_is_expired(expires_at, now)) def test_actionability_is_separate_from_metric_recommendation(self): summary = pd.DataFrame( [ { "entity_type": ENTITY_SELF, "channel": "小程序投流-稳定", "账号id": "84502354", "广告id": "1001", "创意id": "2001", "动作": "关停", "动作原因": "低ROI", }, { "entity_type": ENTITY_SELF, "channel": "小程序投流-稳定", "账号id": "84502354", "广告id": "1001", "创意id": "2002", "动作": "扩量", "动作原因": "高ROI", }, { "entity_type": ENTITY_SELF, "channel": "小程序投流-稳定", "账号id": "99999999", "广告id": "3001", "创意id": "4001", "动作": "关停", "动作原因": "低ROI", }, { "entity_type": ENTITY_GZH, "channel": "公众号合作-即转-稳定", "合作方名": "合作方", "公众号名": "公众号", "动作": "扩量", "动作原因": "高ROI", }, { "entity_type": ENTITY_SELF_AD, "channel": "小程序投流-稳定", "账号id": "84502354", "广告id": "5001", "创意id": "", "动作": "关停", "动作原因": "广告级低ROI", }, ] ) annotated, snapshots, actions = annotate_execution( summary, {84502354}, "roi_20260725_north_star_roi_v1_roi_policy_v1", ) self.assertEqual(len(snapshots), 5) self.assertEqual( {action["action_type"] for action in actions}, {ACTION_PAUSE_CREATIVE, ACTION_PAUSE_AD, ACTION_SCALE_BID}, ) self.assertEqual( annotated.iloc[0]["执行模式"], MODE_ACTIONABLE, ) self.assertEqual(annotated.iloc[2]["执行模式"], MODE_NOTIFY_ONLY) self.assertEqual(annotated.iloc[3]["执行模式"], MODE_NOTIFY_ONLY) self.assertEqual(annotated.iloc[0]["审批选择"], "") self.assertEqual(annotated.iloc[0]["执行状态"], "待审批") self.assertEqual( annotated.iloc[0]["动作幂等键"], actions[0]["idempotency_key"] ) self.assertEqual(annotated.iloc[2]["审批选择"], "不可执行") self.assertEqual(annotated.iloc[4]["执行模式"], MODE_ACTIONABLE) self.assertEqual(annotated.iloc[4]["审批选择"], "") ad_action = next( action for action in actions if action["action_type"] == ACTION_PAUSE_AD ) self.assertIsNone(ad_action["dynamic_creative_id"]) self.assertTrue(ad_action["idempotency_key"].endswith(":5001")) def test_sheet_approval_parser_uses_hidden_idempotency_key(self): rows = [ ["账号id", "审批选择", "动作幂等键", "执行状态", "执行结果"], ["99999999", "批准", "roi_x:PAUSE_CREATIVE:1:2", "待审批", ""], ["1", "拒绝", "roi_x:PAUSE_CREATIVE:1:3", "待审批", ""], ["1", "随便填", "roi_x:PAUSE_CREATIVE:1:4", "待审批", ""], ] decisions = parse_approval_rows(rows) self.assertEqual( [(row["decision"], row["idempotency_key"]) for row in decisions], [ ("APPROVED", "roi_x:PAUSE_CREATIVE:1:2"), ("REJECTED", "roi_x:PAUSE_CREATIVE:1:3"), ], ) def test_sheet_client_reads_both_current_approval_sheets(self): client = RoiSheetClient.__new__(RoiSheetClient) client._token = Mock(return_value="token") client._sheet_ids = Mock( return_value=[ ("小程序创意级三日汇总", "creative-sheet"), ("小程序广告级三日汇总", "ad-sheet"), ] ) client._read_sheet_approvals = Mock( side_effect=[ [{"sheet_id": "creative-sheet", "idempotency_key": "creative"}], [{"sheet_id": "ad-sheet", "idempotency_key": "ad"}], ] ) decisions = client.read_approvals("spreadsheet") self.assertEqual( [decision["sheet_id"] for decision in decisions], ["creative-sheet", "ad-sheet"], ) def test_pause_ad_marks_prepared_then_success_after_readback(self): now = datetime.fromisoformat("2026-08-03T12:00:00+08:00") class FakeTencent: def get_ad(self, account_id, adgroup_id): self.get_args = (account_id, adgroup_id) return { "adgroup_id": adgroup_id, "configured_status": ACTIVE_STATUS, } def update_ad(self, account_id, adgroup_id, **kwargs): self.update_args = (account_id, adgroup_id, kwargs) return { "adgroup_id": adgroup_id, "configured_status": SUSPEND_STATUS, } client = FakeTencent() item = { "id": 99, "account_id": 84502354, "adgroup_id": 5001, "execution_status": "PENDING", } with patch("roi_control.execution.update_action_item") as update_item: _execute_pause_ad(item, client=client, now=now) self.assertEqual(client.get_args, (84502354, 5001)) self.assertEqual( client.update_args, (84502354, 5001, {"target_status": SUSPEND_STATUS}), ) self.assertEqual(update_item.call_count, 2) self.assertEqual( update_item.call_args_list[0].kwargs["execution_status"], "PREPARED" ) self.assertEqual( update_item.call_args_list[1].kwargs["execution_status"], "SUCCESS" ) self.assertEqual( update_item.call_args_list[1].kwargs["readback_status"], SUSPEND_STATUS ) def test_scale_actions_are_deduplicated_by_ad(self): rows = [ { "entity_type": ENTITY_SELF, "channel": "小程序投流-稳定", "账号id": "84502354", "广告id": "1001", "创意id": str(2000 + index), "动作": "扩量", "动作原因": "高ROI", } for index in range(3) ] _, _, actions = annotate_execution( pd.DataFrame(rows), {84502354}, "roi_20260725_north_star_roi_v1_roi_policy_v1", ) self.assertEqual(len(actions), 1) self.assertEqual(actions[0]["action_type"], ACTION_SCALE_BID) def test_observe_recommendation_never_creates_tencent_action(self): summary = pd.DataFrame( [ { "entity_type": ENTITY_SELF, "channel": "小程序投流-稳定", "账号id": "84502354", "广告id": "1001", "创意id": "2001", "动作": "观察", "动作原因": "连续两天表现不一致", } ] ) annotated, _, actions = annotate_execution( summary, {84502354}, "roi_20260730_m6_p3", ) self.assertEqual(actions, []) self.assertEqual(annotated.iloc[0]["执行模式"], MODE_NOTIFY_ONLY) self.assertEqual( annotated.iloc[0]["执行说明"], "观察行仅展示，不执行腾讯写操作", ) def test_invalid_apply_without_daily_job_is_rejected(self): environment = { "DAILY_ROI_ENABLED": "0", "ROI_APPLY_ENABLED": "1", } with patch.dict(os.environ, environment, clear=False): with self.assertRaisesRegex(ValueError, "requires DAILY_ROI_ENABLED"): RoiConfig.from_env() def test_sheet_approval_requires_roi_writes(self): environment = { "DAILY_ROI_ENABLED": "1", "ROI_APPLY_ENABLED": "0", "ROI_SHEET_APPROVAL_ENABLED": "1", } with patch.dict(os.environ, environment, clear=False): with self.assertRaisesRegex(ValueError, "requires ROI_APPLY_ENABLED"): RoiConfig.from_env() def test_roi_batch_commands_are_deterministic(self): confirm = parse_command("确认 roi_20260725_north_star_roi_v1") reject = parse_command("拒绝 roi_20260725_north_star_roi_v1") old_confirm = parse_command("确认 cmd_20260725120000_abcd1234") self.assertEqual(confirm.action, ACTION_CONFIRM) self.assertEqual(reject.action, ACTION_REJECT) self.assertEqual(confirm.command_id, "roi_20260725_north_star_roi_v1") self.assertEqual(old_confirm.action, ACTION_CONFIRM) def test_restored_intraday_boost_is_not_reapplied_by_roi(self): plan = plan_scale_bid( current_bid_fen=100, base_bid_fen=100, initial_base_bid_fen=100, boosted_date=date(2026, 7, 26), control_date=date(2026, 7, 26), intraday_ratio=Decimal("1.25"), scale_ratio=Decimal("1.10"), max_base_ratio=Decimal("2.00"), ) self.assertTrue(plan["allowed"]) self.assertEqual(plan["new_base_bid_fen"], 110) self.assertEqual(plan["target_bid_fen"], 110) self.assertTrue(plan["boosted_date_today"]) self.assertFalse(plan["preserve_intraday_boost"]) def test_active_intraday_boost_is_preserved_relative_to_new_base(self): plan = plan_scale_bid( current_bid_fen=125, base_bid_fen=100, initial_base_bid_fen=100, boosted_date=date(2026, 7, 26), control_date=date(2026, 7, 26), intraday_ratio=Decimal("1.25"), scale_ratio=Decimal("1.10"), max_base_ratio=Decimal("2.00"), ) self.assertEqual(plan["target_bid_fen"], 138) self.assertTrue(plan["preserve_intraday_boost"]) if __name__ == "__main__": unittest.main()