from __future__ import annotations import json import sys import unittest from pathlib import Path from tempfile import TemporaryDirectory from unittest.mock import patch ROOT = Path(__file__).resolve().parents[1] SCRIPTS = ROOT / "scripts" if str(SCRIPTS) not in sys.path: sys.path.insert(0, str(SCRIPTS)) import wechat_draft # noqa: E402 class FakeResponse: def __init__(self, payload: dict, status_code: int = 200) -> None: self._payload = payload self.status_code = status_code self.text = json.dumps(payload, ensure_ascii=False) def json(self) -> dict: return self._payload class WechatDraftTests(unittest.TestCase): def test_push_draft_uploads_article_images_cover_and_adds_draft(self) -> None: with TemporaryDirectory() as temp_dir: root = Path(temp_dir) article_image = root / "article.jpg" cover_image = root / "cover.jpg" article_image.write_bytes(b"fake-image") cover_image.write_bytes(b"fake-cover") article = root / "article.md" article.write_text( "# 测试文章\n\n正文第一段。\n\n![正文图](article.jpg)\n\n正文第二段。", encoding="utf-8", ) credentials = root / "credentials.txt" credentials.write_text("AppID：wx-test\nAppSecret：secret-test", encoding="utf-8") output_dir = root / "dist" calls: list[dict] = [] def fake_request(method: str, url: str, **kwargs): calls.append({"method": method, "url": url, "kwargs": kwargs}) if "cgi-bin/token" in url: return FakeResponse({"access_token": "token-test", "expires_in": 7200}) if "media/uploadimg" in url: return FakeResponse({"url": "https://mmbiz.qpic.cn/article-image.jpg"}) if "material/add_material" in url: return FakeResponse({"media_id": "cover-media-id", "url": "https://mmbiz.qpic.cn/cover.jpg"}) if "draft/add" in url: self.assertNotIn("json", kwargs) raw_body = kwargs.get("data") self.assertIsInstance(raw_body, bytes) self.assertIn("application/json; charset=utf-8", kwargs.get("headers", {}).get("Content-Type", "")) self.assertIn("测试文章".encode("utf-8"), raw_body) self.assertNotIn(b"\\u6d4b\\u8bd5", raw_body) payload = json.loads(raw_body.decode("utf-8")) content = payload["articles"][0]["content"] self.assertIn("https://mmbiz.qpic.cn/article-image.jpg", content) self.assertIn("cover-media-id", json.dumps(payload, ensure_ascii=False)) self.assertLessEqual(len(payload["articles"][0]["digest"].encode("utf-8")), 54) self.assertLessEqual(len(payload["articles"][0]["title"].encode("utf-8")), 64) return FakeResponse({"media_id": "draft-media-id"}) raise AssertionError(f"unexpected url: {url}") args = wechat_draft.argparse.Namespace( input=str(article), theme="wechat-native", title="", author="YuanFlow", digest="", content_source_url="", cover_image=str(cover_image), thumb_media_id="", cover_material_type="image", output_dir=str(output_dir), credentials_file=str(credentials), app_id="", app_secret="", push_draft=True, update_draft=False, draft_media_id="", draft_index=0, json=True, ) with patch.object(wechat_draft.requests, "request", side_effect=fake_request): result = wechat_draft.run(args) self.assertTrue(result["ok"]) self.assertTrue(result["pushed"]) self.assertEqual(result["draft"]["media_id"], "draft-media-id") self.assertEqual(result["article_images"][0]["url"], "https://mmbiz.qpic.cn/article-image.jpg") self.assertTrue((output_dir / "wechat.html").exists()) self.assertTrue((output_dir / "draft_payload.json").exists()) called_urls = "\n".join(call["url"] for call in calls) self.assertIn("cgi-bin/token", called_urls) self.assertIn("media/uploadimg", called_urls) self.assertIn("material/add_material", called_urls) self.assertIn("draft/add", called_urls) def test_update_draft_replaces_article_by_media_id_and_index(self) -> None: with TemporaryDirectory() as temp_dir: root = Path(temp_dir) article_image = root / "article.jpg" article_image.write_bytes(b"fake-image") article = root / "article.md" article.write_text( "# 更新文章\n\n正文第一段。\n\n![正文图](article.jpg)\n\n正文第二段。", encoding="utf-8", ) credentials = root / "credentials.txt" credentials.write_text("AppID：wx-test\nAppSecret：secret-test", encoding="utf-8") output_dir = root / "dist" calls: list[dict] = [] def fake_request(method: str, url: str, **kwargs): calls.append({"method": method, "url": url, "kwargs": kwargs}) if "cgi-bin/token" in url: return FakeResponse({"access_token": "token-test", "expires_in": 7200}) if "media/uploadimg" in url: return FakeResponse({"url": "https://mmbiz.qpic.cn/updated-image.jpg"}) if "draft/update" in url: raw_body = kwargs.get("data") self.assertIsInstance(raw_body, bytes) payload = json.loads(raw_body.decode("utf-8")) self.assertEqual(payload["media_id"], "draft-media-id") self.assertEqual(payload["index"], 0) self.assertIsInstance(payload["articles"], dict) self.assertEqual(payload["articles"]["title"], "更新文章") self.assertIn("https://mmbiz.qpic.cn/updated-image.jpg", payload["articles"]["content"]) self.assertNotIn("thumb_media_id", payload["articles"]) return FakeResponse({"errcode": 0, "errmsg": "ok"}) raise AssertionError(f"unexpected url: {url}") args = wechat_draft.argparse.Namespace( input=str(article), theme="wechat-native", title="", author="YuanFlow", digest="", content_source_url="", cover_image="", thumb_media_id="", cover_material_type="image", output_dir=str(output_dir), credentials_file=str(credentials), app_id="", app_secret="", push_draft=False, update_draft=True, draft_media_id="draft-media-id", draft_index=0, json=True, ) with patch.object(wechat_draft.requests, "request", side_effect=fake_request): result = wechat_draft.run(args) self.assertTrue(result["ok"]) self.assertTrue(result["pushed"]) self.assertEqual(result["operation"], "update") called_urls = "\n".join(call["url"] for call in calls) self.assertIn("cgi-bin/token", called_urls) self.assertIn("media/uploadimg", called_urls) self.assertIn("draft/update", called_urls) self.assertNotIn("draft/add", called_urls) if __name__ == "__main__": unittest.main()