254 lines
13 KiB
Python
254 lines
13 KiB
Python
"""Guarded outreach state-machine tests; no real browser or recipient."""
|
|
from copy import deepcopy
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
|
|
from backend.prospecting import outreach
|
|
|
|
ACCOUNT = "b" * 24
|
|
TARGET = "a" * 24
|
|
OTHER = "c" * 24
|
|
ACCOUNT_URL = "https://www.xiaohongshu.com/user/profile/" + ACCOUNT
|
|
TARGET_URL = "https://www.xiaohongshu.com/user/profile/" + TARGET
|
|
OTHER_URL = "https://www.xiaohongshu.com/user/profile/" + OTHER
|
|
|
|
|
|
@pytest.fixture
|
|
def memory(monkeypatch, tmp_path):
|
|
task = {"id": "t", "knowledge_id": "xiaohongshu_prospecting", "school_id": "workspace", "inputs": {"_goal_version": 1},
|
|
"artifacts": {"sources": [{"source_id": "s", "author_profile_url": TARGET_URL}, {"source_id": "s2", "author_profile_url": OTHER_URL}],
|
|
"profiles": [{"source_id": "s", "profile_url": TARGET_URL, "observed_at": "now"}, {"source_id": "s2", "profile_url": OTHER_URL, "observed_at": "now"}],
|
|
"outreach_account": {"account_id": ACCOUNT, "account_profile_url": ACCOUNT_URL, "goal_version": 1}}}
|
|
store = {"t": deepcopy(task)}
|
|
monkeypatch.setattr(outreach, "ACCOUNT_DIR", tmp_path / "accounts")
|
|
monkeypatch.setattr(outreach, "task_doc_dir", lambda task_id: tmp_path / task_id)
|
|
monkeypatch.setattr(outreach.db, "get_task", lambda task_id: deepcopy(store.get(task_id)))
|
|
def update(task_id, **changes):
|
|
store[task_id].update(deepcopy(changes))
|
|
return deepcopy(store[task_id])
|
|
monkeypatch.setattr(outreach.db, "update_task", update)
|
|
outreach._locks.clear()
|
|
return store
|
|
|
|
|
|
def observed(**changes):
|
|
return {"account_id": ACCOUNT, "account_profile_url": ACCOUNT_URL, "target_id": TARGET,
|
|
"follow_text": "关注", "follow_selector": "#follow", "chat_verified": True,
|
|
"editor_selector": "#editor", "editor_value": "明确话术", "send_selector": "#send", "outgoing_count": 0,
|
|
"send_error": False, "daily_limit": False, "alerts": [], **changes}
|
|
|
|
|
|
def browser(monkeypatch, values):
|
|
page = AsyncMock()
|
|
editor, button = AsyncMock(), AsyncMock()
|
|
async def elements(selector):
|
|
return [editor if selector == "#editor" else button]
|
|
page.get_elements_by_css_selector.side_effect = elements
|
|
monkeypatch.setattr(outreach, "_observe", AsyncMock(side_effect=[(page, value) for value in values]))
|
|
monkeypatch.setattr(outreach, "_save_failure_screenshot", AsyncMock())
|
|
return page, editor, button
|
|
|
|
|
|
def test_authorization_requires_preflight_and_verified_targets(memory):
|
|
store = memory
|
|
del store["t"]["artifacts"]["outreach_account"]
|
|
with pytest.raises(ValueError, match="实际页面核对"):
|
|
outreach.authorize_outreach(store["t"], "follow", ["s"], account_profile_url=ACCOUNT_URL)
|
|
store["t"]["artifacts"]["outreach_account"] = {"account_id": ACCOUNT, "goal_version": 1}
|
|
with pytest.raises(ValueError, match="明确消息内容"):
|
|
outreach.authorize_outreach(store["t"], "message", ["s"], account_profile_url=ACCOUNT_URL)
|
|
store["t"]["artifacts"]["profiles"][0]["profile_url"] = OTHER_URL
|
|
with pytest.raises(ValueError, match="不一致"):
|
|
outreach.authorize_outreach(store["t"], "follow", ["s"], account_profile_url=ACCOUNT_URL)
|
|
|
|
|
|
def test_follow_authorization_does_not_require_message_content(memory):
|
|
grant = outreach.authorize_outreach(memory["t"], "follow", ["s"], account_profile_url=ACCOUNT_URL)
|
|
assert grant["content"] == ""
|
|
assert grant["authorized_by"] == "user_batch_request"
|
|
assert memory["t"]["artifacts"]["outreach_grants"][0]["id"] == grant["id"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unauthorized_action_cannot_click(memory, monkeypatch):
|
|
_, editor, button = browser(monkeypatch, [observed()])
|
|
with pytest.raises(ValueError, match="没有对应"):
|
|
await outreach.perform_action("t", "invented", "s", "follow", object())
|
|
editor.fill.assert_not_awaited()
|
|
button.click.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_follow_is_recorded_before_native_click_and_verified_after(memory, monkeypatch):
|
|
grant = outreach.authorize_outreach(memory["t"], "follow", ["s"], account_profile_url=ACCOUNT_URL)
|
|
_, _, button = browser(monkeypatch, [observed(), observed(follow_text="已关注")])
|
|
async def check_before_click():
|
|
actions = outreach._read_account(memory["t"], ACCOUNT)["actions"]
|
|
assert actions[0]["status"] == "executing"
|
|
button.click.side_effect = check_before_click
|
|
result = await outreach.perform_action("t", grant["id"], "s", "follow", object())
|
|
assert result["status"] == "success"
|
|
assert memory["t"]["artifacts"]["action_ledger"][0]["status"] == "success"
|
|
button.click.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_already_followed_is_skipped_without_click(memory, monkeypatch):
|
|
grant = outreach.authorize_outreach(memory["t"], "follow", ["s"], account_profile_url=ACCOUNT_URL)
|
|
_, _, button = browser(monkeypatch, [observed(follow_text="已关注")])
|
|
result = await outreach.perform_action("t", grant["id"], "s", "follow", object())
|
|
assert result["status"] == "skipped_already_followed"
|
|
button.click.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_account_change_prevents_follow(memory, monkeypatch):
|
|
grant = outreach.authorize_outreach(memory["t"], "follow", ["s"], account_profile_url=ACCOUNT_URL)
|
|
_, _, button = browser(monkeypatch, [observed(account_id=OTHER)])
|
|
with pytest.raises(ValueError, match="账号或目标"):
|
|
await outreach.perform_action("t", grant["id"], "s", "follow", object())
|
|
button.click.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_revision_invalidates_old_authorization(memory, monkeypatch):
|
|
grant = outreach.authorize_outreach(memory["t"], "follow", ["s"], account_profile_url=ACCOUNT_URL)
|
|
memory["t"]["inputs"]["_goal_version"] = 2
|
|
_, _, button = browser(monkeypatch, [observed()])
|
|
with pytest.raises(ValueError, match="目标已经修订"):
|
|
await outreach.perform_action("t", grant["id"], "s", "follow", object())
|
|
button.click.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_message_requires_new_visible_outgoing_record_for_success(memory, monkeypatch):
|
|
grant = outreach.authorize_outreach(memory["t"], "message", ["s"], "明确话术", ACCOUNT_URL)
|
|
_, editor, button = browser(monkeypatch, [observed(), observed(), observed(outgoing_count=1)])
|
|
result = await outreach.perform_action("t", grant["id"], "s", "message", object())
|
|
assert result["status"] == "success"
|
|
editor.fill.assert_awaited_once_with("明确话术")
|
|
button.click.assert_awaited_once()
|
|
# A new grant cannot duplicate a completed send to the same account.
|
|
second = outreach.authorize_outreach(memory["t"], "message", ["s"], "明确话术", ACCOUNT_URL)
|
|
browser(monkeypatch, [observed(outgoing_count=1)])
|
|
with pytest.raises(ValueError, match="不会重复"):
|
|
await outreach.perform_action("t", second["id"], "s", "message", object())
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("daily", [False, True])
|
|
async def test_red_failure_pauses_all_account_messages_without_guessing_daily_limit(memory, monkeypatch, daily):
|
|
grant = outreach.authorize_outreach(memory["t"], "message", ["s", "s2"], "明确话术", ACCOUNT_URL)
|
|
_, _, button = browser(monkeypatch, [observed(), observed(), observed(send_error=True, daily_limit=daily)])
|
|
result = await outreach.perform_action("t", grant["id"], "s", "message", object())
|
|
assert result["status"] == "failed"
|
|
assert result["daily_limit"] is daily
|
|
pause = outreach._read_account(memory["t"], ACCOUNT)["message_pause"]
|
|
assert pause["daily_limit"] is daily
|
|
browser(monkeypatch, [observed(target_id=OTHER)])
|
|
with pytest.raises(ValueError, match="所有后续私信已暂停"):
|
|
await outreach.perform_action("t", grant["id"], "s2", "message", object())
|
|
with pytest.raises(ValueError, match="私信阶段已暂停"):
|
|
outreach.authorize_outreach(memory["t"], "message", ["s2"], "另一话术", ACCOUNT_URL)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_timeout_during_click_is_unknown_and_never_retried(memory, monkeypatch):
|
|
grant = outreach.authorize_outreach(memory["t"], "message", ["s"], "明确话术", ACCOUNT_URL)
|
|
_, _, button = browser(monkeypatch, [observed(), observed()])
|
|
button.click.side_effect = TimeoutError("interrupted")
|
|
with pytest.raises(TimeoutError):
|
|
await outreach.perform_action("t", grant["id"], "s", "message", object())
|
|
ledger = outreach._read_account(memory["t"], ACCOUNT)
|
|
assert ledger["actions"][0]["status"] == "unknown"
|
|
assert ledger["message_pause"]["daily_limit"] is False
|
|
assert memory["t"]["artifacts"]["action_ledger"][0]["status"] == "unknown"
|
|
|
|
|
|
def test_broken_account_ledger_is_not_reset_to_empty(memory):
|
|
path = outreach._account_path(memory["t"], ACCOUNT)
|
|
path.parent.mkdir(parents=True)
|
|
path.write_text('{"actions":', encoding="utf-8")
|
|
with pytest.raises(ValueError, match="不能将损坏记录重置"):
|
|
outreach.authorize_outreach(memory["t"], "message", ["s"], "话术", ACCOUNT_URL)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_wrong_editor_content_is_never_sent(memory, monkeypatch):
|
|
grant = outreach.authorize_outreach(memory["t"], "message", ["s"], "明确话术", ACCOUNT_URL)
|
|
_, _, button = browser(monkeypatch, [observed(), observed(editor_value="页面中出现了其他内容")])
|
|
with pytest.raises(ValueError, match="话术不一致"):
|
|
await outreach.perform_action("t", grant["id"], "s", "message", object())
|
|
button.click.assert_not_awaited()
|
|
assert memory["t"]["artifacts"]["action_ledger"][0]["status"] == "not_sent"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_read_only_open_target_does_not_require_preexisting_profile(memory, monkeypatch):
|
|
memory["t"]["artifacts"]["profiles"] = []
|
|
page = AsyncMock()
|
|
page.evaluate.return_value = "#actual-avatar"
|
|
button = AsyncMock()
|
|
page.get_elements_by_css_selector.return_value = [button]
|
|
session = AsyncMock()
|
|
session.get_current_page.return_value = page
|
|
result = await outreach.open_target("t", "s", session)
|
|
assert result["status"] == "clicked_observed_author_link"
|
|
button.click.assert_awaited_once()
|
|
|
|
|
|
def paused_account(store):
|
|
ledger = outreach._read_account(store["t"], ACCOUNT)
|
|
ledger["actions"] = [{"id": "uncertain", "status": "unknown", "dedup_key": "old"}]
|
|
ledger["message_pause"] = {"at": "2026-10-28T19:00:00+08:00", "daily_limit": False, "message": "发送结果待核"}
|
|
outreach._write_account(store["t"], ledger)
|
|
store["t"]["artifacts"]["outreach_pause"] = deepcopy(ledger["message_pause"])
|
|
store["t"]["artifacts"]["outreach_account"].update(observed_at="2026-10-28T19:01:00+08:00", send_error=False, daily_limit=False, chat_verified=True)
|
|
|
|
|
|
def test_acknowledgment_requires_explicit_user_confirmation_and_fresh_clean_observation(memory):
|
|
paused_account(memory)
|
|
with pytest.raises(ValueError, match="明确确认"):
|
|
outreach.acknowledge_account_pause(memory["t"], ACCOUNT_URL, False)
|
|
# Non-empty arbitrary strings are not treated as explicit boolean approval.
|
|
with pytest.raises(ValueError, match="明确确认"):
|
|
outreach.acknowledge_account_pause(memory["t"], ACCOUNT_URL, "我还没有确认")
|
|
memory["t"]["artifacts"]["outreach_account"]["observed_at"] = "2026-10-28T18:59:00+08:00"
|
|
with pytest.raises(ValueError, match="旧截图"):
|
|
outreach.acknowledge_account_pause(memory["t"], ACCOUNT_URL, True)
|
|
memory["t"]["artifacts"]["outreach_account"].update(observed_at="2026-10-28T19:01:00+08:00", send_error=True)
|
|
with pytest.raises(ValueError, match="仍有发送失败"):
|
|
outreach.acknowledge_account_pause(memory["t"], ACCOUNT_URL, True)
|
|
memory["t"]["artifacts"]["outreach_account"].update(send_error=False, daily_limit=True)
|
|
with pytest.raises(ValueError, match="平台限制"):
|
|
outreach.acknowledge_account_pause(memory["t"], ACCOUNT_URL, True)
|
|
|
|
|
|
def test_acknowledgment_clears_pause_but_preserves_unknown_action_and_pause_history(memory):
|
|
paused_account(memory)
|
|
event = outreach.acknowledge_account_pause(memory["t"], ACCOUNT_URL, True)
|
|
ledger = outreach._read_account(memory["t"], ACCOUNT)
|
|
assert ledger["message_pause"] is None
|
|
assert ledger["actions"][0]["status"] == "unknown"
|
|
assert ledger["message_pause_history"][0]["previous_pause"]["message"] == "发送结果待核"
|
|
assert event["confirmed_by"] == "user_request"
|
|
assert "outreach_pause" not in memory["t"]["artifacts"]
|
|
assert len(memory["t"]["artifacts"]["outreach_pause_history"]) == 1
|
|
|
|
|
|
def test_acknowledgment_cannot_use_old_goal_or_unknown_failure_state(memory):
|
|
paused_account(memory)
|
|
memory["t"]["artifacts"]["outreach_account"]["goal_version"] = 0
|
|
with pytest.raises(ValueError, match="旧目标"):
|
|
outreach.acknowledge_account_pause(memory["t"], ACCOUNT_URL, True)
|
|
memory["t"]["artifacts"]["outreach_account"]["goal_version"] = 1
|
|
del memory["t"]["artifacts"]["outreach_account"]["send_error"]
|
|
with pytest.raises(ValueError, match="页面仍有"):
|
|
outreach.acknowledge_account_pause(memory["t"], ACCOUNT_URL, True)
|
|
|
|
|
|
def test_profile_identity_rejects_nonstandard_https_port():
|
|
with pytest.raises(ValueError, match="主页身份"):
|
|
outreach.profile_id(TARGET_URL.replace(".com/", ".com:8443/"))
|