Files
ewoooc/tests/test_openclaw_bot_menu_keyboards.py
OoO 1a886d962b
All checks were successful
CD Pipeline / deploy (push) Successful in 8m50s
fix(telegram): dedupe webhook+polling updates via shared DB guard
Webhook (Flask) and polling (momo-telegram-bot) consumed the same
Telegram update_id, causing /menu callbacks to fire twice. Add a
shared dedup module backed by telegram_update_dedup table (300s TTL,
60s cleanup) with in-memory fallback, wired into both paths.

Polling launcher now skips startup when webhook is configured to
prevent dual-consumption at the source.

38 tests across webhook, menu keyboards, telegram_api, dedup guard,
and trend bot service.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-02 12:01:04 +08:00

152 lines
5.0 KiB
Python

from datetime import timezone, timedelta
from pathlib import Path
def test_sales_menu_uses_injected_latest_date():
from services.openclaw_bot import menu_keyboards
menu_keyboards.configure_menu_keyboards(
latest_date_provider=lambda: "2026/04/30",
goals={},
taipei_tz=timezone(timedelta(hours=8)),
)
rows = menu_keyboards._submenu_sales()
assert rows[0][0]["callback_data"] == "cmd:sales:2026/04/30"
assert rows[0][0]["text"] == "📊 今日 (04/30)"
assert rows[0][1]["callback_data"] == "cmd:sales:2026/04/29"
assert rows[-1] == menu_keyboards._BACK
def test_goal_menu_uses_injected_goal_state():
from services.openclaw_bot import menu_keyboards
menu_keyboards.configure_menu_keyboards(goals={"daily": 1_500_000, "monthly": 0})
rows = menu_keyboards._submenu_goals()
assert rows[1][0]["text"] == "日目標 (150萬)"
assert rows[1][1]["text"] == "月目標 (未設)"
def test_category_menu_and_submenu_registry_are_stable():
from services.openclaw_bot import menu_keyboards
menu_keyboards.configure_menu_keyboards(latest_date_provider=lambda: "2026/04/30")
rows = menu_keyboards._submenu_category()
assert rows[0][0]["callback_data"] == "cmd:catdetail:美妝保養:2026/04/30"
assert rows[-2] == [{"text": "🗂 全分類清單", "callback_data": "cmd:category:2026/04/30"}]
assert rows[-1] == menu_keyboards._BACK
assert set(menu_keyboards._SUBMENUS) >= {
"main",
"sales",
"products",
"goals",
"analysis",
"trend",
"reports",
"market",
"competitor",
"competitor_ppt",
"category",
}
def test_competitor_menu_keeps_date_input_action():
from services.openclaw_bot import menu_keyboards
menu_keyboards.configure_menu_keyboards(latest_date_provider=lambda: "2026/04/30")
rows = menu_keyboards._submenu_competitor()
assert any("指定日期" in button["text"] and button["callback_data"] == "await:date_competitor"
for row in rows for button in row)
def test_competitor_ppt_menu_layout_stays_row_based():
from services.openclaw_bot import menu_keyboards
rows = menu_keyboards._submenu_competitor_ppt()
assert rows[0][0]["text"] == "📆 半年比較"
assert rows[0][1]["text"] == "🗓 年比較"
assert rows[-1] == menu_keyboards._BACK
def test_chunk_rows_respects_row_size():
from services.openclaw_bot import menu_keyboards
rows = menu_keyboards._chunk_rows(
[
('A', 'a'),
('B', 'b'),
('C', 'c'),
],
row_size=2,
)
assert rows == [
[{'text': 'A', 'callback_data': 'a'}, {'text': 'B', 'callback_data': 'b'}],
[{'text': 'C', 'callback_data': 'c'}],
]
def test_openclaw_routes_import_menu_keyboard_helpers():
route_source = Path("routes/openclaw_bot_routes.py").read_text(encoding="utf-8")
assert "from services.openclaw_bot.menu_keyboards import" in route_source
assert "configure_menu_keyboards(latest_date_provider=latest_date" in route_source
assert "def main_menu_keyboard():" not in route_source
def test_quick_menu_keyboard_has_two_column_layout():
from services.openclaw_bot import menu_keyboards
menu_keyboards.configure_menu_keyboards()
rows = menu_keyboards.quick_menu_keyboard()
assert len(rows) == 3
assert all(1 <= len(row) <= 2 for row in rows)
assert rows[0][0]['callback_data'].startswith('menu:')
def test_polling_telegram_bot_bridges_openclaw_menu_callbacks():
service_source = Path("services/telegram_bot_service.py").read_text(encoding="utf-8")
assert 'CommandHandler("menu", self.cmd_menu)' in service_source
assert "data.startswith(('menu:', 'cmd:', 'await:'))" in service_source
assert "openclaw_waiting_for" in service_source
def test_market_info_handlers_accept_text_mcp_contract():
route_source = Path("routes/openclaw_bot_routes.py").read_text(encoding="utf-8")
assert "def _send_mcp_text_result" in route_source
assert "data = get_upcoming_events()" in route_source
assert "get_upcoming_events(60)" not in route_source
def test_mcp_collector_has_stable_fallbacks():
source = Path("services/mcp_collector_service.py").read_text(encoding="utf-8")
assert "def _fallback_topic_content" in source
assert "def _looks_unreliable" in source
assert '["google_search"]' in source
assert "return self._fallback_topic_content" in source
def test_polling_menu_imports_openclaw_routes_for_runtime_configuration():
service_source = Path("services/telegram_bot_service.py").read_text(encoding="utf-8")
assert "from routes import openclaw_bot_routes as openclaw" in service_source
assert "openclaw.main_menu_keyboard()" in service_source
assert "openclaw._SUBMENUS.get(key)" in service_source
def test_help_keyboard_uses_reusable_quick_menu():
route_source = Path("routes/openclaw_bot_routes.py").read_text(encoding="utf-8")
assert "help_kb = quick_menu_keyboard()" in route_source