"""RED tests for the twitter_threads pipeline: thread_from_dict, download_main, render_main."""
from __future__ import annotations
import json
from pathlib import Path
from unittest.mock import patch, MagicMock
import pytest
from scripts.twitter_threads.error_types import PostMetrics, PostData, ThreadData, thread_from_dict
from scripts.twitter_threads.fetch_thread import fetch_thread_from_html, thread_to_dict
from scripts.twitter_threads.download_media import download_media, main as download_main
from scripts.twitter_threads.render_markdown import render_markdown, main as render_main
class _FakeResp:
def __init__(self, data: bytes) -> None:
self._data = data
def read(self) -> bytes:
return self._data
def __enter__(self) -> "_FakeResp":
return self
def __exit__(self, *a: object) -> bool:
return False
def _post(post_id: str, text: str = "hi", media_urls: tuple[str, ...] = (), reply_to_id: str | None = None) -> PostData:
metrics = PostMetrics(reply_count=1, repost_count=2, like_count=3, view_count=None)
return PostData(post_id=post_id, author="Tim", handle="NOTimothyLottes", text=text, timestamp="2024-02-01T00:00:00Z", media_urls=media_urls, reply_to_id=reply_to_id, quote_of_id=None, metrics=metrics)
def test_thread_json_roundtrip() -> None:
p1 = _post("1")
p2 = _post("2", media_urls=("https://pbs.twimg.com/media/AAA.jpg",), reply_to_id="1")
t = ThreadData(root_post_id="1", posts=(p1, p2), source_url="https://x.com/x/x/1")
d = thread_to_dict(t)
t2 = thread_from_dict(json.loads(json.dumps(d)))
assert t2 == t
def test_pipeline_html_to_markdown(tmp_path: Path) -> None:
html = """
first
second
"""
path = tmp_path / "thread.html"
path.write_text(html, encoding="utf-8")
res = fetch_thread_from_html(path)
assert res.is_ok
td = res.data
assert td is not None
media_names = {"10": ["10_img1.jpg"]}
out = tmp_path / "thread.md"
rr = render_markdown(td, media_names, out)
assert rr.is_ok
text = out.read_text(encoding="utf-8")
assert "## Post 1" in text
assert "## Post 2" in text
assert "reply to Post 1" in text
assert "" in text
def test_download_cli(tmp_path: Path) -> None:
p1 = _post("5", media_urls=("https://pbs.twimg.com/media/BBB.jpg",))
t = ThreadData(root_post_id="5", posts=(p1,), source_url="https://x.com/x/x/5")
thread_data_json = tmp_path / "thread_data.json"
thread_data_json.write_text(json.dumps(thread_to_dict(t)), encoding="utf-8")
mediadir = tmp_path / "media"
mediadir.mkdir()
(mediadir / "5_1.jpg").write_bytes(b"x")
with patch("scripts.twitter_threads.download_media.subprocess.run", MagicMock(return_value=MagicMock(returncode=0))):
code = download_main(["--input", str(thread_data_json), "--output", str(mediadir)])
assert code == 0
assert (mediadir / "5_1.jpg").exists()
def test_render_cli(tmp_path: Path) -> None:
p1 = _post("7", text="body", media_urls=("https://pbs.twimg.com/media/CCC.jpg",))
t = ThreadData(root_post_id="7", posts=(p1,), source_url="https://x.com/x/x/7")
thread_data_json = tmp_path / "thread_data.json"
thread_data_json.write_text(json.dumps(thread_to_dict(t)), encoding="utf-8")
mediadir = tmp_path / "media"
mediadir.mkdir()
(mediadir / "7_img1.jpg").write_bytes(b"x")
out = tmp_path / "thread.md"
code = render_main(["--input", str(thread_data_json), "--media-dir", str(mediadir), "--output", str(out)])
assert code == 0
assert out.exists()
text = out.read_text(encoding="utf-8")
assert "[Media 1](./media/7_img1.jpg)" in text
assert "body" in text