255 lines
9.6 KiB
Python
255 lines
9.6 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" / "threads-page-generator"
|
|
SCRIPT_PATH = SKILL_ROOT / "scripts" / "render_threads_page.py"
|
|
FAST_SCRIPT_PATH = SKILL_ROOT / "scripts" / "fast_page.py"
|
|
LEAD_PROMPT_PATH = BACKEND_ROOT / "packages" / "harness" / "deerflow" / "agents" / "lead_agent" / "prompt.py"
|
|
|
|
|
|
def _load_renderer():
|
|
spec = importlib.util.spec_from_file_location(
|
|
"threads_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 == "threads-page-generator"
|
|
assert "Threads" 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["threads-page-generator"].category == SkillCategory.PUBLIC
|
|
assert discovered_by_name["threads-page-generator"].enabled is True
|
|
|
|
lead_prompt_source = LEAD_PROMPT_PATH.read_text(encoding="utf-8")
|
|
assert "/mnt/skills/public/threads-page-generator/SKILL.md" in lead_prompt_source
|
|
assert "专用 Threads 路由" in lead_prompt_source
|
|
|
|
|
|
def test_default_json_renders_editable_threads_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" / "threads-template.html").read_text(encoding="utf-8")
|
|
runtime = (SKILL_ROOT / "assets" / "threads-runtime.js").read_text(encoding="utf-8")
|
|
html = renderer.render_document(data, template, runtime)
|
|
|
|
assert summary["modules"] == list(renderer.MODULES)
|
|
assert summary["counts"]["threads"] >= 1
|
|
assert renderer.DATA_MARKER not in html
|
|
assert renderer.RUNTIME_MARKER not in html
|
|
assert 'id="threadsPageData"' in html
|
|
assert 'id="threadList"' in html
|
|
assert 'id="composeOverlay"' in html
|
|
assert 'id="openSettings"' 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 ".thread-media,.image-grid" in html
|
|
assert 'data-thread-id="' in html
|
|
assert "replaceBuiltInVisuals" not in html
|
|
assert "readAsDataURL" in html
|
|
assert '"brand": "Threads"' in html
|
|
assert "<title>Threads · 串联想法</title>" in html
|
|
assert '"threads-page-data-v1:" + pageFingerprint' in html
|
|
assert "const readStorage = (key)" in html
|
|
assert "window.localStorage.getItem" in html
|
|
assert "const cachedData = localStorage.getItem" not in html
|
|
assert '<img src="http' not in html
|
|
assert '<video src="http' not in html
|
|
|
|
data["site"]["pageTitle"] = "Threads · 智能体讨论"
|
|
titled_html = renderer.render_document(data, template, runtime)
|
|
assert "<title>Threads · 智能体讨论</title>" in titled_html
|
|
assert "<title>Threads · 串联想法</title>" not in titled_html
|
|
|
|
data["threads"]["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 / "threads-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_duplicate_and_remote_media() -> None:
|
|
renderer = _load_renderer()
|
|
data = json.loads((SKILL_ROOT / "assets" / "default-data.json").read_text(encoding="utf-8"))
|
|
data.pop("activity")
|
|
data["threads"]["items"][1]["id"] = data["threads"]["items"][0]["id"]
|
|
data["threads"]["items"][0]["media"]["type"] = "remote-image"
|
|
|
|
with pytest.raises(renderer.DataValidationError) as exc_info:
|
|
renderer.validate_document(data)
|
|
|
|
message = str(exc_info.value)
|
|
assert "Missing top-level module: activity" in message
|
|
assert "must be unique" in message
|
|
assert "media.type must be one of" in message
|
|
|
|
|
|
def test_remote_webpage_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["threads"]["items"][0]["media"] = {
|
|
"type": "webpage",
|
|
"pageUrl": "https://example.com/story",
|
|
"title": "演示网页",
|
|
"description": "串文中的网页卡片",
|
|
"embed": True,
|
|
}
|
|
renderer.validate_document(data)
|
|
runtime = (SKILL_ROOT / "assets" / "threads-runtime.js").read_text(encoding="utf-8")
|
|
assert 'media.type === "webpage"' in runtime
|
|
assert 'sandbox="allow-forms allow-popups allow-scripts"' in runtime
|
|
|
|
|
|
def test_high_level_fast_spec_renders_once_without_deep_paths(tmp_path: Path) -> None:
|
|
spec_path = tmp_path / "threads-page-spec.json"
|
|
spec_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"accountName": "茂达是🥚",
|
|
"handle": "@maoda",
|
|
"verified": True,
|
|
"postText": "今天想和大家分享茂达品牌日常，也听听大家的看法。",
|
|
"topic": "品牌日常",
|
|
"replyTopics": ["品牌表达", "产品故事"],
|
|
"displayReplyCount": 3,
|
|
"totalReplyCount": 128,
|
|
"replies": ["这个中文表达很自然，也很适合放在串文首页。"],
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
command = [
|
|
sys.executable,
|
|
str(FAST_SCRIPT_PATH),
|
|
"--spec",
|
|
str(spec_path),
|
|
"--output-dir",
|
|
str(tmp_path),
|
|
]
|
|
|
|
completed = subprocess.run(
|
|
command,
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
)
|
|
|
|
assert completed.returncode == 0, completed.stdout + completed.stderr
|
|
result = json.loads(completed.stdout)
|
|
assert result["status"] == "ok"
|
|
assert result["action"] == "fast-threads-page"
|
|
data_path = Path(result["data"])
|
|
html_path = Path(result["html"])
|
|
assert data_path.name == "threads-page-data.json"
|
|
assert html_path.name == "threads-page.html"
|
|
|
|
data = json.loads(data_path.read_text(encoding="utf-8"))
|
|
assert data["site"]["currentUser"]["name"] == "茂达是🥚"
|
|
assert len(data["threads"]["items"]) == 1
|
|
thread = data["threads"]["items"][0]
|
|
assert thread["author"]["handle"] == "@maoda"
|
|
assert thread["author"]["verified"] is True
|
|
assert thread["metrics"]["replies"] == 128
|
|
assert len(thread["replies"]) == 3
|
|
assert thread["replies"][0]["text"] == "这个中文表达很自然，也很适合放在串文首页。"
|
|
assert thread["media"]["type"] == "autumn-grid"
|
|
assert "今天想和大家分享茂达品牌日常" in html_path.read_text(encoding="utf-8")
|
|
|
|
second = subprocess.run(
|
|
command,
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
)
|
|
assert second.returncode == 0, second.stdout + second.stderr
|
|
second_result = json.loads(second.stdout)
|
|
assert Path(second_result["data"]).name == "threads-page-2-data.json"
|
|
assert Path(second_result["html"]).name == "threads-page-2.html"
|
|
|
|
|
|
def test_high_level_fast_spec_rejects_english_only_post(tmp_path: Path) -> None:
|
|
spec_path = tmp_path / "invalid-spec.json"
|
|
spec_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"accountName": "茂达",
|
|
"postText": "This post should remain Chinese.",
|
|
},
|
|
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 == 2
|
|
result = json.loads(completed.stdout)
|
|
assert result["status"] == "error"
|
|
assert "must contain Simplified Chinese" in result["error"]
|
|
|
|
|
|
def test_threads_fast_workflow_forbids_generic_patch_paths() -> None:
|
|
skill_text = (SKILL_ROOT / "SKILL.md").read_text(encoding="utf-8")
|
|
prompt_text = LEAD_PROMPT_PATH.read_text(encoding="utf-8")
|
|
|
|
assert '"accountName": "茂达是🥚"' in skill_text
|
|
assert "不得写六模块 JSON、`$set` 或深层字段路径" in skill_text
|
|
assert "Threads" in prompt_text
|
|
assert "禁止生成 `$set`、六模块补丁、深层字段路径" in prompt_text
|