from services import telegram_templates
from services.telegram_templates import _sanitize_telegram_html, price_decision, triaged_alert
def test_telegram_html_sanitizer_converts_br_tags_to_newlines():
msg = _sanitize_telegram_html("第一行
第二行
第三行
第四行")
assert "
第二行", parse_mode=None) == "第一行
第二行"
def test_send_telegram_with_result_sanitizes_html_payload(monkeypatch):
sent_payloads = []
class Response:
ok = True
status_code = 200
text = "ok"
def fake_post(url, json=None, timeout=None):
sent_payloads.append({"url": url, "json": json, "timeout": timeout})
return Response()
monkeypatch.setattr(telegram_templates, "_get_bot_token", lambda: "telegram-token")
monkeypatch.setattr("requests.post", fake_post)
result = telegram_templates.send_telegram_with_result(
"第一行
第二行
第三行",
chat_ids=[101, 202],
parse_mode="HTML",
)
assert result["ok"] is True
assert result["sent"] == 2
assert [item["json"]["chat_id"] for item in sent_payloads] == [101, 202]
assert all(item["json"]["text"] == "第一行\n第二行\n第三行" for item in sent_payloads)
assert all(item["json"]["parse_mode"] == "HTML" for item in sent_payloads)
def test_price_decision_accepts_report_url_and_escapes_dynamic_fields():
message, keyboard = price_decision(
product_name="精華 ",
insight_id=42,
report_url="https://mo.wooo.work/report?a=1&b=",
)
assert "