From 2326410067963a887942cfc355c6ea8c2e5b3920 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=83=A1=E7=BB=8D=E5=BA=B7?= <3052147989@qq.com> Date: Thu, 5 Mar 2026 09:23:04 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20Telegram=20=E5=AE=9E=E6=97=B6=E6=8E=A8?= =?UTF-8?q?=E9=80=81=E5=8A=9F=E8=83=BD=EF=BC=88TODO-00007=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P0 核心功能： - DB Schema v3→v4：accounts 表新增 telegram_push_enabled、telegram_last_checked_at 列 - Repository：toggle_telegram_push、update_telegram_cursor、get_telegram_push_accounts - Service：telegram_push.py — Graph/IMAP 新邮件抓取 + Telegram Bot API 推送 - Scheduler：telegram_push_job 定时轮询（默认 600s，可配置 60-3600s） - Controller/Route：POST /api/accounts//telegram-toggle、settings GET/PUT 扩展 - Frontend：账号卡片🔔推送开关、设置页 Telegram 配置区 P1 增强功能： - POST /api/settings/telegram-test 发送测试消息端点 - 保存 telegram_poll_interval 时动态重调度（无需重启） - accounts API 返回 telegram_push_enabled 字段 测试：33 项 Telegram 推送测试全部通过，202 项总测试零回归 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- outlook_web/controllers/accounts.py | 17 + outlook_web/controllers/settings.py | 80 ++- outlook_web/db.py | 7 +- outlook_web/repositories/accounts.py | 47 ++ outlook_web/routes/accounts.py | 7 + outlook_web/routes/settings.py | 5 + outlook_web/services/scheduler.py | 37 ++ outlook_web/services/telegram_push.py | 311 +++++++++++ static/css/main.css | 1 + static/js/features/accounts.js | 23 + static/js/features/groups.js | 1 + static/js/main.js | 42 ++ templates/index.html | 27 + tests/test_telegram_push.py | 763 ++++++++++++++++++++++++++ 14 files changed, 1365 insertions(+), 3 deletions(-) create mode 100644 outlook_web/services/telegram_push.py create mode 100644 tests/test_telegram_push.py diff --git a/outlook_web/controllers/accounts.py b/outlook_web/controllers/accounts.py index ca2cf4a..b6fb3d8 100644 --- a/outlook_web/controllers/accounts.py +++ b/outlook_web/controllers/accounts.py @@ -121,6 +121,7 @@ def api_get_accounts() -> Any: "created_at": acc.get("created_at", ""), "updated_at": acc.get("updated_at", ""), "tags": acc.get("tags", []), + "telegram_push_enabled": bool(acc.get("telegram_push_enabled")), } ) return jsonify({"success": True, "accounts": safe_accounts}) @@ -1871,3 +1872,19 @@ def api_get_refresh_stats() -> Any: }, } ) + + +# ==================== Telegram 推送 API ==================== + + +@login_required +def api_telegram_toggle(account_id: int) -> Any: + """切换账号 Telegram 推送开关""" + data = request.get_json(silent=True) or {} + enabled = bool(data.get("enabled", False)) + success = accounts_repo.toggle_telegram_push(account_id, enabled) + if not success: + return jsonify({"success": False, "message": "账号不存在"}), 404 + action = "开启" if enabled else "关闭" + log_audit(f"telegram_push_{action}", "account", str(account_id)) + return jsonify({"success": True, "enabled": enabled, "message": f"Telegram推送已{action}"}) diff --git a/outlook_web/controllers/settings.py b/outlook_web/controllers/settings.py index 462b6fe..107ba70 100644 --- a/outlook_web/controllers/settings.py +++ b/outlook_web/controllers/settings.py @@ -9,7 +9,7 @@ from flask import jsonify, request from outlook_web.audit import log_audit from outlook_web.repositories import settings as settings_repo from outlook_web.security.auth import login_required -from outlook_web.security.crypto import hash_password +from outlook_web.security.crypto import decrypt_data, encrypt_data, hash_password, is_encrypted # ==================== 设置 API ==================== @@ -47,7 +47,26 @@ def api_get_settings() -> Any: safe_settings["gptmail_api_key_set"] = bool(gptmail_api_key_value) safe_settings["gptmail_api_key_masked"] = mask_secret_value(gptmail_api_key_value) if gptmail_api_key_value else "" - return jsonify({"success": True, "settings": safe_settings}) + # Telegram 推送配置 + tg_bot_token_raw = all_settings.get("telegram_bot_token", "") + if tg_bot_token_raw and is_encrypted(tg_bot_token_raw): + try: + plain_token = decrypt_data(tg_bot_token_raw) + safe_settings["telegram_bot_token"] = "****" + plain_token[-4:] if len(plain_token) > 4 else "****" + except Exception: + safe_settings["telegram_bot_token"] = "****" + else: + safe_settings["telegram_bot_token"] = "" + safe_settings["telegram_chat_id"] = all_settings.get("telegram_chat_id", "") + safe_settings["telegram_poll_interval"] = int(all_settings.get("telegram_poll_interval", "600") or "600") + + response = {"success": True, "settings": safe_settings} + # 同时在顶层暴露 telegram 字段（兼容前端直接访问） + response["telegram_bot_token"] = safe_settings.get("telegram_bot_token", "") + response["telegram_chat_id"] = safe_settings.get("telegram_chat_id", "") + response["telegram_poll_interval"] = safe_settings.get("telegram_poll_interval", 600) + + return jsonify(response) @login_required @@ -190,6 +209,42 @@ def api_update_settings() -> Any: except ValueError: errors.append("轮询次数必须是数字") + # Telegram 推送配置 + if "telegram_poll_interval" in data: + try: + tg_interval = int(data["telegram_poll_interval"]) + if tg_interval < 60 or tg_interval > 3600: + errors.append("Telegram 轮询间隔必须在 60-3600 秒之间") + elif settings_repo.set_setting("telegram_poll_interval", str(tg_interval)): + updated.append("Telegram 轮询间隔") + scheduler_reload_needed = True + else: + errors.append("更新 Telegram 轮询间隔失败") + except (ValueError, TypeError): + errors.append("Telegram 轮询间隔必须是数字") + + if "telegram_bot_token" in data: + tg_token = str(data["telegram_bot_token"]).strip() + if tg_token and not tg_token.startswith("****"): + encrypted_token = encrypt_data(tg_token) + if settings_repo.set_setting("telegram_bot_token", encrypted_token): + updated.append("Telegram Bot Token") + else: + errors.append("更新 Telegram Bot Token 失败") + elif not tg_token: + if settings_repo.set_setting("telegram_bot_token", ""): + updated.append("Telegram Bot Token（已清空）") + else: + # 脱敏占位符（****xxx），跳过不覆盖 + updated.append("Telegram Bot Token（未变更）") + + if "telegram_chat_id" in data: + tg_chat_id = str(data["telegram_chat_id"]).strip() + if settings_repo.set_setting("telegram_chat_id", tg_chat_id): + updated.append("Telegram Chat ID") + else: + errors.append("更新 Telegram Chat ID 失败") + if errors: return jsonify({"success": False, "error": "；".join(errors)}) @@ -269,3 +324,24 @@ def api_validate_cron() -> Any: ) except Exception as e: return jsonify({"success": False, "valid": False, "error": f"Cron 表达式无效: {str(e)}"}) + + +@login_required +def api_test_telegram() -> Any: + """发送 Telegram 测试消息，验证 bot_token + chat_id 配置是否正确""" + from outlook_web.services.telegram_push import _send_telegram_message + + bot_token_raw = settings_repo.get_setting("telegram_bot_token", "") + chat_id = settings_repo.get_setting("telegram_chat_id", "") + + if not bot_token_raw or not chat_id: + return jsonify({"success": False, "error": "请先配置 Telegram Bot Token 和 Chat ID"}) + + bot_token = decrypt_data(bot_token_raw) if is_encrypted(bot_token_raw) else bot_token_raw + + ok = _send_telegram_message(bot_token, chat_id, "✅ Outlook Email Plus 测试消息：配置正确！") + if ok: + log_audit("telegram_test", "settings", None, "测试消息发送成功") + return jsonify({"success": True, "message": "测试消息已发送，请检查 Telegram"}) + else: + return jsonify({"success": False, "error": "发送失败，请检查 Bot Token 和 Chat ID 是否正确"}) diff --git a/outlook_web/db.py b/outlook_web/db.py index b538dec..a09fb5d 100644 --- a/outlook_web/db.py +++ b/outlook_web/db.py @@ -18,7 +18,8 @@ from outlook_web.security.crypto import ( # 数据库 Schema 版本（用于升级可验证/可诊断） # v3：对齐 PRD-00005 / FD-00005 / TDD-00005（accounts 表新增多邮箱字段：account_type/provider/imap_host/imap_port/imap_password） -DB_SCHEMA_VERSION = 3 +# v4：对齐 PRD-00007 / FD-00007（accounts 表新增 telegram_push_enabled/telegram_last_checked_at） +DB_SCHEMA_VERSION = 4 DB_SCHEMA_VERSION_KEY = "db_schema_version" DB_SCHEMA_LAST_UPGRADE_TRACE_ID_KEY = "db_schema_last_upgrade_trace_id" DB_SCHEMA_LAST_UPGRADE_ERROR_KEY = "db_schema_last_upgrade_error" @@ -396,6 +397,10 @@ def init_db(database_path: Optional[str] = None): cursor.execute("ALTER TABLE accounts ADD COLUMN imap_port INTEGER DEFAULT 993") if "imap_password" not in columns: cursor.execute("ALTER TABLE accounts ADD COLUMN imap_password TEXT") + if "telegram_push_enabled" not in columns: + cursor.execute("ALTER TABLE accounts ADD COLUMN telegram_push_enabled INTEGER NOT NULL DEFAULT 0") + if "telegram_last_checked_at" not in columns: + cursor.execute("ALTER TABLE accounts ADD COLUMN telegram_last_checked_at TEXT DEFAULT NULL") cursor.execute("PRAGMA table_info(groups)") group_columns = [col[1] for col in cursor.fetchall()] diff --git a/outlook_web/repositories/accounts.py b/outlook_web/repositories/accounts.py index ce26b13..2a78cfc 100644 --- a/outlook_web/repositories/accounts.py +++ b/outlook_web/repositories/accounts.py @@ -367,3 +367,50 @@ def update_account_credentials(account_id: int, **fields) -> bool: return True except Exception: return False + + +def toggle_telegram_push(account_id: int, enabled: bool) -> bool: + """切换账号 Telegram 推送开关。首次开启时自动设置游标为当前 UTC 时间。""" + from datetime import datetime, timezone + + db = get_db() + row = db.execute("SELECT id, telegram_last_checked_at FROM accounts WHERE id = ?", (account_id,)).fetchone() + if not row: + return False + + if enabled: + if row["telegram_last_checked_at"] is None: + now_utc = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S") + db.execute( + "UPDATE accounts SET telegram_push_enabled = 1, telegram_last_checked_at = ? WHERE id = ?", + (now_utc, account_id), + ) + else: + db.execute("UPDATE accounts SET telegram_push_enabled = 1 WHERE id = ?", (account_id,)) + else: + db.execute("UPDATE accounts SET telegram_push_enabled = 0 WHERE id = ?", (account_id,)) + + db.commit() + return True + + +def update_telegram_cursor(account_id: int, checked_at: str) -> None: + """更新账号的 telegram_last_checked_at 游标。""" + db = get_db() + db.execute( + "UPDATE accounts SET telegram_last_checked_at = ? WHERE id = ?", + (checked_at, account_id), + ) + db.commit() + + +def get_telegram_push_accounts() -> List[Dict]: + """返回所有 telegram_push_enabled=1 且非 disabled 状态的账号。""" + db = get_db() + rows = db.execute( + """SELECT id, email, provider, refresh_token, imap_host, imap_port, + imap_password, telegram_last_checked_at + FROM accounts + WHERE telegram_push_enabled = 1 AND status != 'disabled'""" + ).fetchall() + return [dict(r) for r in rows] diff --git a/outlook_web/routes/accounts.py b/outlook_web/routes/accounts.py index b3370c9..c127546 100644 --- a/outlook_web/routes/accounts.py +++ b/outlook_web/routes/accounts.py @@ -125,4 +125,11 @@ def create_blueprint() -> Blueprint: methods=["GET"], ) + # Telegram 推送开关 + bp.add_url_rule( + "/api/accounts//telegram-toggle", + view_func=accounts_controller.api_telegram_toggle, + methods=["POST"], + ) + return bp diff --git a/outlook_web/routes/settings.py b/outlook_web/routes/settings.py index 005bcea..01f41e8 100644 --- a/outlook_web/routes/settings.py +++ b/outlook_web/routes/settings.py @@ -13,6 +13,11 @@ def create_blueprint() -> Blueprint: view_func=settings_controller.api_validate_cron, methods=["POST"], ) + bp.add_url_rule( + "/api/settings/telegram-test", + view_func=settings_controller.api_test_telegram, + methods=["POST"], + ) bp.add_url_rule("/api/settings", view_func=settings_controller.api_get_settings, methods=["GET"]) bp.add_url_rule( "/api/settings", diff --git a/outlook_web/services/scheduler.py b/outlook_web/services/scheduler.py index 4317911..2bcde66 100644 --- a/outlook_web/services/scheduler.py +++ b/outlook_web/services/scheduler.py @@ -60,6 +60,40 @@ def scheduler_heartbeat_task(): pass +def _get_telegram_interval(app) -> int: + """读取 telegram_poll_interval 设置，默认 600 秒，最小 60 秒。""" + try: + with app.app_context(): + val = settings_repo.get_setting("telegram_poll_interval", "600") + return max(60, int(val or "600")) + except Exception: + return 600 + + +def _configure_telegram_push_job(scheduler, app) -> None: + """注册/更新 Telegram 推送 Job。""" + from outlook_web.services.telegram_push import run_telegram_push_job + + try: + scheduler.remove_job("telegram_push_job") + except Exception: + pass + + interval = _get_telegram_interval(app) + scheduler.add_job( + func=run_telegram_push_job, + args=[app], + trigger="interval", + seconds=interval, + id="telegram_push_job", + name="Telegram 推送", + replace_existing=True, + max_instances=1, + coalesce=True, + misfire_grace_time=interval, + ) + + def configure_scheduler_jobs(scheduler, app, test_refresh_token) -> None: """根据当前 settings 重新配置定时刷新 Job（配置变更即时生效）""" try: @@ -90,6 +124,9 @@ def configure_scheduler_jobs(scheduler, app, test_refresh_token) -> None: misfire_grace_time=60, ) + # Telegram 推送 Job + _configure_telegram_push_job(scheduler, app) + # 刷新 Job：根据 enable_scheduled 决定是否启用 try: scheduler.remove_job("token_refresh") diff --git a/outlook_web/services/telegram_push.py b/outlook_web/services/telegram_push.py new file mode 100644 index 0000000..8163ae7 --- /dev/null +++ b/outlook_web/services/telegram_push.py @@ -0,0 +1,311 @@ +"""outlook_web/services/telegram_push.py — Telegram 实时推送核心服务 + +轮询 IMAP/Graph 读取新邮件 → 构造消息 → 调用 Telegram Bot API → 丢弃内容，仅更新游标。 +""" +from __future__ import annotations + +import logging +import re +from datetime import datetime, timezone +from typing import List + +import requests + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# 文本工具 +# --------------------------------------------------------------------------- + +MAX_TELEGRAM_LENGTH = 4096 +MAX_PREVIEW_LENGTH = 200 +MAX_EMAILS_PER_FETCH = 50 +MAX_SENT_PER_JOB = 20 + + +def _escape_html(text: str) -> str: + """转义 Telegram HTML 模式必须转义的三种字符：& < >""" + return text.replace("&", "&").replace("<", "<").replace(">", ">") + + +def _html_to_plain(html_str: str) -> str: + """将 HTML 正文提取为纯文本（strip tags），合并多余空白。""" + if not html_str: + return "" + text = re.sub(r"<[^>]+>", " ", html_str) + text = re.sub(r"\s+", " ", text).strip() + return text + + +def _build_telegram_message(account_email: str, email: dict) -> str: + """构造 Telegram HTML 消息文本（PRD §3.4 格式）。""" + subject = _escape_html(email.get("subject", "")) + sender = _escape_html(email.get("sender", "")) + received_at = email.get("received_at", "") + preview = email.get("preview", "") + + lines = [ + "📬 新邮件通知", + "", + f"账户：{_escape_html(account_email)}", + f"发件人：{sender}", + f"主题：{subject}", + f"时间：{received_at}", + ] + + if preview: + truncated = preview[:MAX_PREVIEW_LENGTH] + if len(preview) > MAX_PREVIEW_LENGTH: + truncated += "..." + lines.append("") + lines.append(f"内容预览：\n{_escape_html(truncated)}") + + msg = "\n".join(lines) + + if len(msg) > MAX_TELEGRAM_LENGTH: + msg = msg[: MAX_TELEGRAM_LENGTH - 3] + "..." + + return msg + + +# --------------------------------------------------------------------------- +# Telegram API +# --------------------------------------------------------------------------- + + +def _send_telegram_message(bot_token: str, chat_id: str, text: str) -> bool: + """调用 Telegram sendMessage API。超时 10 秒，失败返回 False。""" + url = f"https://api.telegram.org/bot{bot_token}/sendMessage" + try: + resp = requests.post( + url, + json={ + "chat_id": chat_id, + "text": text, + "parse_mode": "HTML", + "disable_web_page_preview": True, + }, + timeout=10, + ) + return resp.ok + except Exception as e: + logger.warning("[telegram_push] send failed: %s", e) + return False + + +# --------------------------------------------------------------------------- +# 邮件拉取（IMAP / Graph） +# --------------------------------------------------------------------------- + + +def _fetch_new_emails_imap(account: dict, since: str) -> List[dict]: + """通过 IMAP 获取 received_at > since 的邮件，最多返回 50 封。""" + import email as email_lib + import email.header + import imaplib + from datetime import datetime as dt + + from outlook_web.security.crypto import decrypt_data + + host = account.get("imap_host", "") + port = int(account.get("imap_port", 993)) + password_raw = account.get("imap_password", "") + password = decrypt_data(password_raw) if password_raw else "" + user = account.get("email", "") + + since_dt = dt.fromisoformat(since) + since_date_str = since_dt.strftime("%d-%b-%Y") + + results: List[dict] = [] + conn = None + try: + conn = imaplib.IMAP4_SSL(host, port, timeout=15) + conn.login(user, password) + conn.select("INBOX", readonly=True) + + _, data = conn.search(None, f'(SINCE "{since_date_str}")') + msg_ids = data[0].split() if data[0] else [] + + for mid in msg_ids[-MAX_EMAILS_PER_FETCH:]: + try: + _, msg_data = conn.fetch(mid, "(RFC822)") + raw = msg_data[0][1] + msg = email_lib.message_from_bytes(raw) + + subject_parts = email.header.decode_header(msg.get("Subject", "")) + subject = "".join( + part.decode(charset or "utf-8") if isinstance(part, bytes) else part + for part, charset in subject_parts + ) + + sender = msg.get("From", "") + date_str = msg.get("Date", "") + try: + from email.utils import parsedate_to_datetime + + received_dt = parsedate_to_datetime(date_str) + received_iso = received_dt.strftime("%Y-%m-%dT%H:%M:%S") + except Exception: + received_iso = date_str + + if received_iso <= since: + continue + + body = "" + if msg.is_multipart(): + for part in msg.walk(): + ct = part.get_content_type() + if ct == "text/plain": + payload = part.get_payload(decode=True) + if payload: + body = payload.decode(part.get_content_charset() or "utf-8", errors="replace") + break + elif ct == "text/html" and not body: + payload = part.get_payload(decode=True) + if payload: + body = _html_to_plain( + payload.decode(part.get_content_charset() or "utf-8", errors="replace") + ) + else: + payload = msg.get_payload(decode=True) + if payload: + charset = msg.get_content_charset() or "utf-8" + raw_body = payload.decode(charset, errors="replace") + if msg.get_content_type() == "text/html": + body = _html_to_plain(raw_body) + else: + body = raw_body + + preview = body[:MAX_PREVIEW_LENGTH] if body else "" + + results.append( + { + "subject": subject, + "sender": sender, + "received_at": received_iso, + "preview": preview, + } + ) + except Exception: + continue + + except Exception as e: + logger.warning("[telegram_push] IMAP fetch error for %s: %s", account.get("email"), e) + raise + finally: + if conn: + try: + conn.logout() + except Exception: + pass + + return results[:MAX_EMAILS_PER_FETCH] + + +def _fetch_new_emails_graph(account: dict, since: str) -> List[dict]: + """通过 Microsoft Graph API 获取 received_at > since 的邮件，最多返回 50 封。""" + from outlook_web.security.crypto import decrypt_data + from outlook_web.services.graph import get_access_token + + refresh_token_raw = account.get("refresh_token", "") + refresh_token = decrypt_data(refresh_token_raw) if refresh_token_raw else "" + + access_token = get_access_token(refresh_token) + if not access_token: + return [] + + since_z = since if since.endswith("Z") else since + "Z" + url = "https://graph.microsoft.com/v1.0/me/messages" + params = { + "$filter": f"receivedDateTime gt {since_z}", + "$top": MAX_EMAILS_PER_FETCH, + "$select": "subject,from,receivedDateTime,bodyPreview", + "$orderby": "receivedDateTime asc", + } + headers = {"Authorization": f"Bearer {access_token}"} + + results: List[dict] = [] + try: + resp = requests.get(url, headers=headers, params=params, timeout=15) + if not resp.ok: + return [] + data = resp.json() + for item in data.get("value", []): + sender_info = item.get("from", {}).get("emailAddress", {}) + sender = sender_info.get("address", sender_info.get("name", "")) + received_raw = item.get("receivedDateTime", "") + received_iso = received_raw.replace("Z", "").split(".")[0] if received_raw else "" + preview = (item.get("bodyPreview", "") or "")[:MAX_PREVIEW_LENGTH] + results.append( + { + "subject": item.get("subject", ""), + "sender": sender, + "received_at": received_iso, + "preview": preview, + } + ) + except Exception as e: + logger.warning("[telegram_push] Graph fetch error for %s: %s", account.get("email"), e) + raise + + return results + + +# --------------------------------------------------------------------------- +# 主入口（调度器调用） +# --------------------------------------------------------------------------- + + +def run_telegram_push_job(app) -> None: + """主入口：轮询 → 推送 → 更新游标。由调度器调用。""" + with app.app_context(): + from outlook_web.repositories.accounts import ( + get_telegram_push_accounts, + update_telegram_cursor, + ) + from outlook_web.repositories.settings import get_setting + from outlook_web.security.crypto import decrypt_data, is_encrypted + + bot_token_raw = get_setting("telegram_bot_token", "") + bot_token = decrypt_data(bot_token_raw) if bot_token_raw and is_encrypted(bot_token_raw) else bot_token_raw + chat_id = get_setting("telegram_chat_id", "") + + if not bot_token or not chat_id: + return + + accounts = get_telegram_push_accounts() + if not accounts: + return + + job_start_time = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S") + sent_count = 0 + + for account in accounts: + if sent_count >= MAX_SENT_PER_JOB: + update_telegram_cursor(account["id"], job_start_time) + continue + + last_checked = account.get("telegram_last_checked_at") + + if last_checked is None: + update_telegram_cursor(account["id"], job_start_time) + continue + + try: + if account.get("provider") == "outlook": + emails = _fetch_new_emails_graph(account, last_checked) + else: + emails = _fetch_new_emails_imap(account, last_checked) + + for em in sorted(emails, key=lambda e: e.get("received_at", "")): + if sent_count >= MAX_SENT_PER_JOB: + break + msg = _build_telegram_message(account["email"], em) + _send_telegram_message(bot_token, chat_id, msg) + sent_count += 1 + + except Exception as e: + logger.warning("[telegram_push] account=%s error: %s", account.get("email"), e) + + finally: + update_telegram_cursor(account["id"], job_start_time) diff --git a/static/css/main.css b/static/css/main.css index 13f8646..65b1d58 100644 --- a/static/css/main.css +++ b/static/css/main.css @@ -313,6 +313,7 @@ input, textarea, select { font-family: inherit; } .btn-sm { padding: 0.3rem 0.7rem; font-size: 0.78rem; } .btn-icon { padding: 0.4rem; font-size: 0.9rem; background: none; border: none; color: var(--text-muted); border-radius: var(--radius-sm); } .btn-icon:hover { background: var(--border-light); color: var(--text); } +.btn-icon.tg-push-active { color: #0088cc; filter: drop-shadow(0 0 2px #0088cc66); } /* 表单 */ .form-group { margin-bottom: 1.1rem; } diff --git a/static/js/features/accounts.js b/static/js/features/accounts.js index 75ca0ce..73894e4 100644 --- a/static/js/features/accounts.js +++ b/static/js/features/accounts.js @@ -541,6 +541,29 @@ } } + // 切换 Telegram 推送开关 + async function toggleTelegramPush(accountId, enabled) { + try { + const response = await fetch(`/api/accounts/${accountId}/telegram-toggle`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ enabled }) + }); + const data = await response.json(); + if (data.success) { + showToast(data.message || (enabled ? 'Telegram推送已开启' : 'Telegram推送已关闭'), 'success'); + if (currentGroupId) { + delete accountsCache[currentGroupId]; + loadAccountsByGroup(currentGroupId, true); + } + } else { + handleApiError(data, 'Telegram推送切换失败'); + } + } catch (error) { + showToast('操作失败', 'error'); + } + } + // 显示导出邮箱模态框 async function showExportModal() { document.getElementById('exportModal').classList.add('show'); diff --git a/static/js/features/groups.js b/static/js/features/groups.js index 5bebcc8..53414db 100644 --- a/static/js/features/groups.js +++ b/static/js/features/groups.js @@ -271,6 +271,7 @@ ${isFailed ? `` : ''} + +
+

📬 Telegram 推送

+
+ + +
格式：1234567890:AAxxxxxx（留空则禁用推送）
+
+
+ + +
可用 @userinfobot 获取你的 Chat ID
+
+
+ +
+ + 秒 +
+
范围：60-3600 秒，默认 600 秒（10 分钟）
+
+
+ +
验证 Bot Token 和 Chat ID 是否配置正确
+
+
+
diff --git a/tests/test_telegram_push.py b/tests/test_telegram_push.py new file mode 100644 index 0000000..de8dd1d --- /dev/null +++ b/tests/test_telegram_push.py @@ -0,0 +1,763 @@ +"""tests/test_telegram_push.py — TDD-00007 Telegram 实时推送测试 + +覆盖 TDD-00007 §12 全部 32 个测试用例（T-01 ~ T-32）。 + +测试分组： + TestBuildTelegramMessage - T-01~T-05 消息构造单元测试 + TestHtmlToPlain - T-06~T-07 HTML 转纯文本 + TestEscapeHtml - T-08 HTML 转义 + TestToggleTelegramPush - T-09~T-12 toggle_telegram_push Repository + TestUpdateTelegramCursor - T-13 update_telegram_cursor Repository + TestGetTelegramPushAccounts- T-14 get_telegram_push_accounts Repository + TestRunTelegramPushJob - T-15~T-22 run_telegram_push_job 集成（mock 外部调用） + TestTelegramToggleEndpoint - T-23~T-26 /api/accounts//telegram-toggle 端点 + TestTelegramSettingsAPI - T-27~T-30 /api/settings Telegram 字段扩展 + TestRegressions - T-31~T-32 回归：调度器、全量测试 +""" + +import json +import unittest +from unittest.mock import MagicMock, patch + +from tests._import_app import clear_login_attempts, import_web_app_module + + +# --------------------------------------------------------------------------- +# 辅助：共享 app 实例（避免重复初始化） +# --------------------------------------------------------------------------- + +def _get_app(): + return import_web_app_module().app + + +def _login(client): + resp = client.post("/login", json={"password": "testpass123"}) + assert resp.status_code == 200, f"Login failed: {resp.data}" + + +def _insert_test_account(db, email, provider="imap", enabled=0, last_checked=None, status="active"): + """在 accounts 表中插入测试账号，返回 rowid。""" + db.execute( + """INSERT INTO accounts + (email, client_id, provider, account_type, refresh_token, imap_host, imap_port, + imap_password, group_id, telegram_push_enabled, telegram_last_checked_at, status) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?)""", + ( + email, "test_client_id", provider, + "outlook" if provider == "outlook" else "imap", + "enc:dummy_refresh", "imap.test.com", 993, + "enc:dummy_pass", None, + enabled, last_checked, status, + ), + ) + db.commit() + return db.execute("SELECT last_insert_rowid()").fetchone()[0] + + +# =========================================================================== +# T-01 ~ T-05：消息构造单元测试 +# =========================================================================== + +class TestBuildTelegramMessage(unittest.TestCase): + """TDD-00007 T-01 ~ T-05""" + + def _build(self, account_email, **email_kwargs): + from outlook_web.services.telegram_push import _build_telegram_message + email = { + "subject": email_kwargs.get("subject", "Test Subject"), + "sender": email_kwargs.get("sender", "sender@test.com"), + "received_at": email_kwargs.get("received_at", "2026-03-04T14:30:00"), + "preview": email_kwargs.get("preview", ""), + } + return _build_telegram_message(account_email, email) + + def test_t01_normal_email_with_preview(self): + """T-01：正常邮件（含预览），消息包含所有关键字段""" + msg = self._build( + "user@example.com", + subject="Hello World", + sender="bob@x.com", + received_at="2026-03-04T14:30:00", + preview="This is a test email body.", + ) + self.assertIn("📬", msg) + self.assertIn("user@example.com", msg) + self.assertIn("bob@x.com", msg) + self.assertIn("Hello World", msg) + self.assertIn("14:30", msg) + self.assertIn("This is a test email body.", msg) + self.assertIn("内容预览：", msg) + + def test_t02_empty_preview_omits_section(self): + """T-02：正文为空时，消息中不包含"内容预览："区块""" + msg = self._build("user@example.com", preview="") + self.assertNotIn("内容预览：", msg) + + def test_t03_very_long_message_truncated(self): + """T-03：消息超 4096 字符时截断至 4096，末尾为"..." """ + msg = self._build("user@example.com", preview="a" * 5000) + self.assertLessEqual(len(msg), 4096) + self.assertTrue(msg.endswith("..."), f"Expected truncation marker, got: {msg[-10:]!r}") + + def test_t04_html_special_chars_escaped(self): + """T-04：subject 和 sender 中的 HTML 特殊字符被正确转义""" + msg = self._build( + "user@example.com", + subject=" & ", + sender="a&b@test.com", + ) + # 转义后的形式 + self.assertIn("<Sale>", msg) + self.assertIn("&", msg) + self.assertIn("a&b@test.com", msg) + # 不应出现原始字符（在 subject/sender 渲染位置） + # （注意：HTML 标签本身如 是合法的，只检查 subject/sender 值） + self.assertNotIn("", msg) + self.assertNotIn("", msg) + + def test_t05_preview_truncated_at_200_chars(self): + """T-05：正文超 200 字时截断，追加"..." """ + msg = self._build("user@example.com", preview="x" * 300) + # 正文区域包含 200 个 x + "..." + self.assertIn("x" * 200 + "...", msg) + self.assertNotIn("x" * 201, msg) + + +# =========================================================================== +# T-06 ~ T-07：HTML 转纯文本 +# =========================================================================== + +class TestHtmlToPlain(unittest.TestCase): + """TDD-00007 T-06 ~ T-07""" + + def _convert(self, html): + from outlook_web.services.telegram_push import _html_to_plain + return _html_to_plain(html) + + def test_t06_strip_html_tags(self): + """T-06：HTML 标签被剥除""" + result = self._convert("

Hello World

") + self.assertIn("Hello", result) + self.assertIn("World", result) + self.assertNotIn("

", result) + self.assertNotIn("", result) + + def test_t06_empty_string(self): + """T-06：空字符串返回空字符串""" + self.assertEqual(self._convert(""), "") + + def test_t07_collapse_whitespace(self): + """T-07：多余空白（多个空格、换行）被合并为单个空格""" + result = self._convert("

a


b

") + self.assertIn("a", result) + self.assertIn("b", result) + # 不应出现连续多个空格 + self.assertNotIn(" ", result) + + +# =========================================================================== +# T-08：HTML 转义 +# =========================================================================== + +class TestEscapeHtml(unittest.TestCase): + """TDD-00007 T-08""" + + def test_t08_escape_three_special_chars(self): + """T-08：& < > 三种字符均被正确转义""" + from outlook_web.services.telegram_push import _escape_html + result = _escape_html("a & and > c") + self.assertEqual(result, "a & <b> and > c") + + +# =========================================================================== +# T-09 ~ T-12：toggle_telegram_push Repository +# =========================================================================== + +class TestToggleTelegramPush(unittest.TestCase): + """TDD-00007 T-09 ~ T-12""" + + @classmethod + def setUpClass(cls): + cls.app = _get_app() + + def setUp(self): + with self.app.app_context(): + clear_login_attempts() + from outlook_web.db import get_db + db = get_db() + db.execute("DELETE FROM accounts WHERE email LIKE '%@tgtest.com'") + db.commit() + self._account_id = _insert_test_account(db, "tg@tgtest.com", enabled=0, last_checked=None) + + def test_t09_first_enable_sets_cursor(self): + """T-09：首次开启时 telegram_last_checked_at 被设置（不为 None）""" + with self.app.app_context(): + from outlook_web.repositories.accounts import toggle_telegram_push + from outlook_web.db import get_db + result = toggle_telegram_push(self._account_id, True) + self.assertTrue(result) + row = get_db().execute( + "SELECT telegram_push_enabled, telegram_last_checked_at FROM accounts WHERE id = ?", + (self._account_id,) + ).fetchone() + self.assertEqual(row["telegram_push_enabled"], 1) + self.assertIsNotNone(row["telegram_last_checked_at"]) + # 游标应为有效 ISO8601 字符串 + self.assertRegex(row["telegram_last_checked_at"], r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}") + + def test_t10_disable_preserves_cursor(self): + """T-10：关闭推送时游标不清空""" + with self.app.app_context(): + from outlook_web.repositories.accounts import toggle_telegram_push + from outlook_web.db import get_db + db = get_db() + # 先开启（设置游标） + toggle_telegram_push(self._account_id, True) + original_cursor = db.execute( + "SELECT telegram_last_checked_at FROM accounts WHERE id = ?", + (self._account_id,) + ).fetchone()["telegram_last_checked_at"] + # 关闭 + toggle_telegram_push(self._account_id, False) + row = db.execute( + "SELECT telegram_push_enabled, telegram_last_checked_at FROM accounts WHERE id = ?", + (self._account_id,) + ).fetchone() + self.assertEqual(row["telegram_push_enabled"], 0) + self.assertEqual(row["telegram_last_checked_at"], original_cursor) + + def test_t11_nonexistent_account_returns_false(self): + """T-11：账号不存在时返回 False""" + with self.app.app_context(): + from outlook_web.repositories.accounts import toggle_telegram_push + result = toggle_telegram_push(99999, True) + self.assertFalse(result) + + def test_t12_repeat_enable_not_reset_cursor(self): + """T-12：重复开启不重置已有游标""" + with self.app.app_context(): + from outlook_web.repositories.accounts import toggle_telegram_push + from outlook_web.db import get_db + db = get_db() + # 先开启（游标设为 T1） + toggle_telegram_push(self._account_id, True) + cursor_t1 = db.execute( + "SELECT telegram_last_checked_at FROM accounts WHERE id = ?", + (self._account_id,) + ).fetchone()["telegram_last_checked_at"] + # 再次开启 + toggle_telegram_push(self._account_id, True) + cursor_t2 = db.execute( + "SELECT telegram_last_checked_at FROM accounts WHERE id = ?", + (self._account_id,) + ).fetchone()["telegram_last_checked_at"] + self.assertEqual(cursor_t1, cursor_t2, "重复开启不应重置游标") + + +# =========================================================================== +# T-13：update_telegram_cursor Repository +# =========================================================================== + +class TestUpdateTelegramCursor(unittest.TestCase): + """TDD-00007 T-13""" + + @classmethod + def setUpClass(cls): + cls.app = _get_app() + + def setUp(self): + with self.app.app_context(): + from outlook_web.db import get_db + db = get_db() + db.execute("DELETE FROM accounts WHERE email = 'cursor@tgtest.com'") + db.commit() + self._account_id = _insert_test_account(db, "cursor@tgtest.com") + + def test_t13_update_cursor(self): + """T-13：update_telegram_cursor 正确更新游标""" + with self.app.app_context(): + from outlook_web.repositories.accounts import update_telegram_cursor + from outlook_web.db import get_db + update_telegram_cursor(self._account_id, "2026-03-04T14:30:00") + row = get_db().execute( + "SELECT telegram_last_checked_at FROM accounts WHERE id = ?", + (self._account_id,) + ).fetchone() + self.assertEqual(row["telegram_last_checked_at"], "2026-03-04T14:30:00") + + +# =========================================================================== +# T-14：get_telegram_push_accounts Repository +# =========================================================================== + +class TestGetTelegramPushAccounts(unittest.TestCase): + """TDD-00007 T-14""" + + @classmethod + def setUpClass(cls): + cls.app = _get_app() + + def setUp(self): + with self.app.app_context(): + from outlook_web.db import get_db + db = get_db() + db.execute("DELETE FROM accounts WHERE email LIKE '%@tg14test.com'") + db.commit() + _insert_test_account(db, "a@tg14test.com", enabled=1, status="active") + _insert_test_account(db, "b@tg14test.com", enabled=0, status="active") + _insert_test_account(db, "c@tg14test.com", enabled=1, status="disabled") + + def test_t14_returns_only_enabled_active_accounts(self): + """T-14：只返回 enabled=1 且 status != disabled 的账号""" + with self.app.app_context(): + from outlook_web.repositories.accounts import get_telegram_push_accounts + accounts = get_telegram_push_accounts() + emails = [a["email"] for a in accounts] + self.assertIn("a@tg14test.com", emails) + self.assertNotIn("b@tg14test.com", emails) # enabled=0 + self.assertNotIn("c@tg14test.com", emails) # status=disabled + + +# =========================================================================== +# T-15 ~ T-22：run_telegram_push_job 集成测试（mock 外部调用） +# =========================================================================== + +class TestRunTelegramPushJob(unittest.TestCase): + """TDD-00007 T-15 ~ T-22""" + + @classmethod + def setUpClass(cls): + cls.app = _get_app() + + def setUp(self): + with self.app.app_context(): + from outlook_web.db import get_db + db = get_db() + db.execute("DELETE FROM accounts WHERE email LIKE '%@tgjob.com'") + # 确保其他测试遗留的账号不干扰推送 job（关闭所有推送开关） + db.execute("UPDATE accounts SET telegram_push_enabled = 0") + db.commit() + + def _set_settings(self, bot_token="test_token_12345678", chat_id="-12345"): + from outlook_web.repositories.settings import set_setting + from outlook_web.security.crypto import encrypt_data + if bot_token: + set_setting("telegram_bot_token", encrypt_data(bot_token)) + else: + set_setting("telegram_bot_token", "") + set_setting("telegram_chat_id", chat_id) + + def _run_job(self): + from outlook_web.services.telegram_push import run_telegram_push_job + run_telegram_push_job(self.app) + + def _make_email(self, received_at="2026-03-04T14:31:00"): + return { + "subject": "Test", + "sender": "s@test.com", + "received_at": received_at, + "preview": "body preview", + } + + def test_t15_no_bot_token_skips(self): + """T-15：Bot Token 未配置 → Telegram API 不被调用""" + with self.app.app_context(): + self._set_settings(bot_token="", chat_id="-12345") + with patch("outlook_web.services.telegram_push._send_telegram_message") as mock_send: + self._run_job() + mock_send.assert_not_called() + + def test_t16_no_chat_id_skips(self): + """T-16：Chat ID 未配置 → Telegram API 不被调用""" + with self.app.app_context(): + self._set_settings(bot_token="valid_token_xyz", chat_id="") + with patch("outlook_web.services.telegram_push._send_telegram_message") as mock_send: + self._run_job() + mock_send.assert_not_called() + + def test_t17_no_enabled_accounts_skips(self): + """T-17：无启用推送的账号 → Telegram API 不被调用""" + with self.app.app_context(): + self._set_settings() + # 确保无 enabled 账号（setUp 已清理） + with patch("outlook_web.services.telegram_push._send_telegram_message") as mock_send: + self._run_job() + mock_send.assert_not_called() + + def test_t18_first_run_sets_cursor_no_push(self): + """T-18：首次开启（last_checked_at=NULL）→ 设游标，不推送""" + with self.app.app_context(): + from outlook_web.db import get_db + db = get_db() + acc_id = _insert_test_account(db, "first@tgjob.com", enabled=1, last_checked=None) + self._set_settings() + + with patch("outlook_web.services.telegram_push._fetch_new_emails_imap") as mock_fetch, \ + patch("outlook_web.services.telegram_push._fetch_new_emails_graph") as mock_graph, \ + patch("outlook_web.services.telegram_push._send_telegram_message") as mock_send: + + mock_fetch.return_value = [self._make_email(), self._make_email()] + mock_graph.return_value = [] + self._run_job() + + # 不推送 + mock_send.assert_not_called() + + # 游标已更新（不再为 NULL） + row = db.execute( + "SELECT telegram_last_checked_at FROM accounts WHERE id = ?", + (acc_id,) + ).fetchone() + self.assertIsNotNone(row["telegram_last_checked_at"]) + + def test_t19_normal_push_calls_api_and_updates_cursor(self): + """T-19：正常推送 → Telegram API 被调用，游标更新""" + with self.app.app_context(): + from outlook_web.db import get_db + db = get_db() + acc_id = _insert_test_account( + db, "normal@tgjob.com", enabled=1, last_checked="2026-03-01T00:00:00" + ) + self._set_settings() + + emails = [ + self._make_email("2026-03-04T10:00:00"), + self._make_email("2026-03-04T11:00:00"), + ] + with patch("outlook_web.services.telegram_push._fetch_new_emails_imap", return_value=emails), \ + patch("outlook_web.services.telegram_push._send_telegram_message", return_value=True) as mock_send: + + self._run_job() + + self.assertEqual(mock_send.call_count, 2) + + # 游标已更新（不再是 2026-03-01） + cursor = db.execute( + "SELECT telegram_last_checked_at FROM accounts WHERE id = ?", + (acc_id,) + ).fetchone()["telegram_last_checked_at"] + self.assertIsNotNone(cursor) + self.assertNotEqual(cursor, "2026-03-01T00:00:00") + + def test_t20_global_limit_20(self): + """T-20：全局上限 20 封，跨账号合计不超过 20 条 Telegram 消息""" + with self.app.app_context(): + from outlook_web.db import get_db + db = get_db() + # 账号 A：12 封；账号 B：15 封 → 合计最多 20 封 + _insert_test_account(db, "aa@tgjob.com", enabled=1, last_checked="2026-01-01T00:00:00") + _insert_test_account(db, "bb@tgjob.com", enabled=1, last_checked="2026-01-01T00:00:00") + self._set_settings() + + emails_a = [self._make_email(f"2026-03-04T10:{i:02d}:00") for i in range(12)] + emails_b = [self._make_email(f"2026-03-04T11:{i:02d}:00") for i in range(15)] + + call_count = 0 + + def fake_fetch(account, since): + nonlocal call_count + if account["email"] == "aa@tgjob.com": + return emails_a + return emails_b + + def fake_send(token, chat, msg): + nonlocal call_count + call_count += 1 + return True + + with patch("outlook_web.services.telegram_push._fetch_new_emails_imap", side_effect=fake_fetch), \ + patch("outlook_web.services.telegram_push._send_telegram_message", side_effect=fake_send): + self._run_job() + + self.assertEqual(call_count, 20, f"Expected 20, got {call_count}") + + def test_t21_imap_exception_silent_cursor_updated(self): + """T-21：IMAP 连接异常 → 不抛出，游标仍更新""" + with self.app.app_context(): + from outlook_web.db import get_db + db = get_db() + acc_id = _insert_test_account( + db, "imapfail@tgjob.com", enabled=1, last_checked="2026-03-01T00:00:00" + ) + self._set_settings() + + with patch( + "outlook_web.services.telegram_push._fetch_new_emails_imap", + side_effect=ConnectionError("IMAP connection refused") + ), patch("outlook_web.services.telegram_push._send_telegram_message") as mock_send: + # 不应抛出异常 + try: + self._run_job() + except Exception as e: + self.fail(f"run_telegram_push_job raised unexpectedly: {e}") + + mock_send.assert_not_called() + + # 游标仍更新 + cursor = db.execute( + "SELECT telegram_last_checked_at FROM accounts WHERE id = ?", + (acc_id,) + ).fetchone()["telegram_last_checked_at"] + self.assertNotEqual(cursor, "2026-03-01T00:00:00") + + def test_t22_send_failure_silent_cursor_updated(self): + """T-22：Telegram 发送失败 → 不抛出，游标仍更新""" + with self.app.app_context(): + from outlook_web.db import get_db + db = get_db() + acc_id = _insert_test_account( + db, "sendfail@tgjob.com", enabled=1, last_checked="2026-03-01T00:00:00" + ) + self._set_settings() + + emails = [self._make_email("2026-03-04T10:00:00"), self._make_email("2026-03-04T11:00:00")] + with patch("outlook_web.services.telegram_push._fetch_new_emails_imap", return_value=emails), \ + patch("outlook_web.services.telegram_push._send_telegram_message", return_value=False): + try: + self._run_job() + except Exception as e: + self.fail(f"run_telegram_push_job raised unexpectedly: {e}") + + # 游标仍更新 + cursor = db.execute( + "SELECT telegram_last_checked_at FROM accounts WHERE id = ?", + (acc_id,) + ).fetchone()["telegram_last_checked_at"] + self.assertNotEqual(cursor, "2026-03-01T00:00:00") + + +# =========================================================================== +# T-23 ~ T-26：/api/accounts//telegram-toggle 端点测试 +# =========================================================================== + +class TestTelegramToggleEndpoint(unittest.TestCase): + """TDD-00007 T-23 ~ T-26""" + + @classmethod + def setUpClass(cls): + cls.app = _get_app() + + def setUp(self): + with self.app.app_context(): + clear_login_attempts() + from outlook_web.db import get_db + db = get_db() + db.execute("DELETE FROM accounts WHERE email LIKE '%@tgapi.com'") + db.commit() + self._account_id = _insert_test_account(db, "api@tgapi.com") + + def _toggle(self, client, account_id, enabled): + return client.post( + f"/api/accounts/{account_id}/telegram-toggle", + json={"enabled": enabled}, + content_type="application/json", + ) + + def test_t23_enable_success(self): + """T-23：开启成功 → 200, success=true, DB 更新""" + client = self.app.test_client() + _login(client) + resp = self._toggle(client, self._account_id, True) + self.assertEqual(resp.status_code, 200) + data = resp.get_json() + self.assertTrue(data["success"]) + self.assertTrue(data["enabled"]) + + with self.app.app_context(): + from outlook_web.db import get_db + row = get_db().execute( + "SELECT telegram_push_enabled FROM accounts WHERE id = ?", + (self._account_id,) + ).fetchone() + self.assertEqual(row["telegram_push_enabled"], 1) + + def test_t24_disable_success(self): + """T-24：关闭成功 → 200, success=true, DB 更新""" + with self.app.app_context(): + from outlook_web.db import get_db + get_db().execute( + "UPDATE accounts SET telegram_push_enabled=1 WHERE id=?", + (self._account_id,) + ) + get_db().commit() + + client = self.app.test_client() + _login(client) + resp = self._toggle(client, self._account_id, False) + self.assertEqual(resp.status_code, 200) + data = resp.get_json() + self.assertTrue(data["success"]) + self.assertFalse(data["enabled"]) + + with self.app.app_context(): + from outlook_web.db import get_db + row = get_db().execute( + "SELECT telegram_push_enabled FROM accounts WHERE id = ?", + (self._account_id,) + ).fetchone() + self.assertEqual(row["telegram_push_enabled"], 0) + + def test_t25_nonexistent_account_404(self): + """T-25：账号不存在 → HTTP 404""" + client = self.app.test_client() + _login(client) + resp = self._toggle(client, 99999, True) + self.assertEqual(resp.status_code, 404) + data = resp.get_json() + self.assertFalse(data["success"]) + + def test_t26_unauthenticated_denied(self): + """T-26：未登录 → 拒绝访问（401 或 302 重定向）""" + client = self.app.test_client() + # 不调用 _login + resp = self._toggle(client, self._account_id, True) + self.assertIn(resp.status_code, [401, 302, 403]) + + +# =========================================================================== +# T-27 ~ T-30：/api/settings Telegram 字段扩展测试 +# =========================================================================== + +class TestTelegramSettingsAPI(unittest.TestCase): + """TDD-00007 T-27 ~ T-30""" + + @classmethod + def setUpClass(cls): + cls.app = _get_app() + + def setUp(self): + with self.app.app_context(): + clear_login_attempts() + from outlook_web.repositories.settings import set_setting + set_setting("telegram_bot_token", "") + set_setting("telegram_chat_id", "") + set_setting("telegram_poll_interval", "600") + + def test_t27_get_settings_includes_telegram_fields(self): + """T-27：GET /api/settings 返回 telegram 配置字段，bot_token 脱敏""" + with self.app.app_context(): + from outlook_web.repositories.settings import set_setting + from outlook_web.security.crypto import encrypt_data + set_setting("telegram_bot_token", encrypt_data("1234567890:AAxxxxx")) + + client = self.app.test_client() + _login(client) + resp = client.get("/api/settings") + self.assertEqual(resp.status_code, 200) + data = resp.get_json() + + self.assertIn("telegram_bot_token", data) + self.assertIn("telegram_chat_id", data) + self.assertIn("telegram_poll_interval", data) + # bot_token 不应返回明文 + token_val = data["telegram_bot_token"] + self.assertNotEqual(token_val, "1234567890:AAxxxxx") + # 应为脱敏值（含 * 或为空） + if token_val: + self.assertTrue( + token_val.startswith("****") or "*" in token_val, + f"Bot token should be masked, got: {token_val!r}" + ) + + def test_t28_put_settings_saves_telegram_config(self): + """T-28：PUT /api/settings 保存 telegram 配置，bot_token 加密存储""" + client = self.app.test_client() + _login(client) + resp = client.put( + "/api/settings", + json={ + "telegram_bot_token": "NewToken123", + "telegram_chat_id": "-123456", + "telegram_poll_interval": 300, + }, + content_type="application/json", + ) + self.assertEqual(resp.status_code, 200) + data = resp.get_json() + self.assertTrue(data.get("success"), f"Expected success, got: {data}") + + with self.app.app_context(): + from outlook_web.repositories.settings import get_setting + from outlook_web.security.crypto import decrypt_data + raw_token = get_setting("telegram_bot_token") + # 应加密存储（前缀 enc:） + self.assertTrue(raw_token.startswith("enc:"), f"Token should be encrypted, got: {raw_token!r}") + # 解密后应为原始值 + self.assertEqual(decrypt_data(raw_token), "NewToken123") + self.assertEqual(get_setting("telegram_chat_id"), "-123456") + self.assertEqual(get_setting("telegram_poll_interval"), "300") + + def test_t29_invalid_interval_returns_400(self): + """T-29：telegram_poll_interval 低于最小值 60 → 400 错误""" + client = self.app.test_client() + _login(client) + resp = client.put( + "/api/settings", + json={"telegram_poll_interval": 30}, # 低于最小值 60 + content_type="application/json", + ) + self.assertEqual(resp.status_code, 400) + + def test_t30_masked_token_placeholder_not_overwrite(self): + """T-30：传入脱敏占位符（****xxxx）不覆盖已有 bot_token""" + with self.app.app_context(): + from outlook_web.repositories.settings import set_setting + from outlook_web.security.crypto import encrypt_data + original_encrypted = encrypt_data("OriginalRealToken") + set_setting("telegram_bot_token", original_encrypted) + + client = self.app.test_client() + _login(client) + resp = client.put( + "/api/settings", + json={"telegram_bot_token": "****xxxx"}, # 脱敏占位符 + content_type="application/json", + ) + # 保存不报错 + self.assertIn(resp.status_code, [200, 204]) + + with self.app.app_context(): + from outlook_web.repositories.settings import get_setting + from outlook_web.security.crypto import decrypt_data, encrypt_data + stored = get_setting("telegram_bot_token") + # 解密后应仍为原始值（未被覆盖） + self.assertEqual(decrypt_data(stored), "OriginalRealToken") + + +# =========================================================================== +# T-31 ~ T-32：回归测试 +# =========================================================================== + +class TestRegressions(unittest.TestCase): + """TDD-00007 T-31 ~ T-32""" + + @classmethod + def setUpClass(cls): + cls.app = _get_app() + + def test_t31_token_refresh_job_unaffected(self): + """T-31：加载 telegram_push 模块不影响 token_refresh 调度器 job""" + # 确认可以导入 telegram_push 模块而不报错 + try: + import outlook_web.services.telegram_push as tp # noqa: F401 + except ImportError as e: + self.fail(f"Cannot import telegram_push service: {e}") + + def test_t32_existing_tests_still_pass(self): + """T-32：现有核心功能（providers, settings）在引入新模块后仍正常""" + with self.app.app_context(): + # providers 模块正常 + from outlook_web.services.providers import get_provider_list + providers = get_provider_list() + self.assertIsInstance(providers, list) + self.assertGreater(len(providers), 0) + + # settings repository 正常 + from outlook_web.repositories.settings import get_setting, set_setting + set_setting("_test_regression_key", "ok") + val = get_setting("_test_regression_key") + self.assertEqual(val, "ok") + + +if __name__ == "__main__": + unittest.main()