deerflow-code/offline-backend-20260512/backend/tests/test_x_page_generator_skill.py

from __future__ import annotations

import importlib.util
import json
import subprocess
import sys
from pathlib import Path

import pytest

from deerflow.skills.parser import parse_skill_file
from deerflow.skills.storage.local_skill_storage import LocalSkillStorage
from deerflow.skills.types import SkillCategory
from deerflow.skills.validation import _validate_skill_frontmatter

BACKEND_ROOT = Path(__file__).resolve().parents[1]
SKILL_ROOT = BACKEND_ROOT / "skills" / "public" / "x-page-generator"
SCRIPT_PATH = SKILL_ROOT / "scripts" / "render_x_page.py"
FAST_SCRIPT_PATH = SKILL_ROOT / "scripts" / "fast_single_post.py"
LEAD_PROMPT_PATH = BACKEND_ROOT / "packages" / "harness" / "deerflow" / "agents" / "lead_agent" / "prompt.py"


def _load_renderer():
    spec = importlib.util.spec_from_file_location(
        "x_page_renderer",
        SCRIPT_PATH,
    )
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def test_skill_is_discoverable_and_valid() -> None:
    valid, error, _metadata = _validate_skill_frontmatter(SKILL_ROOT)
    assert valid, error
    skill = parse_skill_file(SKILL_ROOT / "SKILL.md", SkillCategory.PUBLIC)
    assert skill is not None
    assert skill.name == "x-page-generator"
    assert "Twitter" in skill.description

    discovered = LocalSkillStorage(host_path=str(SKILL_ROOT.parent.parent)).load_skills(enabled_only=True)
    discovered_by_name = {item.name: item for item in discovered}
    assert discovered_by_name["x-page-generator"].category == SkillCategory.PUBLIC
    assert discovered_by_name["x-page-generator"].enabled is True

    lead_prompt_source = LEAD_PROMPT_PATH.read_text(encoding="utf-8")
    assert "/mnt/skills/public/x-page-generator/SKILL.md" in lead_prompt_source
    assert "专用 X 路由" in lead_prompt_source
    assert "fast_single_post.py --spec" in lead_prompt_source
    assert "推文与回复正文必须使用简体中文" in lead_prompt_source


def test_default_json_renders_editable_x_html(tmp_path: Path) -> None:
    renderer = _load_renderer()
    data = json.loads((SKILL_ROOT / "assets" / "default-data.json").read_text(encoding="utf-8"))
    summary = renderer.validate_document(data)
    template = (SKILL_ROOT / "assets" / "x-template.html").read_text(encoding="utf-8")
    runtime = (SKILL_ROOT / "assets" / "x-runtime.js").read_text(encoding="utf-8")
    html = renderer.render_document(data, template, runtime)

    assert summary["modules"] == list(renderer.MODULES)
    assert summary["counts"]["posts"] >= 1
    assert renderer.DATA_MARKER not in html
    assert renderer.RUNTIME_MARKER not in html
    assert 'id="xPageData"' in html
    assert 'id="openSettings"' in html
    assert 'id="settingsModal"' in html
    assert 'id="moduleTabs"' in html
    assert 'id="applyJson"' in html
    assert "uploadJsonMedia" in html
    assert "installInlineMediaReplace" in html
    assert "dfInlineMediaInput" in html
    assert 'input.accept = "image/*,video/*"' in html
    assert 'button.title = "上传图片或视频并替换"' in html
    assert "active.expected" not in html
    assert "previous.posterUrl" not in html
    assert "decorateBuiltInVisuals" in html
    assert "replaceBuiltInVisuals" not in html
    assert "readAsDataURL" in html
    assert '"brand": "X"' in html
    assert "<title>X · 正在发生</title>" in html
    assert '"x-page-data-v1:" + pageFingerprint' in html
    assert "const readStorage = (key)" in html
    assert "const storedTheme = readStorage(themeKey)" in html
    assert "window.localStorage.getItem" in html
    assert "const storedTheme = localStorage.getItem" not in html

    data["site"]["pageTitle"] = "X · 智能体开发圈"
    titled_html = renderer.render_document(data, template, runtime)
    assert "<title>X · 智能体开发圈</title>" in titled_html
    assert "<title>X · 正在发生</title>" not in titled_html

    data["posts"]["items"][0]["text"] = "</script><script>alert('x')</script>"
    escaped_html = renderer.render_document(data, template, runtime)
    assert "</script><script>alert" not in escaped_html
    assert "<\\/script><script>alert" in escaped_html

    output = tmp_path / "x-page.html"
    renderer._write_new(output, html)
    assert output.is_file()
    with pytest.raises(
        renderer.DataValidationError,
        match="Refusing to overwrite",
    ):
        renderer._write_new(output, html)


def test_validation_rejects_missing_module_and_duplicate_posts() -> None:
    renderer = _load_renderer()
    data = json.loads((SKILL_ROOT / "assets" / "default-data.json").read_text(encoding="utf-8"))
    data.pop("trends")
    data["posts"]["items"][1]["id"] = data["posts"]["items"][0]["id"]

    with pytest.raises(renderer.DataValidationError) as exc_info:
        renderer.validate_document(data)

    message = str(exc_info.value)
    assert "Missing top-level module: trends" in message
    assert "must be unique" in message


def test_remote_video_media_is_validated_and_supported_by_runtime() -> None:
    renderer = _load_renderer()
    data = json.loads((SKILL_ROOT / "assets" / "default-data.json").read_text(encoding="utf-8"))
    data["posts"]["items"][0]["media"] = {
        "type": "video",
        "mediaUrl": "https://cdn.example.com/post.mp4",
        "posterUrl": "https://cdn.example.com/poster.jpg",
        "pageUrl": "https://example.com/post",
    }
    renderer.validate_document(data)
    runtime = (SKILL_ROOT / "assets" / "x-runtime.js").read_text(encoding="utf-8")
    assert 'media.type === "video"' in runtime
    assert 'media.type === "webpage"' in runtime
    assert 'rel="noopener noreferrer"' in runtime


def test_single_post_supports_fifty_preloaded_hot_replies() -> None:
    renderer = _load_renderer()
    data = json.loads((SKILL_ROOT / "assets" / "default-data.json").read_text(encoding="utf-8"))
    post = data["posts"]["items"][0]
    data["posts"]["items"] = [post]
    post["metrics"]["replies"] = 50
    post["repliesExpanded"] = True
    post["disclosure"] = "情景模拟页面 · 帖文与热评为演示内容"
    post["sourceUrl"] = "https://example.com/source-post"
    post["replies"] = [
        {
            "id": f"reply-{index:02d}",
            "author": f"演示用户{index:02d}",
            "handle": f"@demo{index:02d}",
            "initial": str(index % 10),
            "colors": ["#5662f6", "#9f45d8"],
            "verified": index % 11 == 0,
            "time": f"{index + 1} 分钟",
            "text": f"这是第 {index + 1} 条演示热评。",
            "likes": 500 - index,
            "replyCount": index % 7,
        }
        for index in range(50)
    ]

    summary = renderer.validate_document(data)
    html = renderer.render_document(
        data,
        (SKILL_ROOT / "assets" / "x-template.html").read_text(encoding="utf-8"),
        (SKILL_ROOT / "assets" / "x-runtime.js").read_text(encoding="utf-8"),
    )

    assert summary["counts"]["posts"] == 1
    assert summary["counts"]["replies"] == 50
    assert '"id": "reply-49"' in html
    assert "reply-pagination" in html
    assert 'data-sort="hot"' in html
    assert "showAllLabel" in html
    assert "post.disclosure" not in html
    assert "post-source" in html
    assert "查看原帖" in html


def test_reply_limit_rejects_more_than_one_hundred_items() -> None:
    renderer = _load_renderer()
    data = json.loads((SKILL_ROOT / "assets" / "default-data.json").read_text(encoding="utf-8"))
    post = data["posts"]["items"][0]
    post["metrics"]["replies"] = 101
    post["replies"] = [{"author": f"用户{index}", "text": "演示回复"} for index in range(101)]

    with pytest.raises(renderer.DataValidationError, match="at most 100"):
        renderer.validate_document(data)


def test_fast_single_post_builds_json_and_html_in_one_call(tmp_path: Path) -> None:
    command = [
        sys.executable,
        str(FAST_SCRIPT_PATH),
        "--output-dir",
        str(tmp_path),
        "--author",
        "测试账号",
        "--handle",
        "@test_account",
        "--text",
        "这是一条用于验证快速生成路径的推文。",
        "--topic",
        "快速生成",
        "--reply-count",
        "50",
        "--media-type",
        "image",
        "--media-url",
        "https://example.com/image.jpg",
        "--tag",
        "快速生成",
    ]

    completed = subprocess.run(
        command,
        check=False,
        capture_output=True,
        text=True,
        encoding="utf-8",
    )

    assert completed.returncode == 0, completed.stderr
    result = json.loads(completed.stdout)
    assert result["status"] == "ok"
    assert result["action"] == "fast-single-post"
    assert result["counts"]["posts"] == 1
    assert result["counts"]["replies"] == 50

    data_path = Path(result["data"])
    html_path = Path(result["html"])
    assert data_path.exists()
    assert html_path.exists()
    data = json.loads(data_path.read_text(encoding="utf-8"))
    post = data["posts"]["items"][0]
    assert len(post["replies"]) == 50
    assert post["replies"][-1]["id"] == "reply-050"
    assert post["media"]["mediaUrl"] == "https://example.com/image.jpg"
    assert "disclosure" not in post
    html = html_path.read_text(encoding="utf-8")
    assert '"id": "reply-050"' in html
    assert "settings-trigger" in html

    second = subprocess.run(
        command,
        check=False,
        capture_output=True,
        text=True,
        encoding="utf-8",
    )
    assert second.returncode == 0, second.stderr
    second_result = json.loads(second.stdout)
    assert second_result["data"] != result["data"]
    assert second_result["html"] != result["html"]
    assert Path(second_result["data"]).exists()
    assert Path(second_result["html"]).exists()


def test_fast_single_post_spec_separates_total_and_displayed_replies(
    tmp_path: Path,
) -> None:
    spec_path = tmp_path / "x-spec.json"
    spec_path.write_text(
        json.dumps(
            {
                "author": "Donald J. Trump",
                "handle": "@realDonaldTrump",
                "verified": True,
                "text": "刚刚结束一次重要会晤，双方将继续推进对话与合作。",
                "topic": "中美元首会晤",
                "tags": ["中美关系"],
                "displayReplyCount": 20,
                "totalReplyCount": 10000,
                "replyTopics": ["中美经济影响", "两岸关系"],
                "replies": ["关税调整将直接影响企业成本和供应链安排。"],
                "likes": 214000,
                "reposts": 38600,
                "views": 2380000,
            },
            ensure_ascii=False,
        ),
        encoding="utf-8",
    )

    completed = subprocess.run(
        [
            sys.executable,
            str(FAST_SCRIPT_PATH),
            "--spec",
            str(spec_path),
            "--output-dir",
            str(tmp_path),
        ],
        check=False,
        capture_output=True,
        text=True,
        encoding="utf-8",
    )

    assert completed.returncode == 0, completed.stderr
    result = json.loads(completed.stdout)
    data = json.loads(Path(result["data"]).read_text(encoding="utf-8"))
    post = data["posts"]["items"][0]
    assert post["metrics"]["replies"] == 10000
    assert len(post["replies"]) == 20
    assert post["replies"][0]["text"] == "关税调整将直接影响企业成本和供应链安排。"
    assert any(topic in post["replies"][1]["text"] for topic in ("中美经济影响", "两岸关系"))
    assert post["media"]["type"] == "skyline"
    assert result["counts"]["replies"] == 20


def test_fast_single_post_rejects_english_only_tweet(tmp_path: Path) -> None:
    spec_path = tmp_path / "x-english-spec.json"
    spec_path.write_text(
        json.dumps(
            {
                "author": "Example",
                "handle": "@example",
                "text": "This post is written only in English.",
            }
        ),
        encoding="utf-8",
    )

    completed = subprocess.run(
        [
            sys.executable,
            str(FAST_SCRIPT_PATH),
            "--spec",
            str(spec_path),
            "--output-dir",
            str(tmp_path),
        ],
        check=False,
        capture_output=True,
        text=True,
        encoding="utf-8",
    )

    assert completed.returncode == 2
    result = json.loads(completed.stdout)
    assert "Simplified Chinese" in result["error"]

    spec_path.write_text(
        json.dumps(
            {
                "author": "Example",
                "handle": "@example",
                "text": "这是一条中文推文。",
                "replies": ["This reply is written only in English."],
            },
            ensure_ascii=False,
        ),
        encoding="utf-8",
    )
    reply_completed = subprocess.run(
        [
            sys.executable,
            str(FAST_SCRIPT_PATH),
            "--spec",
            str(spec_path),
            "--output-dir",
            str(tmp_path),
        ],
        check=False,
        capture_output=True,
        text=True,
        encoding="utf-8",
    )
    assert reply_completed.returncode == 2
    assert "Simplified Chinese" in json.loads(reply_completed.stdout)["error"]