胡绍康andClaude Opus 4.6
64d8a6c357
核心变更： - 数据库：Schema v3，accounts 表新增 account_type/provider/imap_host/imap_port/imap_password 字段 - 后端：新增通用 IMAP 服务（imap_generic.py）与邮箱提供商配置（providers.py） - 导入：支持 Outlook/QQ/163/Gmail/自定义 IMAP 多格式批量导入，兼容注释行与导出格式 - 导出：按邮箱类型分组导出，保留 provider 信息便于重新导入 - 邮件读取：IMAP 账号走通用 IMAP 服务，Outlook 保持原 Graph/XOAUTH2 链路 - 验证码提取：IMAP 账号通过 IMAP 获取邮件详情后提取 - 安全约束：IMAP 账号禁用远程删除，敏感字段（imap_password）加密存储 - 前端：新增邮箱类型下拉、自定义 IMAP 配置区、provider 标签显示 - 调度器：定时刷新仅处理 Outlook 账号（IMAP 无 OAuth token 刷新语义） - 日志：配置 outlook_web 命名空间日志输出到 stderr 技术细节： - 保持 Outlook 旧格式完全兼容（四段式：email----password----client_id----refresh_token） - IMAP 账号使用空字符串占位 client_id/refresh_token（满足 DB NOT NULL 约束） - 支持导出格式智能识别（5 段 custom、3 段 provider、2 段配合输入框） - 前端 DOMPurify 未加载时回退为基本 HTML 过滤（CDN 不可达场景） Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
356 lines
12 KiB
Python
356 lines
12 KiB
Python
import unittest
|
||
import uuid
|
||
from unittest.mock import patch
|
||
|
||
from tests._import_app import clear_login_attempts, import_web_app_module
|
||
|
||
|
||
class MultiMailboxSupportTests(unittest.TestCase):
|
||
"""
|
||
对齐：PRD-00005 / FD-00005 / TDD-00005 / TEST-00005
|
||
目标：验证多邮箱（Outlook + IMAP provider）核心能力与回归门禁。
|
||
"""
|
||
|
||
@classmethod
|
||
def setUpClass(cls):
|
||
cls.module = import_web_app_module()
|
||
cls.app = cls.module.app
|
||
|
||
def setUp(self):
|
||
with self.app.app_context():
|
||
clear_login_attempts()
|
||
|
||
def _login(self, client, password: str = "testpass123"):
|
||
resp = client.post("/login", json={"password": password})
|
||
self.assertEqual(resp.status_code, 200)
|
||
data = resp.get_json()
|
||
self.assertEqual(data.get("success"), True)
|
||
|
||
def _default_group_id(self) -> int:
|
||
conn = self.module.create_sqlite_connection()
|
||
try:
|
||
row = conn.execute(
|
||
"SELECT id FROM groups WHERE name = '默认分组' LIMIT 1"
|
||
).fetchone()
|
||
return int(row["id"]) if row else 1
|
||
finally:
|
||
conn.close()
|
||
|
||
def test_db_schema_v3_has_multi_mailbox_columns(self):
|
||
conn = self.module.create_sqlite_connection()
|
||
try:
|
||
cols = conn.execute("PRAGMA table_info(accounts)").fetchall()
|
||
names = {c[1] for c in cols} # (cid, name, type, notnull, dflt_value, pk)
|
||
finally:
|
||
conn.close()
|
||
|
||
self.assertIn("account_type", names)
|
||
self.assertIn("provider", names)
|
||
self.assertIn("imap_host", names)
|
||
self.assertIn("imap_port", names)
|
||
self.assertIn("imap_password", names)
|
||
|
||
def test_providers_api_returns_fixed_order(self):
|
||
client = self.app.test_client()
|
||
self._login(client)
|
||
|
||
resp = client.get("/api/providers")
|
||
self.assertEqual(resp.status_code, 200)
|
||
data = resp.get_json()
|
||
self.assertEqual(data.get("success"), True)
|
||
|
||
providers = data.get("providers") or []
|
||
self.assertEqual(len(providers), 8)
|
||
self.assertEqual(providers[0].get("key"), "outlook")
|
||
self.assertEqual(providers[-1].get("key"), "custom")
|
||
|
||
keys = [p.get("key") for p in providers]
|
||
self.assertIn("qq", keys)
|
||
self.assertIn("163", keys)
|
||
|
||
def test_provider_folder_candidates_contains_utf7_for_qq_junk(self):
|
||
from outlook_web.services.providers import get_imap_folder_candidates
|
||
|
||
candidates = get_imap_folder_candidates("qq", "junkemail")
|
||
self.assertIn("&V4NXPpCuTvY-", candidates)
|
||
|
||
default_candidates = get_imap_folder_candidates("unknown", "inbox")
|
||
self.assertIn("INBOX", default_candidates)
|
||
|
||
def test_import_imap_qq_stores_encrypted_password_and_imap_fields(self):
|
||
client = self.app.test_client()
|
||
self._login(client)
|
||
|
||
unique = uuid.uuid4().hex
|
||
email_addr = f"qq_{unique}@qq.com"
|
||
imap_pwd = f"auth_{unique}"
|
||
|
||
resp = client.post(
|
||
"/api/accounts",
|
||
json={
|
||
"provider": "qq",
|
||
"account_string": f"{email_addr}----{imap_pwd}",
|
||
"group_id": self._default_group_id(),
|
||
},
|
||
)
|
||
self.assertEqual(resp.status_code, 200)
|
||
data = resp.get_json()
|
||
self.assertEqual(data.get("success"), True)
|
||
|
||
conn = self.module.create_sqlite_connection()
|
||
try:
|
||
row = conn.execute(
|
||
"""
|
||
SELECT account_type, provider, imap_host, imap_port, imap_password, client_id, refresh_token
|
||
FROM accounts
|
||
WHERE email = ?
|
||
LIMIT 1
|
||
""",
|
||
(email_addr,),
|
||
).fetchone()
|
||
finally:
|
||
conn.close()
|
||
|
||
self.assertIsNotNone(row)
|
||
self.assertEqual(row["account_type"], "imap")
|
||
self.assertEqual(row["provider"], "qq")
|
||
self.assertEqual(row["imap_host"], "imap.qq.com")
|
||
self.assertEqual(int(row["imap_port"]), 993)
|
||
self.assertEqual(row["client_id"], "")
|
||
self.assertEqual(row["refresh_token"], "")
|
||
|
||
stored_imap_pwd = row["imap_password"]
|
||
self.assertTrue(stored_imap_pwd)
|
||
self.assertNotEqual(stored_imap_pwd, imap_pwd)
|
||
self.assertTrue(stored_imap_pwd.startswith("enc:"))
|
||
|
||
def test_emails_api_routes_to_imap_generic_by_account_type(self):
|
||
client = self.app.test_client()
|
||
self._login(client)
|
||
|
||
unique = uuid.uuid4().hex
|
||
email_addr = f"imap_{unique}@example.com"
|
||
|
||
encrypted_pwd = self.module.encrypt_data("pw_" + unique)
|
||
|
||
conn = self.module.create_sqlite_connection()
|
||
try:
|
||
conn.execute(
|
||
"""
|
||
INSERT INTO accounts (email, password, client_id, refresh_token, account_type, provider, imap_host, imap_port, imap_password, group_id, remark, status)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
""",
|
||
(
|
||
email_addr,
|
||
"",
|
||
"",
|
||
"",
|
||
"imap",
|
||
"qq",
|
||
"imap.qq.com",
|
||
993,
|
||
encrypted_pwd,
|
||
self._default_group_id(),
|
||
"",
|
||
"active",
|
||
),
|
||
)
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
|
||
fake_result = {
|
||
"success": True,
|
||
"emails": [
|
||
{
|
||
"id": "1",
|
||
"subject": "s",
|
||
"from": "f",
|
||
"date": "d",
|
||
"is_read": True,
|
||
"has_attachments": False,
|
||
"body_preview": "p",
|
||
}
|
||
],
|
||
"method": "IMAP (Generic)",
|
||
"has_more": False,
|
||
}
|
||
|
||
with patch(
|
||
"outlook_web.controllers.emails.get_emails_imap_generic",
|
||
return_value=fake_result,
|
||
):
|
||
resp = client.get(f"/api/emails/{email_addr}?folder=inbox&skip=0&top=20")
|
||
self.assertEqual(resp.status_code, 200)
|
||
self.assertEqual(resp.get_json(), fake_result)
|
||
|
||
delete_resp = client.post(
|
||
"/api/emails/delete", json={"email": email_addr, "ids": ["1"]}
|
||
)
|
||
self.assertEqual(delete_resp.status_code, 400)
|
||
delete_data = delete_resp.get_json()
|
||
self.assertEqual(delete_data.get("success"), False)
|
||
err = delete_data.get("error") or {}
|
||
self.assertIsInstance(err, dict)
|
||
self.assertIn("不支持远程删除", err.get("message", ""))
|
||
|
||
def test_imap_generic_connect_error_does_not_leak_password(self):
|
||
from outlook_web.services.imap_generic import get_emails_imap_generic
|
||
|
||
secret_pwd = "top-secret-imap-password"
|
||
result = get_emails_imap_generic(
|
||
email_addr="u@example.com",
|
||
imap_password=secret_pwd,
|
||
imap_host="", # 触发 ValueError -> connect failed
|
||
imap_port=993,
|
||
folder="inbox",
|
||
provider="qq",
|
||
skip=0,
|
||
top=1,
|
||
)
|
||
self.assertEqual(result.get("success"), False)
|
||
self.assertEqual(result.get("error_code"), "IMAP_CONNECT_FAILED")
|
||
|
||
# 确保返回内容不包含明文密码
|
||
self.assertNotIn(secret_pwd, str(result))
|
||
|
||
def test_scheduler_skips_imap_accounts(self):
|
||
from outlook_web.services.scheduler import scheduled_refresh_task
|
||
|
||
unique = uuid.uuid4().hex
|
||
outlook_email = f"out_{unique}@outlook.com"
|
||
imap_email = f"imap_{unique}@example.com"
|
||
|
||
conn = self.module.create_sqlite_connection()
|
||
try:
|
||
conn.execute(
|
||
"""
|
||
INSERT INTO accounts (email, password, client_id, refresh_token, account_type, provider, group_id, remark, status)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
""",
|
||
(
|
||
outlook_email,
|
||
"",
|
||
"client_" + unique,
|
||
self.module.encrypt_data("rt_" + unique),
|
||
"outlook",
|
||
"outlook",
|
||
self._default_group_id(),
|
||
"",
|
||
"active",
|
||
),
|
||
)
|
||
conn.execute(
|
||
"""
|
||
INSERT INTO accounts (email, password, client_id, refresh_token, account_type, provider, imap_host, imap_port, imap_password, group_id, remark, status)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
""",
|
||
(
|
||
imap_email,
|
||
"",
|
||
"",
|
||
"",
|
||
"imap",
|
||
"qq",
|
||
"imap.qq.com",
|
||
993,
|
||
self.module.encrypt_data("pw_" + unique),
|
||
self._default_group_id(),
|
||
"",
|
||
"active",
|
||
),
|
||
)
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
|
||
called = []
|
||
|
||
def fake_test_refresh_token(client_id, refresh_token, proxy_url):
|
||
called.append((client_id, refresh_token, proxy_url))
|
||
return True, "ok"
|
||
|
||
with patch("outlook_web.services.scheduler.time.sleep", return_value=None), patch(
|
||
"outlook_web.services.scheduler.acquire_distributed_lock",
|
||
return_value=(True, {}),
|
||
), patch("outlook_web.services.scheduler.release_distributed_lock", return_value=None):
|
||
scheduled_refresh_task(self.app, fake_test_refresh_token)
|
||
|
||
self.assertGreaterEqual(len(called), 1)
|
||
self.assertTrue(any(cid == "client_" + unique for cid, _, _ in called))
|
||
self.assertFalse(any(cid == "" for cid, _, _ in called)) # 不应包含 IMAP 账号（空 client_id）
|
||
|
||
def test_export_format_outlook_first_then_imap_grouped(self):
|
||
client = self.app.test_client()
|
||
self._login(client)
|
||
|
||
unique = uuid.uuid4().hex
|
||
outlook_email = f"exp_out_{unique}@outlook.com"
|
||
imap_email = f"exp_qq_{unique}@qq.com"
|
||
|
||
conn = self.module.create_sqlite_connection()
|
||
try:
|
||
conn.execute(
|
||
"""
|
||
INSERT INTO accounts (email, password, client_id, refresh_token, account_type, provider, group_id, remark, status)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
""",
|
||
(
|
||
outlook_email,
|
||
self.module.encrypt_data("p_" + unique),
|
||
"cid_" + unique,
|
||
self.module.encrypt_data("rt_" + unique),
|
||
"outlook",
|
||
"outlook",
|
||
self._default_group_id(),
|
||
"",
|
||
"active",
|
||
),
|
||
)
|
||
conn.execute(
|
||
"""
|
||
INSERT INTO accounts (email, password, client_id, refresh_token, account_type, provider, imap_host, imap_port, imap_password, group_id, remark, status)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
""",
|
||
(
|
||
imap_email,
|
||
"",
|
||
"",
|
||
"",
|
||
"imap",
|
||
"qq",
|
||
"imap.qq.com",
|
||
993,
|
||
self.module.encrypt_data("imap_pw_" + unique),
|
||
self._default_group_id(),
|
||
"",
|
||
"active",
|
||
),
|
||
)
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
|
||
verify = client.post("/api/export/verify", json={"password": "testpass123"})
|
||
self.assertEqual(verify.status_code, 200)
|
||
verify_data = verify.get_json()
|
||
self.assertEqual(verify_data.get("success"), True)
|
||
token = verify_data.get("verify_token")
|
||
self.assertTrue(token)
|
||
|
||
export_resp = client.get("/api/accounts/export", headers={"X-Export-Token": token})
|
||
self.assertEqual(export_resp.status_code, 200)
|
||
content = export_resp.get_data(as_text=True)
|
||
|
||
outlook_pos = content.find("# === Outlook 账号 ===")
|
||
imap_pos = content.find("# === IMAP 账号（QQ 邮箱）===")
|
||
self.assertNotEqual(outlook_pos, -1)
|
||
self.assertNotEqual(imap_pos, -1)
|
||
self.assertLess(outlook_pos, imap_pos)
|
||
|
||
self.assertIn(outlook_email, content)
|
||
self.assertIn(imap_email, content)
|
||
self.assertIn("----qq", content) # IMAP 行格式：email----imap_password----provider
|
||
|
||
|