admin/claw
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

sgclaw: stop zhihu publish flow before editor on creator page

Browse Source
This commit is contained in:
zyl
2026-03-30 13:35:50 +08:00
parent bf09de6700
commit cd94904329
3 changed files with 1334 additions and 156 deletions
Download Patch File Download Diff File Expand all files Collapse all files

823
tests/compat_runtime_test.rs
View File

@@ -1919,6 +1919,62 @@ fn browser_attached_excel_request_uses_execution_contract_not_skill_source_stuff
assert!(!first_request.contains("Preloaded skill context:"));
}
#[test]
fn browser_attached_publish_request_injects_confirmation_contract() {
let _guard = env_lock().lock().unwrap_or_else(|err| err.into_inner());
let response = json!({
"choices": [{
"message": {
"content": "请先确认是否发布"
}
}]
});
let (base_url, requests, server_handle) = start_fake_deepseek_server(vec![response]);
let workspace_root = temp_workspace_root();
let mut settings = SgClawSettings::from_legacy_deepseek_fields(
"deepseek-test-key".to_string(),
base_url,
"deepseek-chat".to_string(),
Some(real_skill_lib_root()),
)
.unwrap();
settings.runtime_profile = RuntimeProfile::BrowserAttached;
let transport = Arc::new(MockTransport::new(vec![]));
let browser_tool = BrowserPipeTool::new(
transport.clone(),
zhihu_test_policy(),
vec![1, 2, 3, 4, 5, 6, 7, 8],
)
.with_response_timeout(Duration::from_secs(1));
execute_task_with_sgclaw_settings(
transport.as_ref(),
browser_tool,
"请直接发表这篇知乎文章，标题是测试标题，正文是第一段内容",
&CompatTaskContext {
conversation_id: None,
messages: vec![],
page_url: Some("https://www.zhihu.com/creator".to_string()),
page_title: Some("知乎创作中心".to_string()),
},
&workspace_root,
&settings,
)
.unwrap();
server_handle.join().unwrap();
let request_bodies = requests.lock().unwrap().clone();
let first_request = request_bodies[0].to_string();
assert!(first_request.contains("Zhihu article publish contract"));
assert!(first_request.contains("must not click publish without explicit human confirmation"));
assert!(first_request.contains("ask for confirmation concisely"));
assert!(first_request.contains("stop after the confirmation request"));
}
#[test]
fn handle_browser_message_executes_real_zhihu_hotlist_skill_flow() {
let _guard = env_lock().lock().unwrap_or_else(|err| err.into_inner());
@@ -2491,6 +2547,7 @@ fn browser_submit_path_prefers_zeroclaw_process_message_orchestrator() {
.unwrap();
let sent = transport.sent_messages();
dbg!(&sent);
assert!(sent.iter().any(|message| {
matches!(
@@ -2508,6 +2565,592 @@ fn browser_submit_path_prefers_zeroclaw_process_message_orchestrator() {
}));
}
#[test]
fn browser_submit_path_prefers_zeroclaw_process_message_orchestrator_for_zhihu_publish() {
let _guard = env_lock().lock().unwrap_or_else(|err| err.into_inner());
let workspace_root = temp_workspace_root();
let config_path = write_deepseek_config_with_skills_dir(
&workspace_root,
"deepseek-test-key",
"http://127.0.0.1:9",
"deepseek-chat",
Some(real_skill_lib_root().to_str().unwrap()),
);
let runtime_context = AgentRuntimeContext::new(Some(config_path), workspace_root.clone());
let transport = Arc::new(MockTransport::new(vec![]));
let browser_tool = BrowserPipeTool::new(
transport.clone(),
zhihu_test_policy(),
vec![1, 2, 3, 4, 5, 6, 7, 8],
)
.with_response_timeout(Duration::from_secs(1));
handle_browser_message_with_context(
transport.as_ref(),
&browser_tool,
&runtime_context,
BrowserMessage::SubmitTask {
instruction: "请直接发表这篇知乎文章，标题是测试标题，正文是第一段内容".to_string(),
conversation_id: String::new(),
messages: vec![],
page_url: "https://www.zhihu.com/".to_string(),
page_title: "知乎".to_string(),
},
)
.unwrap();
let sent = transport.sent_messages();
assert!(sent.iter().any(|message| {
matches!(
message,
AgentMessage::LogEntry { level, message }
if level == "mode" && message == "zeroclaw_process_message_primary"
)
}));
assert!(!sent.iter().any(|message| {
matches!(
message,
AgentMessage::LogEntry { level, message }
if level == "mode" && (message == "compat_llm_primary" || message == "compat_skill_runner_primary")
)
}));
}
#[test]
fn zhihu_publish_task_matches_primary_orchestration_gate() {
assert!(sgclaw::compat::orchestration::should_use_primary_orchestration(
"请直接发表这篇知乎文章，标题是测试标题，正文是第一段内容",
Some("https://www.zhihu.com/"),
Some("知乎"),
));
}
#[test]
fn zhihu_article_entry_task_matches_primary_orchestration_gate() {
assert!(sgclaw::compat::orchestration::should_use_primary_orchestration(
"打开知乎发文章页面",
Some("https://www.zhihu.com/"),
Some("知乎"),
));
}
#[test]
fn zhihu_publish_without_article_inputs_returns_missing_fields_prompt() {
let _guard = env_lock().lock().unwrap_or_else(|err| err.into_inner());
let workspace_root = temp_workspace_root();
let config_path = write_deepseek_config_with_skills_dir(
&workspace_root,
"deepseek-test-key",
"http://127.0.0.1:9",
"deepseek-chat",
Some(real_skill_lib_root().to_str().unwrap()),
);
let runtime_context = AgentRuntimeContext::new(Some(config_path), workspace_root.clone());
let transport = Arc::new(MockTransport::new(vec![]));
let browser_tool = BrowserPipeTool::new(
transport.clone(),
zhihu_test_policy(),
vec![1, 2, 3, 4, 5, 6, 7, 8],
)
.with_response_timeout(Duration::from_secs(1));
handle_browser_message_with_context(
transport.as_ref(),
&browser_tool,
&runtime_context,
BrowserMessage::SubmitTask {
instruction: "请发表知乎文章".to_string(),
conversation_id: String::new(),
messages: vec![],
page_url: "https://www.zhihu.com/".to_string(),
page_title: "知乎".to_string(),
},
)
.unwrap();
let sent = transport.sent_messages();
assert!(sent.iter().any(|message| {
matches!(
message,
AgentMessage::TaskComplete { success, summary }
if *success &&
summary.contains("标题") &&
summary.contains("正文")
)
}));
assert!(!sent.iter().any(|message| {
matches!(message, AgentMessage::Command { .. })
}));
}
#[test]
fn zhihu_publish_accepts_literal_backslash_n_between_title_and_body() {
let _guard = env_lock().lock().unwrap_or_else(|err| err.into_inner());
let workspace_root = temp_workspace_root();
let config_path = write_deepseek_config_with_skills_dir(
&workspace_root,
"deepseek-test-key",
"http://127.0.0.1:9",
"deepseek-chat",
Some(real_skill_lib_root().to_str().unwrap()),
);
let runtime_context = AgentRuntimeContext::new(Some(config_path), workspace_root.clone());
let transport = Arc::new(MockTransport::new(vec![
success_browser_response(1, json!({ "navigated": true })),
success_browser_response(
2,
json!({
"text": {
"status": "creator_entry_clicked",
"current_url": "https://www.zhihu.com/creator",
"next_url": "https://zhuanlan.zhihu.com/write"
}
}),
),
success_browser_response(3, json!({ "navigated": true })),
success_browser_response(
4,
json!({
"text": {
"status": "editor_ready",
"current_url": "https://zhuanlan.zhihu.com/write"
}
}),
),
success_browser_response(
5,
json!({
"text": {
"status": "draft_ready",
"current_url": "https://zhuanlan.zhihu.com/write",
"title": "ai时代，普通人如何自救"
}
}),
),
]));
let browser_tool = BrowserPipeTool::new(
transport.clone(),
zhihu_test_policy(),
vec![1, 2, 3, 4, 5, 6, 7, 8],
)
.with_response_timeout(Duration::from_secs(1));
handle_browser_message_with_context(
transport.as_ref(),
&browser_tool,
&runtime_context,
BrowserMessage::SubmitTask {
instruction: "标题：ai时代，普通人如何自救 \\n正文：第一段内容。 第二段内容。".to_string(),
conversation_id: String::new(),
messages: vec![],
page_url: "https://www.zhihu.com/creator".to_string(),
page_title: "知乎创作中心".to_string(),
},
)
.unwrap();
let sent = transport.sent_messages();
assert!(sent.iter().any(|message| {
matches!(
message,
AgentMessage::TaskComplete { success, summary }
if *success && summary == "已进入知乎文章编辑器并写入草稿《ai时代，普通人如何自救》"
)
}));
}
#[test]
fn zhihu_article_entry_opens_editor_without_generic_selector_probing() {
let _guard = env_lock().lock().unwrap_or_else(|err| err.into_inner());
let workspace_root = temp_workspace_root();
let config_path = write_deepseek_config_with_skills_dir(
&workspace_root,
"deepseek-test-key",
"http://127.0.0.1:9",
"deepseek-chat",
Some(real_skill_lib_root().to_str().unwrap()),
);
let runtime_context = AgentRuntimeContext::new(Some(config_path), workspace_root.clone());
let transport = Arc::new(MockTransport::new(vec![
success_browser_response(1, json!({ "navigated": true })),
success_browser_response(
2,
json!({
"text": {
"status": "creator_entry_clicked",
"current_url": "https://www.zhihu.com/creator",
"next_url": "https://zhuanlan.zhihu.com/write"
}
}),
),
success_browser_response(3, json!({ "navigated": true })),
success_browser_response(
4,
json!({
"text": {
"status": "editor_ready",
"current_url": "https://zhuanlan.zhihu.com/write"
}
}),
),
]));
let browser_tool = BrowserPipeTool::new(
transport.clone(),
zhihu_test_policy(),
vec![1, 2, 3, 4, 5, 6, 7, 8],
)
.with_response_timeout(Duration::from_secs(1));
handle_browser_message_with_context(
transport.as_ref(),
&browser_tool,
&runtime_context,
BrowserMessage::SubmitTask {
instruction: "打开知乎发文章页面".to_string(),
conversation_id: String::new(),
messages: vec![],
page_url: "https://www.zhihu.com/".to_string(),
page_title: "知乎".to_string(),
},
)
.unwrap();
let sent = transport.sent_messages();
assert!(sent.iter().any(|message| {
matches!(
message,
AgentMessage::TaskComplete { success, summary }
if *success && summary.contains("编辑器")
)
}));
assert!(sent.iter().any(|message| {
matches!(
message,
AgentMessage::LogEntry { level, message }
if level == "mode" && message == "zeroclaw_process_message_primary"
)
}));
assert!(!sent.iter().any(|message| {
matches!(
message,
AgentMessage::LogEntry { level, message }
if level == "info" &&
(message.starts_with("getText ") || message.starts_with("click "))
)
}));
assert!(sent.iter().any(|message| {
matches!(
message,
AgentMessage::Command { action, params, .. }
if action == &Action::Navigate &&
params["url"].as_str() == Some("https://zhuanlan.zhihu.com/write")
)
}));
}
#[test]
fn zhihu_article_entry_reports_editor_unavailable_without_protocol_error() {
let _guard = env_lock().lock().unwrap_or_else(|err| err.into_inner());
let workspace_root = temp_workspace_root();
let config_path = write_deepseek_config_with_skills_dir(
&workspace_root,
"deepseek-test-key",
"http://127.0.0.1:9",
"deepseek-chat",
Some(real_skill_lib_root().to_str().unwrap()),
);
let runtime_context = AgentRuntimeContext::new(Some(config_path), workspace_root.clone());
let transport = Arc::new(MockTransport::new(vec![
success_browser_response(1, json!({ "navigated": true })),
success_browser_response(
2,
json!({
"text": {
"status": "creator_entry_clicked",
"current_url": "https://www.zhihu.com/creator",
"next_url": "https://zhuanlan.zhihu.com/write"
}
}),
),
success_browser_response(3, json!({ "navigated": true })),
success_browser_response(
4,
json!({
"text": {
"status": "editor_unavailable",
"current_url": "https://zhuanlan.zhihu.com/write"
}
}),
),
]));
let browser_tool = BrowserPipeTool::new(
transport.clone(),
zhihu_test_policy(),
vec![1, 2, 3, 4, 5, 6, 7, 8],
)
.with_response_timeout(Duration::from_secs(1));
handle_browser_message_with_context(
transport.as_ref(),
&browser_tool,
&runtime_context,
BrowserMessage::SubmitTask {
instruction: "打开知乎发文章页面".to_string(),
conversation_id: String::new(),
messages: vec![],
page_url: "https://www.zhihu.com/".to_string(),
page_title: "知乎".to_string(),
},
)
.unwrap();
let sent = transport.sent_messages();
assert!(sent.iter().any(|message| {
matches!(
message,
AgentMessage::TaskComplete { success, summary }
if *success &&
summary.contains("未检测到文章编辑器")
)
}));
assert!(sent.iter().any(|message| {
matches!(
message,
AgentMessage::Command { action, params, .. }
if action == &Action::Navigate &&
params["url"].as_str() == Some("https://zhuanlan.zhihu.com/write")
)
}));
}
#[test]
fn zhihu_article_entry_stops_when_creator_page_has_no_write_entry() {
let _guard = env_lock().lock().unwrap_or_else(|err| err.into_inner());
let workspace_root = temp_workspace_root();
let config_path = write_deepseek_config_with_skills_dir(
&workspace_root,
"deepseek-test-key",
"http://127.0.0.1:9",
"deepseek-chat",
Some(real_skill_lib_root().to_str().unwrap()),
);
let runtime_context = AgentRuntimeContext::new(Some(config_path), workspace_root.clone());
let transport = Arc::new(MockTransport::new(vec![
success_browser_response(1, json!({ "navigated": true })),
success_browser_response(
2,
json!({
"text": {
"status": "creator_home",
"current_url": "https://www.zhihu.com/creator",
"desired_target": "article_editor"
}
}),
),
]));
let browser_tool = BrowserPipeTool::new(
transport.clone(),
zhihu_test_policy(),
vec![1, 2, 3, 4, 5, 6, 7, 8],
)
.with_response_timeout(Duration::from_secs(1));
handle_browser_message_with_context(
transport.as_ref(),
&browser_tool,
&runtime_context,
BrowserMessage::SubmitTask {
instruction: "打开知乎发文章页面".to_string(),
conversation_id: String::new(),
messages: vec![],
page_url: "https://www.zhihu.com/".to_string(),
page_title: "知乎".to_string(),
},
)
.unwrap();
let sent = transport.sent_messages();
assert!(sent.iter().any(|message| {
matches!(
message,
AgentMessage::TaskComplete { success, summary }
if *success && summary.contains("未找到“写文章”入口")
)
}));
assert_eq!(
sent.iter()
.filter(|message| matches!(message, AgentMessage::Command { .. }))
.count(),
2
);
assert!(!sent.iter().any(|message| {
matches!(
message,
AgentMessage::LogEntry { level, message }
if level == "info" && message == "call zhihu-write.prepare_article_editor"
)
}));
}
#[test]
fn zhihu_publish_without_confirmation_returns_confirmation_before_any_browser_probing() {
let _guard = env_lock().lock().unwrap_or_else(|err| err.into_inner());
let workspace_root = temp_workspace_root();
let config_path = write_deepseek_config_with_skills_dir(
&workspace_root,
"deepseek-test-key",
"http://127.0.0.1:9",
"deepseek-chat",
Some(real_skill_lib_root().to_str().unwrap()),
);
let runtime_context = AgentRuntimeContext::new(Some(config_path), workspace_root.clone());
let transport = Arc::new(MockTransport::new(vec![]));
let browser_tool = BrowserPipeTool::new(
transport.clone(),
zhihu_test_policy(),
vec![1, 2, 3, 4, 5, 6, 7, 8],
)
.with_response_timeout(Duration::from_secs(1));
handle_browser_message_with_context(
transport.as_ref(),
&browser_tool,
&runtime_context,
BrowserMessage::SubmitTask {
instruction: "请直接发表这篇知乎文章，标题是测试标题，正文是第一段内容".to_string(),
conversation_id: String::new(),
messages: vec![],
page_url: "https://www.zhihu.com/".to_string(),
page_title: "知乎".to_string(),
},
)
.unwrap();
let sent = transport.sent_messages();
assert!(sent.iter().any(|message| {
matches!(
message,
AgentMessage::TaskComplete { success, summary }
if *success && summary.contains("确认发布")
)
}));
assert!(!sent.iter().any(|message| {
matches!(message, AgentMessage::Command { .. })
}));
assert!(!sent.iter().any(|message| {
matches!(
message,
AgentMessage::LogEntry { level, message }
if level == "info" &&
(message.starts_with("navigate ") ||
message.starts_with("getText ") ||
message.starts_with("click ") ||
message.starts_with("type "))
)
}));
}
#[test]
fn zhihu_publish_after_confirmation_reports_login_block_without_selector_probing() {
let _guard = env_lock().lock().unwrap_or_else(|err| err.into_inner());
let workspace_root = temp_workspace_root();
let config_path = write_deepseek_config_with_skills_dir(
&workspace_root,
"deepseek-test-key",
"http://127.0.0.1:9",
"deepseek-chat",
Some(real_skill_lib_root().to_str().unwrap()),
);
let runtime_context = AgentRuntimeContext::new(Some(config_path), workspace_root.clone());
let transport = Arc::new(MockTransport::new(vec![
success_browser_response(1, json!({ "navigated": true, "url": "https://www.zhihu.com/signin?next=%2Fcreator" })),
success_browser_response(
2,
json!({
"text": {
"status": "login_required",
"current_url": "https://www.zhihu.com/signin?next=%2Fcreator"
}
}),
),
]));
let browser_tool = BrowserPipeTool::new(
transport.clone(),
zhihu_test_policy(),
vec![1, 2, 3, 4, 5, 6, 7, 8],
)
.with_response_timeout(Duration::from_secs(1));
handle_browser_message_with_context(
transport.as_ref(),
&browser_tool,
&runtime_context,
BrowserMessage::SubmitTask {
instruction: "确认发布".to_string(),
conversation_id: "conversation-1".to_string(),
messages: vec![ConversationMessage {
role: "user".to_string(),
content: "请直接发表这篇知乎文章，标题是测试标题，正文是第一段内容".to_string(),
}],
page_url: "https://www.zhihu.com/".to_string(),
page_title: "知乎".to_string(),
},
)
.unwrap();
let sent = transport.sent_messages();
assert!(sent.iter().any(|message| {
matches!(
message,
AgentMessage::TaskComplete { success, summary }
if *success && (summary.contains("未登录") || summary.contains("登录"))
)
}));
assert!(sent.iter().any(|message| {
matches!(
message,
AgentMessage::Command { action, params, .. }
if action == &Action::Navigate &&
params["url"].as_str() == Some("https://www.zhihu.com/creator")
)
}));
assert!(sent.iter().any(|message| {
matches!(message, AgentMessage::Command { action, .. } if action == &Action::Eval)
}));
assert!(!sent.iter().any(|message| {
matches!(
message,
AgentMessage::LogEntry { level, message }
if level == "info" &&
(message.starts_with("getText ") || message.starts_with("click "))
)
}));
}
#[test]
fn browser_orchestration_registers_superrpa_tools_natively() {
let _guard = env_lock().lock().unwrap_or_else(|err| err.into_inner());
@@ -2948,103 +3591,12 @@ fn handle_browser_message_executes_real_zhihu_navigate_skill_flow() {
fn handle_browser_message_executes_real_zhihu_write_skill_flow() {
let _guard = env_lock().lock().unwrap_or_else(|err| err.into_inner());
let first_response = json!({
"choices": [{
"message": {
"content": "",
"tool_calls": [{
"id": "call_1",
"type": "function",
"function": {
"name": "read_skill",
"arguments": serde_json::to_string(&json!({
"name": "zhihu-write"
})).unwrap()
}
}]
}
}]
});
let second_response = json!({
"choices": [{
"message": {
"content": "",
"tool_calls": [
{
"id": "call_2",
"type": "function",
"function": {
"name": "browser_action",
"arguments": serde_json::to_string(&json!({
"action": "navigate",
"expected_domain": "www.zhihu.com",
"url": "https://www.zhihu.com/creator"
})).unwrap()
}
},
{
"id": "call_3",
"type": "function",
"function": {
"name": "browser_action",
"arguments": serde_json::to_string(&json!({
"action": "click",
"expected_domain": "www.zhihu.com",
"selector": "a[href='https://zhuanlan.zhihu.com/write']"
})).unwrap()
}
},
{
"id": "call_4",
"type": "function",
"function": {
"name": "browser_action",
"arguments": serde_json::to_string(&json!({
"action": "type",
"expected_domain": "zhuanlan.zhihu.com",
"selector": "input[placeholder='请输入标题']",
"text": "测试标题",
"clear_first": true
})).unwrap()
}
},
{
"id": "call_5",
"type": "function",
"function": {
"name": "browser_action",
"arguments": serde_json::to_string(&json!({
"action": "type",
"expected_domain": "zhuanlan.zhihu.com",
"selector": ".public-DraftEditor-content",
"text": "第一段内容",
"clear_first": true
})).unwrap()
}
}
]
}
}]
});
let third_response = json!({
"choices": [{
"message": {
"content": "已完成知乎文章草稿填写"
}
}]
});
let (base_url, requests, server_handle) = start_fake_deepseek_server(vec![
first_response,
second_response,
third_response,
]);
let workspace_root = temp_workspace_root();
let skills_dir = real_skill_lib_root();
let config_path = write_deepseek_config_with_skills_dir(
&workspace_root,
"deepseek-test-key",
&base_url,
"http://127.0.0.1:9",
"deepseek-chat",
Some(skills_dir.to_str().unwrap()),
);
@@ -3052,9 +3604,36 @@ fn handle_browser_message_executes_real_zhihu_write_skill_flow() {
let transport = Arc::new(MockTransport::new(vec![
success_browser_response(1, json!({ "navigated": true })),
success_browser_response(2, json!({ "clicked": true })),
success_browser_response(3, json!({ "typed": true })),
success_browser_response(4, json!({ "typed": true })),
success_browser_response(
2,
json!({
"text": {
"status": "creator_entry_clicked",
"current_url": "https://www.zhihu.com/creator",
"next_url": "https://zhuanlan.zhihu.com/write"
}
}),
),
success_browser_response(3, json!({ "navigated": true })),
success_browser_response(
4,
json!({
"text": {
"status": "editor_ready",
"current_url": "https://zhuanlan.zhihu.com/write"
}
}),
),
success_browser_response(
5,
json!({
"text": {
"status": "draft_ready",
"current_url": "https://zhuanlan.zhihu.com/write",
"title": "测试标题"
}
}),
),
]));
let browser_tool = BrowserPipeTool::new(
transport.clone(),
@@ -3076,24 +3655,42 @@ fn handle_browser_message_executes_real_zhihu_write_skill_flow() {
},
)
.unwrap();
server_handle.join().unwrap();
let sent = transport.sent_messages();
let request_bodies = requests.lock().unwrap().clone();
let tool_content = tool_message_content(&request_bodies[1], "call_1").unwrap();
assert!(sent.iter().any(|message| {
matches!(
message,
AgentMessage::TaskComplete { success, summary }
if *success && summary == "已完成知乎文章草稿填写"
if *success && summary == "已进入知乎文章编辑器并写入草稿《测试标题》"
)
}));
assert!(sent.iter().any(|message| {
matches!(
message,
AgentMessage::LogEntry { level, message }
if level == "info" && message == "read_skill zhihu-write@0.1.0"
if level == "mode" && message == "zeroclaw_process_message_primary"
)
}));
assert!(sent.iter().any(|message| {
matches!(
message,
AgentMessage::LogEntry { level, message }
if level == "info" && message == "call zhihu-navigate.open_creator_entry"
)
}));
assert!(sent.iter().any(|message| {
matches!(
message,
AgentMessage::LogEntry { level, message }
if level == "info" && message == "call zhihu-write.prepare_article_editor"
)
}));
assert!(sent.iter().any(|message| {
matches!(
message,
AgentMessage::LogEntry { level, message }
if level == "info" && message == "call zhihu-write.fill_article_draft"
)
}));
assert!(sent.iter().any(|message| {
@@ -3108,28 +3705,18 @@ fn handle_browser_message_executes_real_zhihu_write_skill_flow() {
matches!(
message,
AgentMessage::Command { action, params, .. }
if action == &Action::Click &&
params["selector"].as_str() ==
Some("a[href='https://zhuanlan.zhihu.com/write']")
if action == &Action::Navigate &&
params["url"].as_str() == Some("https://zhuanlan.zhihu.com/write")
)
}));
assert!(sent.iter().any(|message| {
assert!(sent.iter().filter(|message| {
matches!(message, AgentMessage::Command { action, .. } if action == &Action::Eval)
}).count() >= 2);
assert!(!sent.iter().any(|message| {
matches!(
message,
AgentMessage::Command { action, params, .. }
if action == &Action::Type &&
params["selector"].as_str() == Some("input[placeholder='请输入标题']")
AgentMessage::LogEntry { level, message }
if level == "mode" && (message == "compat_llm_primary" || message == "compat_skill_runner_primary")
)
}));
assert!(sent.iter().any(|message| {
matches!(
message,
AgentMessage::Command { action, params, .. }
if action == &Action::Type &&
params["selector"].as_str() == Some(".public-DraftEditor-content")
)
}));
assert_eq!(request_bodies.len(), 3);
assert!(tool_content.len() > 100);
assert!(tool_content.contains("publish a Zhihu article"));
}

146
tests/skill_script_zhihu_navigate_test.py Normal file
View File

@@ -0,0 +1,146 @@
import json
import subprocess
import textwrap
import unittest
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parents[1]
SCRIPT_PATH = (
REPO_ROOT.parent / "skill_lib" / "skills" / "zhihu-navigate" / "scripts" /
"open_creator_entry.js"
)
def run_open_creator_entry(*, body_text: str, selectors: dict[str, list[dict]]) -> dict:
node_script = textwrap.dedent(
f"""
import fs from 'node:fs';
import vm from 'node:vm';
const scriptPath = {json.dumps(str(SCRIPT_PATH))};
const selectorMap = {json.dumps(selectors, ensure_ascii=False)};
const bodyText = {json.dumps(body_text, ensure_ascii=False)};
const source = fs.readFileSync(scriptPath, 'utf8');
function createNode(spec) {{
const node = {{
tagName: String(spec?.tagName || 'DIV').toUpperCase(),
textContent: String(spec?.textContent ?? ''),
innerText: String(spec?.innerText ?? spec?.textContent ?? ''),
href: String(spec?.href ?? ''),
clicked: false,
click() {{
this.clicked = true;
}},
getBoundingClientRect() {{
return {{
width: spec?.visible === false ? 0 : 120,
height: spec?.visible === false ? 0 : 32,
}};
}},
}};
return node;
}}
const created = new Map();
function createNodeList(selector) {{
const specs = selectorMap[selector] || [];
return specs.map((spec, index) => {{
const key = `${{selector}}#${{index}}`;
if (!created.has(key)) {{
created.set(key, createNode(spec));
}}
return created.get(key);
}});
}}
const bodyNode = createNode({{ tagName: 'BODY', textContent: bodyText, innerText: bodyText }});
const context = {{
args: {{ desired_target: 'article_editor' }},
location: {{ href: 'https://www.zhihu.com/creator' }},
document: {{
body: bodyNode,
querySelector(selector) {{
if (selector === 'body') {{
return bodyNode;
}}
return createNodeList(selector)[0] || null;
}},
querySelectorAll(selector) {{
return createNodeList(selector);
}},
}},
console,
JSON,
Math,
Number,
Object,
RegExp,
Set,
String,
Array,
Error,
}};
try {{
const result = vm.runInNewContext(`(function(){{\\n${{source}}\\n}})()`, context);
process.stdout.write(JSON.stringify({{ ok: true, result, created: Object.fromEntries(created) }}));
}} catch (error) {{
process.stdout.write(JSON.stringify({{
ok: false,
error: String(error && error.message ? error.message : error),
}}));
process.exitCode = 1;
}}
"""
)
completed = subprocess.run(
["node", "--input-type=module", "-e", node_script],
check=False,
capture_output=True,
text=True,
)
payload = json.loads(completed.stdout)
if completed.returncode != 0:
raise AssertionError(payload["error"])
return payload
class SkillScriptZhihuNavigateTest(unittest.TestCase):
def test_open_creator_entry_clicks_anchor_write_entry(self):
payload = run_open_creator_entry(
body_text="创作者中心 写文章",
selectors={
"a[href], button, [role='button']": [
{
"tagName": "a",
"textContent": "写文章",
"href": "https://zhuanlan.zhihu.com/write",
}
]
},
)
self.assertEqual(payload["result"]["status"], "creator_entry_clicked")
self.assertTrue(payload["created"]["a[href], button, [role='button']#0"]["clicked"])
def test_open_creator_entry_clicks_button_write_entry(self):
payload = run_open_creator_entry(
body_text="创作者中心 发布内容",
selectors={
"a[href], button, [role='button']": [
{
"tagName": "button",
"textContent": "写文章",
}
]
},
)
self.assertEqual(payload["result"]["status"], "creator_entry_clicked")
self.assertTrue(payload["created"]["a[href], button, [role='button']#0"]["clicked"])
if __name__ == "__main__":
unittest.main()