128 lines
6.5 KiB
Python
128 lines
6.5 KiB
Python
"""Delivery and learning retain provenance without an extra browser run."""
|
|
import copy
|
|
import csv
|
|
import io
|
|
import json
|
|
import math
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from backend.browser import runner
|
|
from backend.prospecting import workflows
|
|
|
|
|
|
@pytest.fixture
|
|
def task(tmp_path, monkeypatch):
|
|
row = {"id": "delivery-test", "inputs": {"goal": "杭州装修候选"}, "artifacts": {
|
|
"sources": [{"source_id": "c1", "source_type": "comment", "source_selector": "#comment", "content": "找装修", "author": "=1+1", "source_url": "https://www.xiaohongshu.com/explore/a", "observed_at": "2026-10-28"}],
|
|
"leads": [{"source_id": "c1", "quote": "找装修", "reason": "待核", "status": "pending"}],
|
|
"decisions": [{"source_id": "c1", "decision": "keep"}],
|
|
}}
|
|
monkeypatch.setattr(workflows, "task_doc_dir", lambda tid: tmp_path)
|
|
monkeypatch.setattr(workflows.db, "update_task", lambda tid, **fields: row.update(copy.deepcopy(fields)))
|
|
return row
|
|
|
|
|
|
def test_feedback_is_versioned_and_never_changes_decisions(task):
|
|
original = copy.deepcopy(task["artifacts"]["decisions"])
|
|
event = workflows.record_feedback(task, "up")
|
|
assert task["artifacts"]["decisions"] == original
|
|
snapshot = copy.deepcopy(task["artifacts"]["result_snapshots"][event["result_version"]])
|
|
task["artifacts"]["leads"].clear()
|
|
assert task["artifacts"]["result_snapshots"][event["result_version"]] == snapshot
|
|
assert workflows.result_version(task) != event["result_version"]
|
|
|
|
|
|
def test_correction_preserves_original_evidence_and_scope(task):
|
|
before = copy.deepcopy(task["artifacts"]["sources"])
|
|
workflows.record_correction(task, "c1", "这句是玩笑")
|
|
assert task["artifacts"]["sources"] == before
|
|
assert task["artifacts"]["corrections"][0]["source_id"] == "c1"
|
|
with pytest.raises(ValueError):
|
|
workflows.record_correction(task, "foreign", "no")
|
|
|
|
|
|
def test_export_does_not_invent_account_and_escapes_formula(task):
|
|
result = workflows.export_leads(task)
|
|
from pathlib import Path
|
|
rows = list(csv.reader(io.StringIO(Path(result["path"]).read_text(encoding="utf-8-sig"))))
|
|
assert result["count"] == 1
|
|
assert rows[1][0] == "'=1+1"
|
|
assert rows[1][1:3] == ["", ""]
|
|
|
|
|
|
def test_export_uses_visible_account_label_not_internal_profile_id(task):
|
|
task["artifacts"]["profiles"] = [{"source_id": "c1", "profile_id": "internal-id", "xiaohongshu_id": "visible-account", "profile_url": "https://www.xiaohongshu.com/user/profile/aaaaaaaaaaaaaaaaaaaaaaaa", "observed_at": "2026-10-29T10:00:00+08:00", "evidence": "matched_source_author_link_and_visible_profile_header"}]
|
|
result = workflows.export_leads(task)
|
|
from pathlib import Path
|
|
text = Path(result["path"]).read_text(encoding="utf-8-sig")
|
|
assert "visible-account" in text and "internal-id" not in text
|
|
|
|
|
|
def test_review_requires_observed_sources_and_never_shares_personal_words(task):
|
|
reviewed = workflows.finalize_review(task["id"], task["artifacts"])
|
|
assert reviewed["review"]["draft_count"] == 1
|
|
assert "找装修" not in json.dumps(reviewed["learning_candidates"], ensure_ascii=False)
|
|
again = workflows.finalize_review(task["id"], reviewed)
|
|
assert len(again["learning_candidates"]) == len(reviewed["learning_candidates"])
|
|
assert workflows.finalize_review(task["id"], {"sources": [{"source_type": "note", "content": "not verified"}]})["review"]["draft_count"] == 0
|
|
|
|
|
|
async def test_feedback_and_correction_routes_return_current_task_not_event_id(task, tmp_path, monkeypatch):
|
|
import httpx
|
|
from fastapi import FastAPI
|
|
from backend.prospecting import routes
|
|
task.update(status="blocked", current_step="stage_results", created_at="2026-10-28", updated_at="2026-10-28")
|
|
monkeypatch.setattr(routes, "task_for", lambda tid: task)
|
|
monkeypatch.setattr(routes, "task_doc_dir", lambda tid: tmp_path)
|
|
monkeypatch.setattr(routes.runner, "has_live_session", lambda tid: False)
|
|
monkeypatch.setattr(routes.runner, "is_taken_over", lambda tid: False)
|
|
app = FastAPI()
|
|
app.include_router(routes.router)
|
|
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
|
|
response = await client.post("/tasks/delivery-test/feedback", json={"rating": "up"})
|
|
assert response.status_code == 200
|
|
assert response.json()["id"] == task["id"]
|
|
assert response.json()["artifacts"]["feedback"][-1]["rating"] == "up"
|
|
response = await client.post("/tasks/delivery-test/leads/c1/correction", json={"text": "已装修完"})
|
|
assert response.status_code == 200
|
|
assert response.json()["artifacts"]["leads"][0]["correction"] == "已装修完"
|
|
assert response.json()["artifacts"]["alternatives"]
|
|
|
|
|
|
async def test_unlimited_execution_can_resume_beyond_legacy_sixty_steps(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(runner, "task_doc_dir", lambda tid: tmp_path)
|
|
monkeypatch.setattr(runner, "append_progress", lambda *args, **kwargs: None)
|
|
state = SimpleNamespace(n_steps=1)
|
|
class Agent:
|
|
async def run(self, max_steps, on_step_start):
|
|
assert math.isinf(max_steps)
|
|
for _ in range(3):
|
|
await on_step_start(self)
|
|
state.n_steps += 1
|
|
return SimpleNamespace(number_of_steps=lambda: 3, is_done=lambda: True, is_successful=lambda: True,
|
|
final_result=lambda: "阶段结果", total_duration_seconds=lambda: 1, urls=lambda: [],
|
|
action_names=lambda: [], errors=lambda: [])
|
|
agent = Agent()
|
|
agent.state = state
|
|
result = await runner._run_agent("unlimited", runner.LiveSession(agent, None), agent, None, keep_session=True, budget_steps_used=65)
|
|
assert result["history"]["total_steps_used"] == 68
|
|
assert result["stop_reason"] == "agent_done"
|
|
record = runner.read_browser_budget("unlimited")
|
|
assert record["max_total_steps"] is None and record["total_steps_used"] == 68
|
|
|
|
|
|
async def test_goal_agent_omits_budget_context_and_normalizes_telemetry():
|
|
class Base:
|
|
async def _prepare_context(self, info):
|
|
return info
|
|
def _log_agent_event(self, max_steps, agent_run_error=None):
|
|
return max_steps
|
|
agent = runner._prospecting_agent_class(Base)()
|
|
agent._goal_based = True
|
|
assert await agent._prepare_context(SimpleNamespace(max_steps=60)) is None
|
|
assert agent._log_agent_event(math.inf) == 0
|
|
assert runner.stop_reason_from_history({"is_done": True, "is_successful": False}, None) == "agent_reported_failure"
|
|
assert runner.stop_reason_from_history({"errors": ["HTTP 429"]}, None) == "model_rate_limit"
|