浅念
207 lines
9.1 KiB
Python
207 lines
9.1 KiB
Python
"""邮件渲染单测: 标签主题/HTML转义/内部key过滤(D2/D12)。不触网。"""
|
||
import asyncio
|
||
|
||
from app.services import email_service
|
||
|
||
|
||
def _capture_send(monkeypatch):
|
||
captured = {}
|
||
|
||
async def fake_send(msg):
|
||
captured["msg"] = msg
|
||
return True
|
||
|
||
monkeypatch.setattr(email_service, "_send", fake_send)
|
||
return captured
|
||
|
||
|
||
def _html(msg) -> str:
|
||
return msg.get_payload()[0].get_payload(decode=True).decode("utf-8")
|
||
|
||
|
||
def test_lead_email_subject_tag_and_escaped_body(monkeypatch):
|
||
captured = _capture_send(monkeypatch)
|
||
ok = asyncio.run(email_service.send_lead_email(
|
||
customer_profile={"目标国家": "肯尼亚", "_internal": "x", "备注": "<b>bold</b>"},
|
||
contact_info={"phone": "+254712345678", "wechat": "<i>wx</i>"},
|
||
reasoning="<b>强意向</b>",
|
||
conversation_messages=[
|
||
{"role": "user", "content": "<script>alert(1)</script>"},
|
||
{"role": "assistant", "content": "ok"},
|
||
],
|
||
intent_level="interested",
|
||
))
|
||
assert ok is True
|
||
subject = str(captured["msg"]["Subject"])
|
||
assert "[新询盘·意向强]" in subject and "肯尼亚" in subject
|
||
body = _html(captured["msg"])
|
||
assert "<script>" not in body # D12: 转义生效
|
||
assert "<script>" in body
|
||
assert "_internal" not in body # 内部 key 不渲染
|
||
assert "<b>bold</b>" not in body
|
||
assert "<b>bold</b>" in body
|
||
assert "<i>wx</i>" not in body
|
||
assert "<i>wx</i>" in body
|
||
assert captured["msg"].as_string().isascii()
|
||
|
||
|
||
def test_lead_email_unevaluated_tag(monkeypatch):
|
||
captured = _capture_send(monkeypatch)
|
||
asyncio.run(email_service.send_lead_email(
|
||
customer_profile={}, contact_info={"email": "a@b.com"},
|
||
reasoning="", conversation_messages=[], intent_level="",
|
||
))
|
||
assert "[新询盘·未评估]" in str(captured["msg"]["Subject"])
|
||
|
||
|
||
def test_returning_and_correction_emails(monkeypatch):
|
||
captured = _capture_send(monkeypatch)
|
||
assert asyncio.run(
|
||
email_service.send_returning_lead_email({"whatsapp": "+254700000000"})
|
||
) is True
|
||
assert "[线索回访]" in str(captured["msg"]["Subject"])
|
||
assert asyncio.run(
|
||
email_service.send_correction_email({"email": "a@b.com"})
|
||
) is True
|
||
assert "[询盘·更正1]" in str(captured["msg"]["Subject"])
|
||
|
||
|
||
# ---- Task 6(spec §5 首封补全号矩阵): (confirmed, source, 有无 e164) × 显示值+标注 ----
|
||
|
||
# 等待/未确认态: 裸号 + SA IP → 高置信建议(attempts=1 = 追问轮轮末已耗一轮)
|
||
_CC_IP_PROPOSED = {"e164": "+966501234567", "intl": "+966 50 123 4567",
|
||
"confirmed": False, "attempts": 1, "source": "ip_proposed",
|
||
"canon": "0501234567"}
|
||
# 裸写 IP 国码前缀,静默规范化(confirmed=true,码是客户自己写的)
|
||
_CC_PREFIX_NORM = {"e164": "+966501234567", "intl": "+966 50 123 4567",
|
||
"confirmed": True, "attempts": 0, "source": "prefix_normalized",
|
||
"canon": "966501234567"}
|
||
# 客户报国名重算后确认(埃及 = +20)
|
||
_CC_CUSTOMER_STATED = {"e164": "+20501234567", "intl": "+20 50 123 4567",
|
||
"confirmed": True, "attempts": 1, "source": "customer_stated",
|
||
"canon": "0501234567"}
|
||
# attempts 耗尽落定(标注保留"未经客户确认",spec §4)
|
||
_CC_FALLBACK = {"e164": "+966501234567", "intl": "+966 50 123 4567",
|
||
"confirmed": False, "attempts": 2, "source": "fallback",
|
||
"canon": "0501234567"}
|
||
# 无建议国(IP 无效): 无 e164
|
||
_CC_NO_IP = {"confirmed": False, "attempts": 1, "source": "no_ip",
|
||
"canon": "0501234567"}
|
||
# Task 3 调和防御: fallback-无-e164(迁移分支可产出,有机状态机不产生)
|
||
_CC_FALLBACK_NO_E164 = {"confirmed": False, "attempts": 0, "source": "fallback",
|
||
"canon": "0501234567"}
|
||
|
||
|
||
def test_lead_email_cc_unconfirmed_e164_with_note(monkeypatch):
|
||
"""首封渲染(未确认态): 正文 e164 +（区号按 IP 推测，未经客户确认）；
|
||
主题行 e164 无标注(主题拥挤)；裸号原文被补全号替换。"""
|
||
captured = _capture_send(monkeypatch)
|
||
assert asyncio.run(email_service.send_lead_email(
|
||
customer_profile={"目标国家": "沙特阿拉伯"},
|
||
contact_info={"whatsapp": "0501234567"},
|
||
reasoning="", conversation_messages=[],
|
||
intent_level="interested",
|
||
cc_info={"whatsapp": dict(_CC_IP_PROPOSED)},
|
||
)) is True
|
||
subject = str(captured["msg"]["Subject"])
|
||
body = _html(captured["msg"])
|
||
assert "+966501234567（区号按 IP 推测，未经客户确认）" in body
|
||
assert "+966501234567" in subject # 主题行号码用 e164
|
||
assert "区号按 IP 推测" not in subject # 主题不加标注
|
||
assert "（缺国家区号）" not in body
|
||
assert "0501234567" not in body # 裸号原文不再出现
|
||
|
||
|
||
def test_lead_email_cc_confirmed_sources_no_note(monkeypatch):
|
||
"""confirmed=true(prefix_normalized/customer_stated): e164 无标注——
|
||
码是客户自己写的/客户确认过的,零披露。"""
|
||
for state, raw in ((dict(_CC_PREFIX_NORM), "966501234567"),
|
||
(dict(_CC_CUSTOMER_STATED), "0501234567")):
|
||
captured = _capture_send(monkeypatch)
|
||
asyncio.run(email_service.send_lead_email(
|
||
customer_profile={}, contact_info={"whatsapp": raw},
|
||
reasoning="", conversation_messages=[],
|
||
cc_info={"whatsapp": state},
|
||
))
|
||
body = _html(captured["msg"])
|
||
assert state["e164"] in body
|
||
assert "未经客户确认" not in body
|
||
assert "（缺国家区号）" not in body
|
||
|
||
|
||
def test_lead_email_cc_fallback_keeps_note(monkeypatch):
|
||
"""fallback 落定(confirmed=false 有 e164): e164 + 未经确认标注。"""
|
||
captured = _capture_send(monkeypatch)
|
||
asyncio.run(email_service.send_lead_email(
|
||
customer_profile={}, contact_info={"whatsapp": "0501234567"},
|
||
reasoning="", conversation_messages=[],
|
||
cc_info={"whatsapp": dict(_CC_FALLBACK)},
|
||
))
|
||
body = _html(captured["msg"])
|
||
assert "+966501234567（区号按 IP 推测，未经客户确认）" in body
|
||
|
||
|
||
def test_lead_email_cc_no_e164_states_raw_with_missing_note(monkeypatch):
|
||
"""无 e164 态: no_ip 与 fallback-无-e164(Task 3 调和防御)→ 裸号原文 +
|
||
（缺国家区号），不崩；主题行同样裸号。"""
|
||
for state in (dict(_CC_NO_IP), dict(_CC_FALLBACK_NO_E164)):
|
||
captured = _capture_send(monkeypatch)
|
||
asyncio.run(email_service.send_lead_email(
|
||
customer_profile={}, contact_info={"whatsapp": "0501234567"},
|
||
reasoning="", conversation_messages=[],
|
||
cc_info={"whatsapp": state},
|
||
))
|
||
body = _html(captured["msg"])
|
||
assert "0501234567（缺国家区号）" in body
|
||
assert "未经客户确认" not in body
|
||
assert "+966" not in body
|
||
assert "0501234567" in str(captured["msg"]["Subject"])
|
||
|
||
|
||
def test_lead_email_merged_phone_shares_whatsapp_state(monkeypatch):
|
||
"""canon 组归并渲染: phone 与 whatsapp 同号(phone 无独立态)共享 whatsapp
|
||
组状态——两行都出补全号,不出现一行 e164 一行裸号的自相矛盾。"""
|
||
captured = _capture_send(monkeypatch)
|
||
asyncio.run(email_service.send_lead_email(
|
||
customer_profile={}, contact_info={"whatsapp": "0501234567",
|
||
"phone": "0501234567"},
|
||
reasoning="", conversation_messages=[],
|
||
cc_info={"whatsapp": dict(_CC_IP_PROPOSED)},
|
||
))
|
||
body = _html(captured["msg"])
|
||
assert body.count("+966501234567（区号按 IP 推测，未经客户确认）") == 2
|
||
|
||
|
||
def test_lead_email_stale_cc_state_ignored(monkeypatch):
|
||
"""canon 不匹配的残留态(换号后旧 _cc)不渲染: 裸号原文、无任何标注——
|
||
旧补全永不跟随新号。"""
|
||
captured = _capture_send(monkeypatch)
|
||
asyncio.run(email_service.send_lead_email(
|
||
customer_profile={}, contact_info={"whatsapp": "0512345678"},
|
||
reasoning="", conversation_messages=[],
|
||
cc_info={"whatsapp": dict(_CC_IP_PROPOSED)}, # canon=0501234567 ≠ 新号
|
||
))
|
||
body = _html(captured["msg"])
|
||
assert "0512345678" in body
|
||
assert "+966501234567" not in body
|
||
assert "（缺国家区号）" not in body
|
||
|
||
|
||
def test_correction_and_returning_render_cc(monkeypatch):
|
||
"""更正邮件正文/回访邮件主题行同样 _cc 感知: 更正正文带标注 e164;
|
||
回访主题纯 e164 无标注。"""
|
||
captured = _capture_send(monkeypatch)
|
||
assert asyncio.run(email_service.send_correction_email(
|
||
{"whatsapp": "0501234567"}, count=2,
|
||
cc_info={"whatsapp": dict(_CC_IP_PROPOSED)},
|
||
)) is True
|
||
assert "[询盘·更正2]" in str(captured["msg"]["Subject"])
|
||
assert "+966501234567（区号按 IP 推测，未经客户确认）" in _html(captured["msg"])
|
||
assert asyncio.run(email_service.send_returning_lead_email(
|
||
{"whatsapp": "0501234567"},
|
||
cc_info={"whatsapp": dict(_CC_IP_PROPOSED)},
|
||
)) is True
|
||
subject = str(captured["msg"]["Subject"])
|
||
assert "+966501234567" in subject
|
||
assert "区号按 IP 推测" not in subject
|