130 lines
9.1 KiB
Python
130 lines
9.1 KiB
Python
"""Real isolated DB + local tool observations, without model or platform work."""
|
|
from copy import deepcopy
|
|
|
|
import httpx
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
|
|
from backend.browser import tools
|
|
from backend.db import database as db
|
|
from backend.prospecting import knowledge, orchestrator, routes, workflows
|
|
from backend.prospecting.contracts import KNOWLEDGE_ID, ProspectingBrief
|
|
from backend.prospecting.prompts import browser_prompt
|
|
|
|
|
|
@pytest.mark.db
|
|
async def test_save_learn_recall_proof_export_and_feedback_are_wired(test_school, monkeypatch, tmp_path):
|
|
monkeypatch.setattr(db, "PROFILES_DIR", tmp_path / "profiles")
|
|
def task_dir(tid):
|
|
path = tmp_path / tid
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
return path
|
|
for module in (tools, orchestrator, workflows, routes):
|
|
monkeypatch.setattr(module, "task_doc_dir", task_dir)
|
|
brief = ProspectingBrief(user_notes="找准备装修的人", target_candidates=1)
|
|
task = db.create_task(KNOWLEDGE_ID, "fixture", {**brief.model_dump(), "goal": "找1位准备装修的人", "_goal_version": 1, "_execution_policy": "conversation-v1", "_needs_interpretation": False})
|
|
tid = task["id"]
|
|
async def observe(session, selector):
|
|
return {"url": "https://www.xiaohongshu.com/explore/" + "a" * 24,
|
|
"text": "我准备装修，想找设计师", "source_type": "note" if selector == "#note" else "comment",
|
|
"author": "本地夹具作者", "author_profile_url": "https://www.xiaohongshu.com/user/profile/" + "b" * 24}
|
|
monkeypatch.setattr(tools, "_observe_source", observe)
|
|
tools.set_prospecting_context(tid)
|
|
try:
|
|
for kind in ("note", "comment"):
|
|
await tools._save_source(tid, object(), source_type=kind, content="我准备装修，想找设计师", source_selector="#" + kind)
|
|
artifacts = deepcopy(tools.get_task_artifacts(tid))
|
|
artifacts.update(leads=[{**row, "decision": "keep", "reason": "本人需求", "seeking_service": "yes"} for row in artifacts["sources"]],
|
|
analysis_goal_version=1, browser_verdict="success", browser_history={"is_successful": True}, stop_reason="target_reached")
|
|
db.update_task(tid, artifacts=artifacts, status="completed", current_step="results")
|
|
orchestrator.ProspectingOrchestrator._learn(tid, artifacts)
|
|
saved = db.get_task(tid)
|
|
assert saved["artifacts"]["knowledge_summary"]["outcomes"]["success"] == 3
|
|
assert all(row["stats"]["success"] == 1 for row in knowledge.task_summary(tid)["methods"])
|
|
orchestrator.ProspectingOrchestrator._learn(tid, saved["artifacts"])
|
|
assert len(knowledge.task_summary(tid)["events"]) == 3
|
|
next_task = db.create_task(KNOWLEDGE_ID, "fixture", {**brief.model_dump(), "_goal_version": 1})
|
|
methods = orchestrator.ranked_experience(next_task)
|
|
assert len(methods) == 3 and "保存" in browser_prompt(brief, methods)
|
|
db.update_task(next_task["id"], artifacts={"knowledge_methods": methods}) # JSONB must accept all public projections.
|
|
app = FastAPI()
|
|
app.include_router(routes.router)
|
|
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://fixture") as client:
|
|
fact = artifacts["operation_evidence"][0]
|
|
proof = await client.get(f"/tasks/{tid}/validation-proof", params={"reference": fact["evidence_refs"]["pre"]})
|
|
assert proof.status_code == 200 and proof.json()["pre"]["quote_matched"] is True
|
|
assert proof.json()["post"]["persisted"] is True
|
|
assert (await client.get(f"/tasks/{tid}/validation-proof", params={"reference": "foreign:proof"})).status_code == 404
|
|
summary = knowledge.task_summary(tid)
|
|
stage_method = next(row for row in summary["methods"] if row["kind"] == "pipeline")
|
|
stage_event = next(row for row in summary["events"] if row["method_id"] == stage_method["id"])
|
|
note_fact, discussion_fact = artifacts["operation_evidence"]
|
|
params = {"reference": stage_event["evidence_refs"]["pre"], "event_id": stage_event["id"]}
|
|
stage_proof = await client.get(f"/tasks/{tid}/validation-proof", params=params)
|
|
assert stage_proof.status_code == 200
|
|
stage_body = stage_proof.json()
|
|
assert stage_body["id"] == stage_event["id"] and stage_body["capability"] == "read_discussion"
|
|
assert stage_body["verifier"] == "same_note_saved_sources" and stage_body["outcome"] == "success"
|
|
assert stage_body["observed_at"] == stage_event["created_at"]
|
|
assert stage_body["pre"] == note_fact["details"]["pre_snapshot"]
|
|
assert stage_body["post"] == discussion_fact["details"]["post_snapshot"]
|
|
assert stage_body["post"]["source_type"] == "comment"
|
|
assert (await client.get(f"/tasks/{tid}/validation-proof", params={**params, "event_id": "validation_foreign"})).status_code == 404
|
|
assert (await client.get(f"/tasks/{next_task['id']}/validation-proof", params=params)).status_code == 404
|
|
assert (await client.get(f"/tasks/{tid}/validation-proof", params={**params, "reference": discussion_fact["evidence_refs"]["pre"]})).status_code == 404
|
|
assert (await client.get(f"/tasks/{tid}/validation-proof", params={**params, "reference": stage_event["evidence_refs"]["post"]})).status_code == 404
|
|
# Missing either persisted half fails closed, rather than returning
|
|
# the note operation's own post-state as the whole stage result.
|
|
original_summary = knowledge.task_summary
|
|
missing = deepcopy(summary)
|
|
missing_event = next(row for row in missing["events"] if row["id"] == stage_event["id"])
|
|
missing_event["evidence_refs"]["post"] = f"{tid}:missing:post"
|
|
with monkeypatch.context() as scoped:
|
|
scoped.setattr(knowledge, "task_summary", lambda _tid: missing)
|
|
assert (await client.get(f"/tasks/{tid}/validation-proof", params=params)).status_code == 404
|
|
missing_event["evidence_refs"]["post"] = stage_event["evidence_refs"]["post"]
|
|
missing_event["evidence_refs"]["pre"] = f"{tid}:missing:pre"
|
|
assert (await client.get(f"/tasks/{tid}/validation-proof", params={**params, "reference": missing_event["evidence_refs"]["pre"]})).status_code == 404
|
|
assert knowledge.task_summary is original_summary
|
|
before = knowledge.list_methods(include_legacy=False)
|
|
feedback = await client.post(f"/tasks/{tid}/feedback", json={"rating": "up", "reason": "本地夹具反馈"})
|
|
assert feedback.status_code == 200
|
|
outcome = await client.post(f"/tasks/{tid}/business-outcome", json={"source_id": artifacts["sources"][0]["source_id"], "outcome": "replied"})
|
|
assert outcome.status_code == 200
|
|
old_inputs = db.get_task(tid)["inputs"]
|
|
correction = await client.post(f"/knowledge/methods/{before[0]['id']}/feedback", json={"task_id": tid, "text": "这次入口和预期不同，请后续核对"})
|
|
assert correction.status_code == 200 and db.get_task(tid)["inputs"] == old_inputs
|
|
assert knowledge.list_methods(include_legacy=False) == before
|
|
for _ in range(2):
|
|
assert (await client.post(f"/tasks/{tid}/export")).status_code == 200
|
|
export_methods = [row for row in knowledge.list_methods(include_legacy=False) if row["capability"] == "export"]
|
|
assert len(export_methods) == 1 and export_methods[0]["stats"]["success"] == 1
|
|
assert len(knowledge.task_summary(tid)["business_feedback"]) == 3
|
|
finally:
|
|
tools.clear_task_context(tid)
|
|
|
|
|
|
@pytest.mark.db
|
|
async def test_export_only_creates_real_current_file_without_key_and_failure_is_not_complete(test_school, monkeypatch, tmp_path):
|
|
monkeypatch.setattr(db, "PROFILES_DIR", tmp_path / "profiles")
|
|
def task_dir(tid):
|
|
path = tmp_path / tid
|
|
path.mkdir(exist_ok=True)
|
|
return path
|
|
for module in (orchestrator, workflows, routes):
|
|
monkeypatch.setattr(module, "task_doc_dir", task_dir)
|
|
brief = ProspectingBrief(intent="export", user_notes="导出已有名单")
|
|
row = db.create_task(KNOWLEDGE_ID, "fixture", {**brief.model_dump(), "goal": "导出已有名单", "_goal_version": 1,
|
|
"_execution_policy": "conversation-v1", "_needs_interpretation": False})
|
|
monkeypatch.setattr(orchestrator, "get_llm_config", lambda: {"api_key": ""})
|
|
engine = orchestrator.ProspectingOrchestrator()
|
|
done = await engine._prepare(row)
|
|
assert done["status"] == "completed"
|
|
files = workflows.list_exports(done)
|
|
assert len(files) == 1 and files[0]["available"] and files[0]["is_current"]
|
|
assert knowledge.task_summary(row["id"])["outcomes"]["success"] == 1
|
|
second = db.create_task(KNOWLEDGE_ID, "fixture", row["inputs"])
|
|
monkeypatch.setattr(workflows, "export_leads", lambda task: (_ for _ in ()).throw(OSError("fixture unavailable storage")))
|
|
failed = await engine._prepare(second)
|
|
assert failed["status"] == "blocked" and failed["artifacts"]["stop_reason"] == "file_error"
|