Private
Public Access
ed
0d51c3541d
Path objects with trailing separators (e.g. './media/') intermittently
failed to glob on Windows after download_media.py had just finished
writing the directory. Normalize via rstrip('/\\\\') before Path()
re-wrapping so the CLI is forgiving regardless of OS path quirks.
Repro: gallery-dl ran + immediate render with trailing slash on the
just-created media/ dir -> glob returned nothing -> markdown emitted
no ![Media N] lines. Removing the trailing slash fixed it; this makes
both work.
130 lines
5.3 KiB
Python
130 lines
5.3 KiB
Python
"""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 = """<html><body>
|
|
<article data-post-id="10" data-author="Tim" data-handle="NOTimothyLottes">
|
|
<time datetime="2024-02-01T00:00:00Z">d1</time>
|
|
<div data-testid="tweetText">first</div>
|
|
<img src="https://pbs.twimg.com/media/ZZZ.jpg">
|
|
</article>
|
|
<article data-post-id="11" data-author="Tim" data-handle="NOTimothyLottes" data-reply-to="10">
|
|
<time datetime="2024-02-01T00:00:00Z">d2</time>
|
|
<div data-testid="tweetText">second</div>
|
|
</article>
|
|
</body></html>"""
|
|
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
|
|
|
|
def test_render_cli_trailing_slash_on_media_dir(tmp_path: Path) -> None:
|
|
p1 = _post("8", text="body", media_urls=("https://pbs.twimg.com/media/DDD.jpg",))
|
|
t = ThreadData(root_post_id="8", posts=(p1,), source_url="https://x.com/x/x/8")
|
|
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 / "8_img1.jpg").write_bytes(b"x")
|
|
out = tmp_path / "thread.md"
|
|
code = render_main(["--input", str(thread_data_json), "--media-dir", "./" + str(mediadir).replace("\\", "/") + "/", "--output", str(out)])
|
|
assert code == 0
|
|
text = out.read_text(encoding="utf-8")
|
|
assert "[Media 1](./media/8_img1.jpg)" in text
|
|
assert "body" in text
|
|
|
|
def test_download_cli_trailing_slash_on_output(tmp_path: Path) -> None:
|
|
p1 = _post("9", media_urls=("https://pbs.twimg.com/media/EEE.jpg",))
|
|
t = ThreadData(root_post_id="9", posts=(p1,), source_url="https://x.com/x/x/9")
|
|
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 / "9_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).replace("\\", "/") + "/"])
|
|
assert code == 0
|
|
assert (mediadir / "9_1.jpg").exists()
|