294 lines
16 KiB
Python
294 lines
16 KiB
Python
"""v0.3 method evidence fixtures: no model, browser, jobs or user data."""
|
|
from copy import deepcopy
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
|
|
from backend.config import settings
|
|
from backend.db import database as db
|
|
from backend.prospecting import knowledge as knowledge
|
|
|
|
|
|
CONTEXT = {"platform": "xiaohongshu", "stage": "search_read", "capability": "save_source",
|
|
"source_kind": "comment", "page_variant": "xhs_visible_dom_v1"}
|
|
|
|
|
|
def method(**updates):
|
|
spec = knowledge.normalize_spec({"key": "xhs_save_comment", "version": "1", "title": "保存当前评论",
|
|
**CONTEXT, "origin": "program_template", "preconditions": ["当前独立容器可见"],
|
|
"steps": ["读取当前容器并核对作者及原话", "持久化来源"],
|
|
"postcondition": {"persisted": True}, "recovery": ["归属未知时停止"], **updates})
|
|
return {"id": knowledge._digest([spec["key"], spec["version"]], "method_"), **spec}
|
|
|
|
|
|
def event(task="task_1", obj="note_1", outcome="success", current=None, **updates):
|
|
result = knowledge.prepare_event(current or method(), task, obj, outcome,
|
|
{"pre": f"{task}:evidence:before", "post": f"{task}:evidence:after"},
|
|
CONTEXT, verified=True, verifier="source_container_quote_and_persisted", **updates)
|
|
return result
|
|
|
|
|
|
def evaluate(events, current=None, context=None):
|
|
return knowledge.evaluate_method(current or method(), events, CONTEXT if context is None else context)
|
|
|
|
|
|
def test_new_templates_and_model_suggestions_have_no_automatic_success():
|
|
for origin in ("program_template", "model", "observed"):
|
|
result = evaluate([], method(origin=origin))
|
|
assert result["status"] == "exploration" and result["stats"]["success"] == 0
|
|
assert "score" not in result and "reliability" not in result
|
|
|
|
|
|
@pytest.mark.parametrize("missing", ["pre", "post", "verifier", "verified"])
|
|
def test_success_requires_program_verdict_and_pre_post_proofs(missing):
|
|
refs = {"pre": "task_1:proof:before", "post": "task_1:proof:after"}
|
|
meta = {"verified": True, "verifier": "fixture_verifier"}
|
|
if missing in refs:
|
|
refs.pop(missing)
|
|
else:
|
|
meta.pop(missing)
|
|
result = knowledge.prepare_event(method(), "task_1", "note_1", "success", refs, CONTEXT, **meta)
|
|
assert result["outcome"] == "unknown"
|
|
assert evaluate([result])["stats"]["success"] == 0
|
|
|
|
|
|
def test_model_reports_are_preserved_separately_and_never_become_success_or_contradiction():
|
|
for outcome in ("confirmed", "contradicted"):
|
|
result = event(outcome=outcome, provenance="model")
|
|
assert result["requested_outcome"] in {"success", "contradiction"}
|
|
assert result["outcome"] == "unknown" and result["provenance"] == "model"
|
|
assert evaluate([result])["status"] == "exploration"
|
|
|
|
|
|
def test_repeated_same_event_task_or_object_cannot_inflate_independent_counts():
|
|
one = event()
|
|
assert event() == one
|
|
rows = [one, deepcopy(one), event(obj="note_2"), event(task="task_2"), event(task="task_3")]
|
|
stats = evaluate(rows)["stats"]
|
|
assert stats["success"] == 2
|
|
assert stats["independent_tasks"] == stats["independent_objects"] == 2
|
|
assert stats["coverage_tasks"] == 3 and stats["coverage_objects"] == 2
|
|
|
|
|
|
def test_independence_uses_task_object_matching_instead_of_greedy_losing_valid_sample():
|
|
rows = [event("t1", "o1"), event("t1", "o2"), event("t2", "o1"), event("t3", "o3")]
|
|
result = evaluate(rows)
|
|
assert result["status"] == "preferred" and result["stats"]["success"] == 3
|
|
assert evaluate(list(reversed(rows)))["sample_event_ids"] == result["sample_event_ids"]
|
|
|
|
|
|
def test_three_different_tasks_and_objects_required_for_preferred():
|
|
assert evaluate([event("t1", "o1")])["status"] == "trial"
|
|
assert evaluate([event("t1", "o1"), event("t2", "o2")])["status"] == "trial"
|
|
assert evaluate([event("t1", "o1"), event("t2", "o2"), event("t3", "o3")])["status"] == "preferred"
|
|
assert evaluate([event("t1", "o1"), event("t2", "o1"), event("t3", "o1")])["status"] == "trial"
|
|
|
|
|
|
def test_any_unsuperseded_contradiction_pauses_formerly_preferred_even_for_repeat_object():
|
|
rows = [event(f"t{i}", f"o{i}") for i in range(1, 5)]
|
|
assert evaluate(rows)["status"] == "preferred"
|
|
rows.extend([event("t5", "o1", "contradiction"), event("t6", "o1", "success")])
|
|
result = evaluate(rows)
|
|
assert result["status"] == "paused" and not result["applicability"]
|
|
assert "unresolved_contradiction" in result["match_reasons"]
|
|
assert result["stats"]["contradiction"] == 1
|
|
assert result["stats"]["success"] == 3
|
|
|
|
|
|
def test_correction_is_append_only_and_recalculates_both_counts_and_pause():
|
|
bad = event(outcome="contradiction")
|
|
repaired = event(supersedes=bad["id"], event_key="rechecked-proof")
|
|
original = deepcopy([bad, repaired])
|
|
result = evaluate([bad, repaired])
|
|
assert result["status"] == "trial" and result["stats"]["contradiction"] == 0
|
|
assert result["stats"]["success"] == 1 and result["sample_event_ids"] == [repaired["id"]]
|
|
assert [bad, repaired] == original
|
|
# Conflicting correction branches retain both facts, conservatively paused.
|
|
conflict = event(outcome="contradiction", supersedes=bad["id"], event_key="other-recheck")
|
|
assert evaluate([bad, repaired, conflict])["status"] == "paused"
|
|
|
|
|
|
def test_correction_to_unknown_removes_old_success_without_creating_failure():
|
|
good = event()
|
|
missing = event(outcome="unknown", supersedes=good["id"], event_key="uncertain")
|
|
result = evaluate([good, missing])
|
|
assert result["status"] == "exploration"
|
|
assert result["stats"]["success"] == result["stats"]["contradiction"] == 0
|
|
assert result["stats"]["unknown"] == 1
|
|
|
|
|
|
def test_block_program_error_unknown_and_unused_are_distinct_non_denominator_events():
|
|
rows = [event(outcome=outcome, event_key=outcome) for outcome in ("blocked", "program_error", "unused", "unknown")]
|
|
result = evaluate(rows)
|
|
assert result["status"] == "exploration"
|
|
assert all(result["stats"][outcome] == 1 for outcome in ("blocked", "program_error", "unused", "unknown"))
|
|
assert result["stats"]["success"] == result["stats"]["contradiction"] == 0
|
|
|
|
|
|
@pytest.mark.parametrize("field,other", [("platform", "other"), ("stage", "delivery"),
|
|
("source_kind", "reply"), ("page_variant", "xhs_dom_v2"),
|
|
("capability", "inspect_profile")])
|
|
def test_incompatible_context_never_inherits_prior_rank(field, other):
|
|
result = evaluate([event(f"t{i}", f"o{i}") for i in range(3)], context={**CONTEXT, field: other})
|
|
assert result["status"] == "paused" and not result["applicability"]
|
|
assert f"{field}_mismatch" in result["match_reasons"]
|
|
|
|
|
|
def test_page_scenes_and_method_versions_have_separate_statistics():
|
|
adaptable = method(page_variant="any")
|
|
old = event(current=adaptable)
|
|
new_context = {**CONTEXT, "page_variant": "xhs_dom_v2"}
|
|
assert evaluate([old], adaptable, new_context)["stats"]["success"] == 0
|
|
assert evaluate([old], method(version="2"))["stats"]["success"] == 0
|
|
|
|
|
|
def test_versioned_dependencies_and_authorization_gate_applicability_before_score():
|
|
dependent = method(dependencies=[{"key": "read_container", "version": "2"}])
|
|
assert not knowledge.applicability(dependent, CONTEXT)[0]
|
|
assert not knowledge.applicability(dependent, {**CONTEXT, "dependency_versions": {"read_container": "1"}})[0]
|
|
assert knowledge.applicability(dependent, {**CONTEXT, "dependency_versions": {"read_container": "2"}})[0]
|
|
follow = method(capability="follow")
|
|
assert not knowledge.applicability(follow, {"capability": "follow"})[0]
|
|
assert knowledge.applicability(follow, {"capability": "follow", "authorized_actions": ["follow"]})[0]
|
|
|
|
|
|
def test_pipeline_cannot_derive_terminal_success_from_components_or_undocumented_branch():
|
|
pipeline = method(kind="pipeline", terminal={"result_version_persisted": True}, branches=["read_and_save"])
|
|
for meta in ({}, {"terminal_complete": True}, {"terminal_complete": True, "executed_branch": "guessed"}):
|
|
assert event(current=pipeline, **meta)["outcome"] == "unknown"
|
|
complete = event(current=pipeline, terminal_complete=True, executed_branch="read_and_save")
|
|
assert complete["outcome"] == "success"
|
|
dependent = method(kind="pipeline", terminal={"persisted": True}, branches=["read_and_save"], dependencies=["method_dependency"])
|
|
assert event(current=dependent, terminal_complete=True, executed_branch="read_and_save")["outcome"] == "unknown"
|
|
|
|
|
|
def test_pipeline_proof_is_invalidated_if_component_event_is_later_corrected():
|
|
component = method()
|
|
component_good = event(current=component)
|
|
pipeline = method(key="read_pipeline", kind="pipeline", terminal={"persisted": True},
|
|
branches=["read_and_save"], dependencies=[component["id"]])
|
|
pipeline_good = event(current=pipeline, terminal_complete=True, executed_branch="read_and_save",
|
|
dependency_event_ids=[component_good["id"]])
|
|
context = {**CONTEXT, "available_method_ids": [component["id"]]}
|
|
rows = [component_good, pipeline_good]
|
|
assert evaluate(rows, pipeline, context)["stats"]["success"] == 1
|
|
corrected = event(current=component, outcome="contradiction", supersedes=component_good["id"], event_key="correction")
|
|
result = evaluate([*rows, corrected], pipeline, context)
|
|
assert result["stats"]["success"] == 0 and result["stats"]["unknown"] == 1
|
|
|
|
|
|
def test_confidence_and_old_proofs_survive_input_immutable_recalculation():
|
|
rows = [event(f"t{i}", f"o{i}") for i in range(3)]
|
|
before = deepcopy(rows)
|
|
result = evaluate(rows)
|
|
assert rows == before
|
|
result["evidence_refs"][0]["pre"] = "modified-copy"
|
|
result["steps"].append("modified-view")
|
|
assert rows == before
|
|
assert "modified-view" not in evaluate(rows)["steps"]
|
|
|
|
|
|
def test_business_strategy_cannot_accumulate_technical_success_from_feedback():
|
|
strategy = method(kind="strategy", key="business_strategy")
|
|
assert event(current=strategy)["outcome"] == "unknown"
|
|
assert evaluate([event(current=strategy)], strategy)["stats"]["success"] == 0
|
|
|
|
|
|
def test_rank_is_tier_then_independent_sample_not_raw_one_shot_success_rate():
|
|
one, stable = method(key="one"), method(key="stable")
|
|
rows = [event(current=one)] + [event(f"t{i}", f"o{i}", current=stable) for i in range(10)]
|
|
ranked = knowledge.rank_methods([one, stable], rows, CONTEXT)
|
|
assert [result["key"] for result in ranked] == ["stable", "one"]
|
|
|
|
|
|
def test_shared_specs_and_proofs_reject_personal_source_material():
|
|
with pytest.raises(ValueError, match="personal/source"):
|
|
method(steps=[{"quote": "a person's actual comment"}])
|
|
with pytest.raises(ValueError, match="parameters"):
|
|
method(steps=["open https://www.xiaohongshu.com/user/profile/private?xsec_token=secret"])
|
|
with pytest.raises(ValueError, match="artifacts"):
|
|
knowledge.prepare_event(method(), "task", "o", "success", {"pre": "https://example.test/private"})
|
|
proof = event(obj="personally identifying profile")
|
|
assert "personally identifying" not in str(proof)
|
|
|
|
|
|
@pytest.fixture
|
|
def knowledge_workspace(monkeypatch):
|
|
workspace = f"pytest_knowledge_{uuid4().hex[:12]}"
|
|
monkeypatch.setattr(settings, "school_id", workspace)
|
|
yield workspace
|
|
with db.get_conn() as conn:
|
|
for table in ("prospecting_validation_events", "prospecting_business_feedback", "prospecting_methods"):
|
|
db._execute(conn, f"DELETE FROM {table} WHERE workspace_id=%s", (workspace,))
|
|
db._execute(conn, "DELETE FROM learned_entries WHERE school_id=%s", (workspace,))
|
|
|
|
|
|
@pytest.mark.db
|
|
def test_registry_append_ledger_idempotency_restarts_and_supersession(knowledge_workspace):
|
|
registered = knowledge.register_method({key: value for key, value in method().items() if key != "id"}, "t1")
|
|
assert knowledge.register_method({key: value for key, value in method().items() if key != "id"}, "t2")["id"] == registered["id"]
|
|
with pytest.raises(ValueError, match="new version"):
|
|
knowledge.register_method({**method(), "steps": ["changed implementation"]})
|
|
args = dict(method_id=registered["id"], task_id="t1", object_key="o1", outcome="success",
|
|
evidence_refs={"pre": "t1:proof:before", "post": "t1:proof:after"}, context=CONTEXT,
|
|
verified=True, verifier="fixture", event_key="proof-one")
|
|
first = knowledge.record_event(**args)
|
|
assert knowledge.record_event(**args) == first
|
|
assert knowledge.select_methods(CONTEXT)[0]["stats"]["success"] == 1
|
|
with pytest.raises(ValueError, match="different evidence"):
|
|
knowledge.record_event(**{**args, "outcome": "contradiction"})
|
|
correction = knowledge.record_event(**{**args, "outcome": "contradiction", "event_key": "corrected", "supersedes": first["id"]})
|
|
summary = knowledge.task_summary("t1")
|
|
assert len(summary["events"]) == 2 and summary["effective_event_ids"] == [correction["id"]]
|
|
assert summary["methods"][0]["status"] == "paused" and knowledge.select_methods(CONTEXT) == []
|
|
|
|
|
|
@pytest.mark.db
|
|
def test_workspace_boundary_and_historical_rows_never_enter_ranked_recall(knowledge_workspace, monkeypatch):
|
|
db.add_learned_drafts(KNOWLEDGE_ID := "xiaohongshu_prospecting", "prospecting",
|
|
[{"key": "old_wrong_mapping", "value": "old unverified prose"}], "old", auto_promote=False)
|
|
before = db.list_learned(KNOWLEDGE_ID, "prospecting")
|
|
current = knowledge.register_method(method(), "new")
|
|
rows = knowledge.list_methods(CONTEXT)
|
|
assert any(row["status"] == "legacy" and row["stats"]["success"] is None for row in rows)
|
|
assert all(row["status"] != "legacy" for row in knowledge.select_methods(CONTEXT))
|
|
assert db.list_learned(KNOWLEDGE_ID, "prospecting") == before
|
|
monkeypatch.setattr(settings, "school_id", "pytest_empty_workspace")
|
|
assert knowledge.list_methods(CONTEXT) == []
|
|
with pytest.raises(ValueError, match="workspace"):
|
|
knowledge.record_event(current["id"], "task", "object", "unknown")
|
|
monkeypatch.setattr(settings, "school_id", knowledge_workspace)
|
|
|
|
|
|
@pytest.mark.db
|
|
def test_business_feedback_is_append_only_versioned_and_does_not_change_technical_stats(knowledge_workspace):
|
|
current = knowledge.register_method(method(), "task")
|
|
before = knowledge.list_methods(CONTEXT, include_legacy=False)
|
|
first = knowledge.record_business_feedback("task", "results_v1", {"rating": "up"}, goal_version=1)
|
|
assert knowledge.record_business_feedback("task", "results_v1", {"rating": "up"}, goal_version=1) == first
|
|
knowledge.record_business_feedback("task", "results_v2", {"rating": "down", "actual_outcome": "no_reply"}, goal_version=2)
|
|
assert knowledge.list_methods(CONTEXT, include_legacy=False) == before
|
|
feedback = knowledge.task_summary("task")["business_feedback"]
|
|
assert len(feedback) == 2 and {row["result_version"] for row in feedback} == {"results_v1", "results_v2"}
|
|
assert all(row["attribution"] == "unproven" for row in feedback)
|
|
|
|
|
|
@pytest.mark.db
|
|
def test_pipeline_receipt_replay_returns_original_after_dependency_correction(knowledge_workspace):
|
|
component = knowledge.register_method(method(), "task")
|
|
args = dict(method_id=component["id"], task_id="task", object_key="object", outcome="success",
|
|
evidence_refs={"pre": "task:proof:before", "post": "task:proof:after"},
|
|
context=CONTEXT, verified=True, verifier="fixture", event_key="component-proof")
|
|
first = knowledge.record_event(**args)
|
|
pipeline = knowledge.register_method(method(key="pipeline", kind="pipeline", terminal={"persisted": True},
|
|
branches=["read_and_save"], dependencies=[component["id"]]), "task")
|
|
pipeline_args = {**args, "method_id": pipeline["id"], "event_key": "pipeline-proof", "terminal_complete": True,
|
|
"executed_branch": "read_and_save", "dependency_event_ids": [first["id"]]}
|
|
original = knowledge.record_event(**pipeline_args)
|
|
assert original["outcome"] == "success"
|
|
knowledge.record_event(**{**args, "outcome": "contradiction", "event_key": "component-correction", "supersedes": first["id"]})
|
|
assert knowledge.record_event(**pipeline_args) == original
|
|
ranked = knowledge.list_methods({**CONTEXT, "available_method_ids": [component["id"]]}, include_legacy=False)
|
|
current = next(row for row in ranked if row["id"] == pipeline["id"])
|
|
assert current["stats"]["success"] == 0 and current["stats"]["unknown"] == 1
|