"""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/"))