89 lines
3.1 KiB
Python
89 lines
3.1 KiB
Python
from pathlib import Path
|
|
|
|
|
|
class FakeResponse:
|
|
def __init__(self, ok=True, payload=None, text=""):
|
|
self.ok = ok
|
|
self._payload = payload if payload is not None else {"ok": ok}
|
|
self.text = text
|
|
|
|
def json(self):
|
|
return self._payload
|
|
|
|
|
|
def test_send_message_retries_plain_text_when_markdown_parse_fails(monkeypatch):
|
|
from services.openclaw_bot import telegram_api
|
|
|
|
calls = []
|
|
|
|
def fake_post(url, json=None, **_kwargs):
|
|
calls.append((url, json))
|
|
if len(calls) == 1:
|
|
return FakeResponse(False, {"ok": False, "description": "can't parse entities"})
|
|
return FakeResponse(True, {"ok": True, "result": {"message_id": 1}})
|
|
|
|
monkeypatch.setattr(telegram_api, "BOT_API_URL", "https://telegram.test/botTOKEN")
|
|
monkeypatch.setattr(telegram_api.requests, "post", fake_post)
|
|
|
|
result = telegram_api.send_message(123, "*bold* [link](https://example.com)")
|
|
|
|
assert result["ok"] is True
|
|
assert len(calls) == 2
|
|
assert calls[0][1]["parse_mode"] == "Markdown"
|
|
assert "parse_mode" not in calls[1][1]
|
|
assert calls[1][1]["text"] == "bold link"
|
|
|
|
|
|
def test_send_message_truncates_overlong_text_after_failed_send(monkeypatch):
|
|
from services.openclaw_bot import telegram_api
|
|
|
|
calls = []
|
|
|
|
def fake_post(url, json=None, **_kwargs):
|
|
calls.append((url, json))
|
|
if len(calls) == 1:
|
|
return FakeResponse(False, {"ok": False, "description": "message is too long"})
|
|
return FakeResponse(True, {"ok": True})
|
|
|
|
monkeypatch.setattr(telegram_api, "BOT_API_URL", "https://telegram.test/botTOKEN")
|
|
monkeypatch.setattr(telegram_api.requests, "post", fake_post)
|
|
|
|
result = telegram_api.send_message(123, "*" + ("x" * 4100) + "*")
|
|
|
|
assert result["ok"] is True
|
|
assert len(calls) == 2
|
|
assert len(calls[1][1]["text"]) < 4000
|
|
assert calls[1][1]["text"].endswith("...(訊息過長已截斷)")
|
|
|
|
|
|
def test_send_document_posts_file_payload(monkeypatch, tmp_path):
|
|
from services.openclaw_bot import telegram_api
|
|
|
|
uploaded = {}
|
|
document = tmp_path / "report.txt"
|
|
document.write_text("hello", encoding="utf-8")
|
|
|
|
def fake_post(url, data=None, files=None, **_kwargs):
|
|
uploaded["url"] = url
|
|
uploaded["data"] = data
|
|
uploaded["file_name"] = Path(files["document"].name).name
|
|
return FakeResponse(True, {"ok": True, "result": {"document": {}}})
|
|
|
|
monkeypatch.setattr(telegram_api, "BOT_API_URL", "https://telegram.test/botTOKEN")
|
|
monkeypatch.setattr(telegram_api.requests, "post", fake_post)
|
|
|
|
result = telegram_api.send_document(123, document, caption="caption", reply_to=9)
|
|
|
|
assert result["ok"] is True
|
|
assert uploaded["url"] == "https://telegram.test/botTOKEN/sendDocument"
|
|
assert uploaded["data"] == {"chat_id": 123, "caption": "caption", "reply_to_message_id": 9}
|
|
assert uploaded["file_name"] == "report.txt"
|
|
|
|
|
|
def test_openclaw_routes_keep_tg_helper_import_for_webhook_management():
|
|
route_source = Path("routes/openclaw_bot_routes.py").read_text(encoding="utf-8")
|
|
|
|
assert "_tg('setMyCommands'" in route_source
|
|
assert "_tg('setWebhook'" in route_source
|
|
assert " _tg,\n" in route_source
|