108 lines
6.1 KiB
Python
108 lines
6.1 KiB
Python
"""Operation identity and learning recovery, without real account or model use."""
|
|
from copy import deepcopy
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from backend.db import database as db
|
|
from backend.prospecting import orchestrator as module
|
|
from backend.prospecting.workflows import finalize_review
|
|
|
|
KID, BID = "xiaohongshu_prospecting", "prospecting"
|
|
|
|
|
|
def reviewed():
|
|
return finalize_review("learning-fixture", {"sources": [
|
|
{"source_id": kind, "source_type": kind, "source_selector": "#" + kind,
|
|
"observed_at": "2026-10-29T10:00:00+08:00"}
|
|
for kind in ("note", "comment", "reply")
|
|
]})
|
|
|
|
|
|
def test_operation_identity_separates_source_kind_capability_version_and_old_namespace():
|
|
drafts = reviewed()["learning_candidates"]
|
|
# All of these descriptions matched the former fuzzy threshold despite
|
|
# referring to distinct source containers and attribution boundaries.
|
|
assert db._same_knowledge(drafts[0]["key"], drafts[0]["value"], drafts[1]["key"], drafts[1]["value"])
|
|
value = drafts[0]["value"]
|
|
original = drafts[0]["key"]
|
|
for different in [drafts[1]["key"], drafts[2]["key"], original.replace("v1", "v2"),
|
|
original.replace("save_source", "inspect_profile"), "xhs_visible_note_v1", "unstructured_alias"]:
|
|
assert not db._same_scoped_knowledge(KID, BID, original, value, different, value)
|
|
assert db._same_scoped_knowledge(KID, BID, original, value, original, value)
|
|
assert db._same_scoped_knowledge("school_pack", BID, original, value, "model_alias", value)
|
|
|
|
|
|
def test_review_reserved_contract_is_generated_from_evidence_and_is_idempotent():
|
|
first = reviewed()
|
|
first["learning_candidates"][0]["value"] = "incorrect prose under a reserved operation key"
|
|
again = finalize_review("learning-fixture", first)
|
|
assert len(again["learning_candidates"]) == 3
|
|
assert "笔记摘录" in again["learning_candidates"][0]["value"]
|
|
assert finalize_review("learning-fixture", again)["learning_candidates"] == again["learning_candidates"]
|
|
|
|
|
|
@pytest.mark.db
|
|
def test_typed_operation_drafts_survive_repetition_and_consolidation_without_merging_kinds(test_school):
|
|
entries = reviewed()["learning_candidates"]
|
|
first = db.add_learned_drafts(KID, BID, entries, "task-one", auto_promote=False)
|
|
repeated = db.add_learned_drafts(KID, BID, entries, "task-one", auto_promote=False)
|
|
assert first == {"written": 3, "merged": 0, "promoted": 0, "skipped": 0}
|
|
assert repeated == {"written": 0, "merged": 3, "promoted": 0, "skipped": 0}
|
|
rows = db.list_learned(KID, BID)
|
|
assert len(rows) == 3 and {row["observations"] for row in rows} == {1}
|
|
assert {row["status"] for row in rows} == {"draft"}
|
|
consolidated = db.consolidate_learned(KID, BID)
|
|
assert consolidated == {"clusters": 0, "removed": 0, "promoted": 0}
|
|
assert len(db.list_learned(KID, BID)) == 3
|
|
|
|
|
|
@pytest.mark.db
|
|
def test_new_note_contract_never_inherits_legacy_note_key_comment_value(test_school):
|
|
entries = reviewed()["learning_candidates"]
|
|
legacy = {"key": "xhs_visible_note_v1", "category": "control_tip", "value": entries[1]["value"]}
|
|
db.add_learned_drafts(KID, BID, [legacy], "legacy-task", auto_promote=False)
|
|
before = db.list_learned(KID, BID)[0]
|
|
stats = db.add_learned_drafts(KID, BID, entries, "future-task", auto_promote=False)
|
|
after = db.list_learned(KID, BID)
|
|
assert stats["written"] == 3 and stats["merged"] == 0
|
|
assert len(after) == 4
|
|
assert next(row for row in after if row["id"] == before["id"]) == before
|
|
assert next(row for row in after if row["key"] == entries[0]["key"])["value"] == entries[0]["value"]
|
|
|
|
|
|
def recover(monkeypatch, tmp_path, snapshot, artifacts):
|
|
directory = tmp_path / "learning-fixture"
|
|
directory.mkdir(exist_ok=True)
|
|
(directory / "evidence.json").write_text(json.dumps(snapshot), encoding="utf-8")
|
|
monkeypatch.setattr(module, "task_doc_dir", lambda _tid: directory)
|
|
return module.ProspectingOrchestrator._recover_evidence("learning-fixture", artifacts)
|
|
|
|
|
|
def test_old_empty_checkpoint_does_not_erase_review_candidates_or_outcomes(monkeypatch, tmp_path):
|
|
artifacts = {"learning_candidates": reviewed()["learning_candidates"],
|
|
"learning_outcomes": [{"entry_id": 7, "outcome": "confirmed", "evidence": "local fixture proof"}]}
|
|
before = deepcopy(artifacts)
|
|
result = recover(monkeypatch, tmp_path, {"learning_candidates": [], "learning_outcomes": []}, artifacts)
|
|
assert result == before and artifacts == before
|
|
|
|
|
|
def test_learning_recovery_unions_both_sides_and_retains_conflict_evidence(monkeypatch, tmp_path):
|
|
snapshot = {"learning_candidates": [{"key": "same", "value": "old", "source_url": "https://fixture.test/old"}, {"key": "checkpoint-only", "value": "page-only"}],
|
|
"learning_outcomes": [{"entry_id": 7, "outcome": "unused", "note": "old observation"}, {"entry_id": 8, "outcome": "confirmed", "evidence": "new snapshot proof"}]}
|
|
artifacts = {"learning_candidates": [{"key": "same", "value": "reviewed", "category": "control_tip"}, {"key": "database-only", "value": "review-only"}],
|
|
"learning_outcomes": [{"entry_id": 7, "outcome": "contradicted", "note": "persisted decision", "evidence": "persisted proof"}]}
|
|
before = deepcopy(artifacts)
|
|
result = recover(monkeypatch, tmp_path, snapshot, artifacts)
|
|
assert {row["key"] for row in result["learning_candidates"]} == {"same", "checkpoint-only", "database-only"}
|
|
assert next(row for row in result["learning_candidates"] if row["key"] == "same")["value"] == "reviewed"
|
|
assert {row["entry_id"] for row in result["learning_outcomes"]} == {7, 8}
|
|
assert next(row for row in result["learning_outcomes"] if row["entry_id"] == 7)["outcome"] == "contradicted"
|
|
conflicts = result["learning_recovery_conflicts"]
|
|
assert len(conflicts) == 2
|
|
assert {row["kept_origin"] for row in conflicts} == {"database"}
|
|
assert any(row["other"].get("source_url") == "https://fixture.test/old" for row in conflicts)
|
|
assert any(row["other"].get("note") == "old observation" for row in conflicts)
|
|
assert artifacts == before
|
|
assert recover(monkeypatch, tmp_path, snapshot, result) == result
|