Chaim
dfb2ffe7ce
WS2b של עיצוב-מחדש זרימת-העבודה. מאפשר ניתוח כתב-הערר לבד ואז איחוד מסמך-עיקרי שנוסף מאוחר, בלי force-delete גורף, עם diff ליו"ר. - SCHEMA_V49: documents.claims_extracted_at + claims_extraction_status (אירוע-חילוץ per-מסמך, לא נגזר מ-doc_type), + אינדקס חלקי idx_documents_claims_pending. idempotent (ADD COLUMN IF NOT EXISTS). V48 שמור ל-#357. - claims_extractor חותם את המסמך אחרי שמירת/אי-מציאת טענות. - db.primary_docs_not_analyzed (is_primary V47 + claims_extracted_at IS NULL) + mark_document_claims_extracted; _row_to_doc חושף claims_analyzed. - reanalyze_claims (כלי-MCP): snapshot→חילוץ-מאחד רק למסמכים חדשים/לא-נותחו (store_claims מחליף per-source ⇒ טענות אחרות נשמרות)→aggregate force=True (מסלול קנוני, מוחק רק legal_arguments)→_impact_diff before↔after ליו"ר. - workflow_status חושף primary_docs_not_analyzed + next-step. - ספ: 02-data-model §2ג (דגל לא-נותח), 04-analysis-writing §1.3. - 5 בדיקות-יחידה ל-_impact_diff/_snapshot (פונקציות טהורות). Invariants: - G1 (נרמול-במקור): claims_extracted_at נחתם בנקודת-החילוץ, claims_analyzed נגזר ממנו — לא תיקון-בקריאה. - G2 (מסלול קנוני יחיד / merge-not-fork): reanalyze מרחיב את מסלול claims_extractor+argument_aggregator הקיים, לא forks; האיחוד דרך store_claims per-source; aggregate force=True מוחק רק נגזר (legal_arguments). - INV-DM (מודל-נתונים): דגל-אירוע per-מסמך, אינדקס חלקי, מקור-אמת יחיד. - G10 (שער-אנושי): בדיקת-ההשפעה מוצגת ליו"ר, לא מוחלת אוטומטית. - INV-TOOL idempotency: SCHEMA_V49 idempotent; ניתוח-מחדש חוזר על מסמך כבר-נותח הוא refresh נקי (store_methods replace per-source). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
77 lines
3.3 KiB
Python
77 lines
3.3 KiB
Python
"""Unit tests for the re-analysis impact diff — WS2 / task #201.
|
|
|
|
`_snapshot` + `_impact_diff` are pure functions (no DB) that produce the
|
|
before↔after summary surfaced to the chair after a merging re-analysis. They
|
|
must report per-party argument titles added/removed and the shift in the
|
|
priority ("recommendation balance") mix, and set ``changed`` correctly.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from legal_mcp.tools.legal_arguments import _impact_diff, _snapshot
|
|
|
|
|
|
def _arg(party, title, priority="substantive"):
|
|
return {"party": party, "argument_title": title, "priority": priority}
|
|
|
|
|
|
def test_no_change_is_flagged_false():
|
|
args = [_arg("appellant", "זכות עמידה"), _arg("committee", "שיהוי")]
|
|
snap = _snapshot(args)
|
|
diff = _impact_diff(snap, snap)
|
|
assert diff["changed"] is False
|
|
assert diff["total_before"] == diff["total_after"] == 2
|
|
# Every party entry reports changed=False.
|
|
assert all(p["changed"] is False for p in diff["by_party"])
|
|
|
|
|
|
def test_added_argument_detected():
|
|
before = _snapshot([_arg("appellant", "זכות עמידה")])
|
|
after = _snapshot([_arg("appellant", "זכות עמידה"), _arg("appellant", "תחולת תמא 38")])
|
|
diff = _impact_diff(before, after)
|
|
assert diff["changed"] is True
|
|
assert diff["total_before"] == 1
|
|
assert diff["total_after"] == 2
|
|
appellant = next(p for p in diff["by_party"] if p["party"] == "appellant")
|
|
assert appellant["added_arguments"] == ["תחולת תמא 38"]
|
|
assert appellant["removed_arguments"] == []
|
|
assert appellant["party_he"] == "עוררים"
|
|
|
|
|
|
def test_removed_argument_detected():
|
|
before = _snapshot([_arg("committee", "שיהוי"), _arg("committee", "סמכות")])
|
|
after = _snapshot([_arg("committee", "שיהוי")])
|
|
diff = _impact_diff(before, after)
|
|
committee = next(p for p in diff["by_party"] if p["party"] == "committee")
|
|
assert committee["removed_arguments"] == ["סמכות"]
|
|
assert committee["count_before"] == 2
|
|
assert committee["count_after"] == 1
|
|
assert diff["changed"] is True
|
|
|
|
|
|
def test_priority_mix_shift_is_a_change():
|
|
# Same title, but priority moved threshold -> substantive: the recommendation
|
|
# balance shifted, so the chair must see a change even with identical titles.
|
|
before = _snapshot([_arg("appellant", "סמכות", priority="threshold")])
|
|
after = _snapshot([_arg("appellant", "סמכות", priority="substantive")])
|
|
diff = _impact_diff(before, after)
|
|
appellant = next(p for p in diff["by_party"] if p["party"] == "appellant")
|
|
assert appellant["added_arguments"] == []
|
|
assert appellant["removed_arguments"] == []
|
|
assert appellant["priority_before"] == {"threshold": 1}
|
|
assert appellant["priority_after"] == {"substantive": 1}
|
|
assert appellant["changed"] is True
|
|
assert diff["changed"] is True
|
|
|
|
|
|
def test_new_party_appears():
|
|
before = _snapshot([_arg("appellant", "זכות עמידה")])
|
|
after = _snapshot([_arg("appellant", "זכות עמידה"), _arg("respondent", "מענה")])
|
|
diff = _impact_diff(before, after)
|
|
parties = {p["party"] for p in diff["by_party"]}
|
|
assert parties == {"appellant", "respondent"}
|
|
respondent = next(p for p in diff["by_party"] if p["party"] == "respondent")
|
|
assert respondent["count_before"] == 0
|
|
assert respondent["count_after"] == 1
|
|
assert respondent["added_arguments"] == ["מענה"]
|