Files
deerflow-code/offline-backend-20260512/backend/tests/test_facebook_page_generator_skill.py

410 lines
14 KiB
Python

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" / "facebook-page-generator"
SCRIPT_PATH = SKILL_ROOT / "scripts" / "render_facebook_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("facebook_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 == "facebook-page-generator"
assert "Facebook" 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["facebook-page-generator"].category == SkillCategory.PUBLIC
assert discovered_by_name["facebook-page-generator"].enabled is True
lead_prompt_source = LEAD_PROMPT_PATH.read_text(encoding="utf-8")
assert "/mnt/skills/public/facebook-page-generator/SKILL.md" in lead_prompt_source
assert "专用 Facebook 路由" 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_facebook_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" / "facebook-template.html").read_text(encoding="utf-8")
html = renderer.render_document(data, template)
assert summary["modules"] == list(renderer.MODULES)
assert summary["counts"]["posts"] >= 1
assert renderer.DATA_MARKER not in html
assert 'id="circleData"' in html
assert 'id="openSettings"' in html
assert 'id="settingsModal"' 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 "文件不能超过 20MB" not in html
assert '"brand": "Facebook"' in html
assert "<title>Facebook · 社交动态</title>" in html
data["site"]["pageTitle"] = "Facebook · 智能体开发圈"
titled_html = renderer.render_document(data, template)
assert "<title>Facebook · 智能体开发圈</title>" in titled_html
assert "<title>Facebook · 社交动态</title>" not in titled_html
assert "facebook-page-data-v2:${dataFingerprint}" in html
assert "const readStorage = (key)" in html
assert "const savedTheme = readStorage('circle-theme')" in html
assert "writeStorage('circle-theme', next)" in html
assert "const savedTheme = localStorage.getItem('circle-theme')" not in html
assert "const storageKey = 'circle-page-data-v1'" not in html
data["posts"]["items"][0]["text"] = "</script><script>alert('x')</script>"
escaped_html = renderer.render_document(data, template)
assert "</script><script>alert" not in escaped_html
assert "<\\/script><script>alert" in escaped_html
output = tmp_path / "facebook.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("contacts")
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: contacts" in message
assert "must be unique" in message
@pytest.mark.parametrize(
"media",
[
{
"type": "image",
"mediaUrl": "https://cdn.example.com/photo.jpg",
"alt": "演示图片",
"pageUrl": "https://example.com/article",
},
{
"type": "video",
"mediaUrl": "https://cdn.example.com/video.mp4",
"posterUrl": "https://cdn.example.com/poster.jpg",
"alt": "演示视频",
},
{
"type": "webpage",
"pageUrl": "https://example.com/article",
"title": "演示网页",
"description": "可在新窗口打开,也可选择内嵌预览。",
"embed": True,
},
],
)
def test_remote_media_contract_renders_safe_editable_markup(media: dict[str, object]) -> None:
renderer = _load_renderer()
data = json.loads((SKILL_ROOT / "assets" / "default-data.json").read_text(encoding="utf-8"))
data["posts"]["items"][0]["media"] = media
renderer.validate_document(data)
html = renderer.render_document(
data,
(SKILL_ROOT / "assets" / "facebook-template.html").read_text(encoding="utf-8"),
)
assert "media.type === 'image'" in html
assert "media.type === 'video'" in html
assert "media.type === 'webpage'" in html
assert 'rel="noopener noreferrer"' in html
assert "safeRemoteUrl" in html
def test_remote_media_contract_rejects_unsafe_urls() -> None:
renderer = _load_renderer()
data = json.loads((SKILL_ROOT / "assets" / "default-data.json").read_text(encoding="utf-8"))
data["posts"]["items"][0]["media"] = {
"type": "image",
"mediaUrl": "javascript:alert(1)",
"alt": "不安全图片",
}
with pytest.raises(renderer.DataValidationError, match=r"mediaUrl must use an http\(s\) URL"):
renderer.validate_document(data)
def test_uploaded_data_url_is_accepted_by_media_contract() -> None:
renderer = _load_renderer()
data = json.loads((SKILL_ROOT / "assets" / "default-data.json").read_text(encoding="utf-8"))
data["posts"]["items"][0]["media"] = {
"type": "image",
"mediaUrl": "data:image/png;base64,iVBORw0KGgo=",
"alt": "用户上传的图片",
}
renderer.validate_document(data)
def test_single_post_supports_fifty_preloaded_hot_comments() -> 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"]["comments"] = 50
post["commentsExpanded"] = True
post["disclosure"] = "情景模拟页面 · 帖文与热评为演示内容"
post["sourceUrl"] = "https://example.com/source-post"
post["comments"] = [
{
"id": f"comment-{index:02d}",
"author": f"演示用户{index:02d}",
"handle": f"@demo{index:02d}",
"initial": str(index % 10),
"colors": ["#4d6aff", "#8f4bd2"],
"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" / "facebook-template.html").read_text(encoding="utf-8"),
)
assert summary["counts"]["posts"] == 1
assert summary["counts"]["comments"] == 50
assert '"id": "comment-49"' in html
assert "comment-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_comment_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"]["comments"] = 101
post["comments"] = [{"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",
"测试账号",
"--text",
"这是一条用于验证快速生成路径的动态。",
"--topic",
"快速生成",
"--comment-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"]["comments"] == 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["comments"]) == 50
assert post["comments"][-1]["id"] == "comment-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": "comment-050"' in html
assert 'id="openSettings"' 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_comments(
tmp_path: Path,
) -> None:
spec_path = tmp_path / "facebook-spec.json"
spec_path.write_text(
json.dumps(
{
"author": "示例账号",
"verified": True,
"text": "今天分享一次重要会晤,双方将继续推进对话与合作。",
"topic": "中美元首会晤",
"tags": ["中美关系"],
"displayCommentCount": 20,
"totalCommentCount": 10000,
"commentTopics": ["中美经济影响", "两岸关系"],
"comments": ["关税调整将直接影响企业成本和供应链安排。"],
"likes": 214000,
"shares": 38600,
},
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"]["comments"] == 10000
assert len(post["comments"]) == 20
assert post["comments"][0]["text"] == "关税调整将直接影响企业成本和供应链安排。"
assert any(topic in post["comments"][1]["text"] for topic in ("中美经济影响", "两岸关系"))
assert post["media"]["type"] == "river"
assert result["counts"]["comments"] == 20
def test_fast_single_post_rejects_english_only_post(tmp_path: Path) -> None:
spec_path = tmp_path / "facebook-english-spec.json"
spec_path.write_text(
json.dumps(
{
"author": "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",
"text": "这是一条中文动态。",
"comments": ["This comment is written only in English."],
},
ensure_ascii=False,
),
encoding="utf-8",
)
comment_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 comment_completed.returncode == 2
assert "Simplified Chinese" in json.loads(comment_completed.stdout)["error"]