286 lines
14 KiB
Python
286 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""x_adapter の pure マッピング関数の単体テスト。
|
|
|
|
ネットワーク・twscrape/twikit/httpx 無しで回る (それらは x_adapter 内で遅延 import)。
|
|
実行: python3 scripts/x-adapter/test_x_adapter.py
|
|
CI/手動どちらでも。失敗で exit 1。
|
|
"""
|
|
import sys
|
|
import types
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
import x_adapter as xa # noqa: E402
|
|
|
|
_fails = []
|
|
|
|
|
|
def check(name, cond):
|
|
print(("PASS" if cond else "FAIL"), name)
|
|
if not cond:
|
|
_fails.append(name)
|
|
|
|
|
|
def ns(**kw):
|
|
return types.SimpleNamespace(**kw)
|
|
|
|
|
|
# ── parse_tweet_ref ──
|
|
check("ref: bare id", xa.parse_tweet_ref("20") == "20")
|
|
check("ref: x.com url", xa.parse_tweet_ref("https://x.com/jack/status/20") == "20")
|
|
check("ref: twitter.com url", xa.parse_tweet_ref("https://twitter.com/jack/statuses/20?s=1") == "20")
|
|
check("ref: article url", xa.parse_tweet_ref("https://x.com/jack/article/2072471529242407210") == "2072471529242407210")
|
|
check("ref: junk -> None", xa.parse_tweet_ref("not a tweet") is None)
|
|
check("ref: empty -> None", xa.parse_tweet_ref("") is None)
|
|
|
|
# ── twitter_time_iso ──
|
|
iso = xa.twitter_time_iso("Wed Oct 10 20:19:24 +0000 2018")
|
|
check("time: iso year", iso.startswith("2018-10-10T20:19:24"))
|
|
check("time: empty", xa.twitter_time_iso("") == "")
|
|
check("time: garbage passthrough", xa.twitter_time_iso("xx") == "xx")
|
|
|
|
# ── twscrape_tweet_to_dict ──
|
|
photo = ns(url="https://pbs.twimg.com/media/AAA.jpg")
|
|
video = ns(thumbnailUrl="https://pbs.twimg.com/poster.jpg",
|
|
variants=[ns(url="https://video.twimg.com/x.mp4", bitrate=832000, contentType="video/mp4")])
|
|
media = ns(photos=[photo], videos=[video], animated=[])
|
|
user = ns(id_str="12", displayname="Jack", username="jack",
|
|
profileImageUrl="https://pbs.twimg.com/pp.jpg")
|
|
tw = ns(id_str="20", id=20, rawContent="hello", user=user,
|
|
likeCount=5, retweetCount=2, replyCount=1, viewCount=99,
|
|
date=ns(isoformat=lambda: "2018-10-10T20:19:24+00:00"),
|
|
retweetedTweet=None, media=media)
|
|
d = xa.twscrape_tweet_to_dict(tw)
|
|
check("tw: id", d["id"] == "20")
|
|
check("tw: text", d["text"] == "hello")
|
|
check("tw: author.screenName", d["author"]["screenName"] == "jack")
|
|
check("tw: author.name", d["author"]["name"] == "Jack")
|
|
check("tw: metrics.views", d["metrics"]["views"] == 99)
|
|
check("tw: createdAtISO", d["createdAtISO"] == "2018-10-10T20:19:24+00:00")
|
|
check("tw: isRetweet False", d["isRetweet"] is False)
|
|
check("tw: photo media", d["media"][0] == {"type": "photo", "url": "https://pbs.twimg.com/media/AAA.jpg"})
|
|
vid = d["media"][1]
|
|
check("tw: video type", vid["type"] == "video")
|
|
check("tw: video poster url", vid["url"] == "https://pbs.twimg.com/poster.jpg")
|
|
check("tw: video variant url", vid["variants"][0]["url"] == "https://video.twimg.com/x.mp4")
|
|
check("tw: video variant bitrate", vid["variants"][0]["bitrate"] == 832000)
|
|
check("tw: video variant contentType", vid["variants"][0]["contentType"] == "video/mp4")
|
|
|
|
tw_rt = ns(id_str="21", rawContent="rt", user=user, retweetedTweet=ns(id="1"), media=None,
|
|
likeCount=0, retweetCount=0, replyCount=0, viewCount=0, date=None)
|
|
check("tw: isRetweet True", xa.twscrape_tweet_to_dict(tw_rt)["isRetweet"] is True)
|
|
check("tw: no media -> empty", xa.twscrape_tweet_to_dict(tw_rt)["media"] == [])
|
|
|
|
# ── parse_graphql_tweet_result: legacy-user shape ──
|
|
result_legacy = {
|
|
"rest_id": "100",
|
|
"core": {"user_results": {"result": {
|
|
"rest_id": "12",
|
|
"legacy": {"name": "Jack", "screen_name": "jack",
|
|
"profile_image_url_https": "https://pbs.twimg.com/pp.jpg"},
|
|
}}},
|
|
"views": {"count": "1234"},
|
|
"legacy": {
|
|
"full_text": "from timeline",
|
|
"created_at": "Wed Oct 10 20:19:24 +0000 2018",
|
|
"favorite_count": 7, "retweet_count": 3, "reply_count": 2,
|
|
"extended_entities": {"media": [
|
|
{"type": "photo", "media_url_https": "https://pbs.twimg.com/media/P.jpg"},
|
|
{"type": "video", "media_url_https": "https://pbs.twimg.com/poster.jpg",
|
|
"video_info": {"variants": [
|
|
{"url": "https://video.twimg.com/v.mp4", "bitrate": 256000, "content_type": "video/mp4"},
|
|
{"url": "https://video.twimg.com/v.m3u8", "content_type": "application/x-mpegURL"},
|
|
]}},
|
|
]},
|
|
},
|
|
}
|
|
g = xa.parse_graphql_tweet_result(result_legacy)
|
|
check("gql: id", g["id"] == "100")
|
|
check("gql: text", g["text"] == "from timeline")
|
|
check("gql: screenName(legacy)", g["author"]["screenName"] == "jack")
|
|
check("gql: name(legacy)", g["author"]["name"] == "Jack")
|
|
check("gql: views parsed", g["metrics"]["views"] == 1234)
|
|
check("gql: likes", g["metrics"]["likes"] == 7)
|
|
check("gql: createdAtISO", g["createdAtISO"].startswith("2018-10-10T20:19:24"))
|
|
check("gql: photo media", g["media"][0]["type"] == "photo")
|
|
check("gql: video variant only-with-url", len(g["media"][1]["variants"]) == 2)
|
|
check("gql: video contentType mapped", g["media"][1]["variants"][0]["contentType"] == "video/mp4")
|
|
|
|
# ── parse_graphql_tweet_result: core-user shape (X migration) ──
|
|
result_core = {
|
|
"rest_id": "101",
|
|
"core": {"user_results": {"result": {
|
|
"rest_id": "12",
|
|
"core": {"name": "Jill", "screen_name": "jill"},
|
|
"avatar": {"image_url": "https://pbs.twimg.com/jill.jpg"},
|
|
}}},
|
|
"views": {"count": "5"},
|
|
"legacy": {"full_text": "core shape", "created_at": "", "favorite_count": 1,
|
|
"retweet_count": 0, "reply_count": 0},
|
|
}
|
|
gc = xa.parse_graphql_tweet_result(result_core)
|
|
check("gql-core: screenName", gc["author"]["screenName"] == "jill")
|
|
check("gql-core: name", gc["author"]["name"] == "Jill")
|
|
check("gql-core: avatar", gc["author"]["profileImageUrl"] == "https://pbs.twimg.com/jill.jpg")
|
|
|
|
# ── parse_graphql_tweet_result: visibility wrapper + no-legacy ──
|
|
wrapped = {"__typename": "TweetWithVisibilityResults", "tweet": result_legacy}
|
|
check("gql: visibility wrapper unwrapped", xa.parse_graphql_tweet_result(wrapped)["id"] == "100")
|
|
check("gql: no legacy -> None", xa.parse_graphql_tweet_result({"rest_id": "9"}) is None)
|
|
check("gql: non-dict -> None", xa.parse_graphql_tweet_result(None) is None)
|
|
|
|
# ── parse_home_timeline ──
|
|
payload = {"data": {"home": {"home_timeline_urt": {"instructions": [
|
|
{"type": "TimelineClearCache"},
|
|
{"type": "TimelineAddEntries", "entries": [
|
|
{"entryId": "tweet-100", "content": {"itemContent": {"tweet_results": {"result": result_legacy}}}},
|
|
{"entryId": "cursor-top-x", "content": {}},
|
|
{"entryId": "tweet-101", "content": {"itemContent": {"tweet_results": {"result": result_core}}}},
|
|
]},
|
|
]}}}}
|
|
rows = xa.parse_home_timeline(payload, limit=50)
|
|
check("home: 2 tweets extracted", len(rows) == 2)
|
|
check("home: order preserved", rows[0]["id"] == "100" and rows[1]["id"] == "101")
|
|
check("home: limit respected", len(xa.parse_home_timeline(payload, limit=1)) == 1)
|
|
check("home: empty payload", xa.parse_home_timeline({}, limit=10) == [])
|
|
|
|
# module-nested tweets (TimelineModule items[] + TimelineAddToModule moduleItems[])
|
|
payload_mod = {"data": {"home": {"home_timeline_urt": {"instructions": [
|
|
{"type": "TimelineAddEntries", "entries": [
|
|
{"entryId": "home-conversation-1", "content": {"items": [
|
|
{"item": {"itemContent": {"tweet_results": {"result": result_legacy}}}},
|
|
]}},
|
|
]},
|
|
{"type": "TimelineAddToModule", "moduleItems": [
|
|
{"item": {"itemContent": {"tweet_results": {"result": result_core}}}},
|
|
]},
|
|
]}}}}
|
|
mrows = xa.parse_home_timeline(payload_mod, limit=50)
|
|
check("home: module items extracted", len(mrows) == 2 and mrows[0]["id"] == "100" and mrows[1]["id"] == "101")
|
|
check("home: who-to-follow user skipped", xa.parse_home_timeline(
|
|
{"data": {"home": {"home_timeline_urt": {"instructions": [
|
|
{"type": "TimelineAddEntries", "entries": [
|
|
{"entryId": "cursor-top", "content": {"cursorType": "Top"}},
|
|
]}]}}}}, limit=10) == [])
|
|
|
|
# ── find_tweet_result_in_detail / extract_article (X 長文記事) ──
|
|
# TweetDetail レスポンスの実構造を模した fixture (2026-07 実データの縮約)
|
|
article_result = {
|
|
"rest_id": "500",
|
|
"legacy": {"full_text": "https://t.co/xxxx", "created_at": "Wed Oct 10 20:19:24 +0000 2018",
|
|
"favorite_count": 1, "retweet_count": 0, "reply_count": 0},
|
|
"article": {"article_results": {"result": {
|
|
"rest_id": "499",
|
|
"title": "記事タイトル",
|
|
"preview_text": "プレビュー文",
|
|
"plain_text": "本文です。" * 10,
|
|
"metadata": {"first_published_at_secs": 1782950691},
|
|
"cover_media": {"media_info": {"original_img_url": "https://pbs.twimg.com/media/COVER.jpg"}},
|
|
}}},
|
|
}
|
|
detail_payload = {"data": {"threaded_conversation_with_injections_v2": {"instructions": [
|
|
{"type": "TimelineAddEntries", "entries": [
|
|
{"entryId": "tweet-1", "content": {"itemContent": {"tweet_results": {"result": {"rest_id": "1", "legacy": {}}}}}},
|
|
{"entryId": "tweet-500", "content": {"itemContent": {"tweet_results": {"result": article_result}}}},
|
|
]},
|
|
]}}}
|
|
found = xa.find_tweet_result_in_detail(detail_payload, "500")
|
|
check("detail: focal tweet found", found is not None and found.get("rest_id") == "500")
|
|
check("detail: missing id -> None", xa.find_tweet_result_in_detail(detail_payload, "999") is None)
|
|
check("detail: empty payload -> None", xa.find_tweet_result_in_detail({}, "500") is None)
|
|
|
|
# visibility wrapper 越しでも見つかる
|
|
wrapped_payload = {"data": {"entries": [{"tweet_results": {"result": {
|
|
"__typename": "TweetWithVisibilityResults", "tweet": article_result}}}]}
|
|
}
|
|
found_w = xa.find_tweet_result_in_detail(wrapped_payload, "500")
|
|
check("detail: visibility wrapper unwrapped", found_w is not None and found_w.get("rest_id") == "500")
|
|
|
|
art = xa.extract_article(article_result)
|
|
check("article: title", art["title"] == "記事タイトル")
|
|
check("article: previewText", art["previewText"] == "プレビュー文")
|
|
check("article: plainText", art["plainText"] == "本文です。" * 10)
|
|
check("article: publishedAtISO", art["publishedAtISO"] == "2026-07-02T00:04:51+00:00")
|
|
check("article: coverImageUrl", art["coverImageUrl"] == "https://pbs.twimg.com/media/COVER.jpg")
|
|
check("article: not truncated", "plainTextTruncated" not in art)
|
|
|
|
# cap: full_text=False では ARTICLE_TEXT_CAP で切る
|
|
long_art = {"article": {"article_results": {"result": {
|
|
"title": "t", "plain_text": "あ" * (xa.ARTICLE_TEXT_CAP + 100)}}}}
|
|
capped = xa.extract_article(long_art)
|
|
check("article: capped length", len(capped["plainText"]) == xa.ARTICLE_TEXT_CAP)
|
|
check("article: truncated flag", capped.get("plainTextTruncated") is True)
|
|
full = xa.extract_article(long_art, full_text=True)
|
|
check("article: full_text lifts cap", len(full["plainText"]) == xa.ARTICLE_TEXT_CAP + 100)
|
|
check("article: full not truncated", "plainTextTruncated" not in full)
|
|
|
|
# 記事なしツイート → None / 壊れた形 → None
|
|
check("article: non-article -> None", xa.extract_article(result_legacy) is None)
|
|
check("article: junk -> None", xa.extract_article({"article": {"article_results": {"result": "?"}}}) is None)
|
|
check("article: non-dict -> None", xa.extract_article(None) is None)
|
|
|
|
# plain_text 無し (fieldToggles 未対応時) でも title/preview は返す
|
|
no_body = {"article": {"article_results": {"result": {"title": "t2", "preview_text": "p2"}}}}
|
|
nb = xa.extract_article(no_body)
|
|
check("article: no body still returns meta", nb["title"] == "t2" and "plainText" not in nb)
|
|
|
|
# ── extract_article_media (記事内の埋め込み画像・動画) ──
|
|
img_entity = {"media_id": "111", "media_info": {
|
|
"__typename": "ApiImage",
|
|
"original_img_url": "https://pbs.twimg.com/media/IMG1.jpg",
|
|
"original_img_width": 1983, "original_img_height": 793}}
|
|
vid_entity = {"media_id": "222", "media_info": {
|
|
"__typename": "ApiVideo",
|
|
"duration_millis": 7658,
|
|
"preview_image": {"original_img_url": "https://pbs.twimg.com/amplify_video_thumb/222/img/P.jpg"},
|
|
"variants": [
|
|
{"bit_rate": 2176000, "content_type": "video/mp4", "url": "https://video.twimg.com/a/720.mp4"},
|
|
{"content_type": "application/x-mpegURL", "url": "https://video.twimg.com/a/pl.m3u8"},
|
|
]}}
|
|
gif_entity = {"media_id": "333", "media_info": {
|
|
"__typename": "ApiGif",
|
|
"preview_image": {"original_img_url": "https://pbs.twimg.com/tweet_video_thumb/G.jpg"},
|
|
"variants": [{"content_type": "video/mp4", "url": "https://video.twimg.com/tweet_video/g.mp4"}]}}
|
|
unknown_entity = {"media_id": "444", "media_info": {"__typename": "ApiAudioSpace"}}
|
|
art_with_media = {"article": {"article_results": {"result": {
|
|
"title": "t", "media_entities": [img_entity, vid_entity, gif_entity, unknown_entity]}}}}
|
|
am = xa.extract_article_media(art_with_media)
|
|
check("amedia: count (unknown skipped)", len(am) == 3)
|
|
check("amedia: photo", am[0] == {"type": "photo", "url": "https://pbs.twimg.com/media/IMG1.jpg"})
|
|
check("amedia: video poster", am[1]["type"] == "video" and am[1]["url"].endswith("P.jpg"))
|
|
check("amedia: video variant mapped", am[1]["variants"][0] ==
|
|
{"url": "https://video.twimg.com/a/720.mp4", "bitrate": 2176000, "contentType": "video/mp4"})
|
|
check("amedia: m3u8 variant kept", am[1]["variants"][1]["contentType"] == "application/x-mpegURL")
|
|
check("amedia: gif -> animated_gif", am[2]["type"] == "animated_gif" and am[2]["variants"][0]["url"].endswith("g.mp4"))
|
|
check("amedia: non-article -> []", xa.extract_article_media(result_legacy) == [])
|
|
check("amedia: no media_entities -> []", xa.extract_article_media(no_body) == [])
|
|
check("amedia: non-dict -> []", xa.extract_article_media(None) == [])
|
|
|
|
# 件数 cap: ARTICLE_MEDIA_CAP を超えたら切る
|
|
many = {"article": {"article_results": {"result": {"media_entities": [img_entity] * (xa.ARTICLE_MEDIA_CAP + 5)}}}}
|
|
check("amedia: capped", len(xa.extract_article_media(many)) == xa.ARTICLE_MEDIA_CAP)
|
|
|
|
# ── emit (optional, needs PyYAML) ──
|
|
try:
|
|
import io
|
|
import yaml # noqa: F401
|
|
buf = io.StringIO()
|
|
old = sys.stdout
|
|
sys.stdout = buf
|
|
try:
|
|
xa.emit([g])
|
|
finally:
|
|
sys.stdout = old
|
|
parsed = yaml.safe_load(buf.getvalue())
|
|
check("emit: ok flag", parsed["ok"] is True)
|
|
check("emit: data roundtrip", parsed["data"][0]["id"] == "100")
|
|
check("emit: unicode preserved", "from timeline" in buf.getvalue())
|
|
except ImportError:
|
|
print("SKIP emit tests (PyYAML not installed)")
|
|
|
|
print()
|
|
if _fails:
|
|
print(f"{len(_fails)} FAILED: {_fails}")
|
|
sys.exit(1)
|
|
print("ALL PASSED")
|