maestro/scripts/x-adapter/test_x_adapter.py

#!/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: 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) == [])

# ── 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")