from pathlib import Path import importlib 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 def test_openclaw_telegram_api_falls_back_to_shared_bot_token(monkeypatch): monkeypatch.delenv("OPENCLAW_BOT_TOKEN", raising=False) monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "shared-token") from services.openclaw_bot import telegram_api reloaded = importlib.reload(telegram_api) assert reloaded.BOT_TOKEN == "shared-token" assert reloaded.BOT_API_URL == "https://api.telegram.org/botshared-token"