fe99b6bbf2
Export reads the extraction JSON instead of a tag fragment and writes the claim. A logged search stores its project so the nightly check can clear. Claims keep the effect and the subject, a later strand tags an existing DOI, the dedup command prints borderline pairs, and due dates compare the calendar day.
235 lines
7.8 KiB
Python
235 lines
7.8 KiB
Python
"""Zotero review rows export into the append-only ledger."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from typer.testing import CliRunner
|
|
|
|
from cli import app
|
|
from lead.ledger.export import export_project
|
|
from lead.ledger.store import open_ledger
|
|
from prisma.ingest import apply_extraction
|
|
from zot.db import TYPE_MAP, Db
|
|
from zot.schema import create_db
|
|
|
|
runner = CliRunner()
|
|
|
|
|
|
def test_export_is_idempotent_until_the_extraction_version_changes(tmp_path):
|
|
zpath = str(tmp_path / "z.sqlite")
|
|
con = create_db(zpath)
|
|
con.close()
|
|
with Db(zpath) as zdb:
|
|
item_id = zdb.create_item(TYPE_MAP["journalArticle"], key="ABCD2345")
|
|
zdb.set_fields(item_id, {"title": "All-cause readmission"})
|
|
for tag in (
|
|
"project:lead-precursors",
|
|
"found-in:pubmed",
|
|
"stage:4-included",
|
|
"screen:include",
|
|
):
|
|
zdb.tag_item(item_id, tag)
|
|
zdb.add_note(item_id, "target_id: Q.ACR\n\nIncluded.")
|
|
anchor = zdb.create_item(TYPE_MAP["journalArticle"], key="EFGH5678")
|
|
zdb.tag_item(anchor, "project:lead-precursors")
|
|
zdb.tag_item(anchor, "module:prisma")
|
|
zdb.tag_item(anchor, "stage:4-included")
|
|
zdb.add_note(anchor, "target_id: Q.ACR")
|
|
zdb.commit()
|
|
|
|
ledger = open_ledger(str(tmp_path / "ledger.duckdb"))
|
|
try:
|
|
first = export_project(
|
|
zdb,
|
|
ledger,
|
|
"lead-precursors",
|
|
extraction_version="v1",
|
|
actor="tester",
|
|
)
|
|
assert first == {"source": 1, "decision": 1, "extraction": 1}
|
|
second = export_project(
|
|
zdb,
|
|
ledger,
|
|
"lead-precursors",
|
|
extraction_version="v1",
|
|
actor="tester",
|
|
)
|
|
assert second == {"source": 0, "decision": 0, "extraction": 0}
|
|
|
|
source_hash = ledger.execute(
|
|
"select content_hash from lead_evidence.source"
|
|
).fetchone()[0]
|
|
stored_key = ledger.execute(
|
|
"select zotero_key from lead_evidence.source"
|
|
).fetchone()[0]
|
|
assert stored_key == "ABCD2345"
|
|
third = export_project(
|
|
zdb,
|
|
ledger,
|
|
"lead-precursors",
|
|
extraction_version="v2",
|
|
actor="tester",
|
|
)
|
|
assert third["extraction"] == 1
|
|
hashes = {
|
|
row[0]
|
|
for row in ledger.execute(
|
|
"select content_hash from lead_evidence.source"
|
|
).fetchall()
|
|
}
|
|
assert hashes != {source_hash}
|
|
versions = {
|
|
row[0]
|
|
for row in ledger.execute(
|
|
"select extraction_version from lead_evidence.extraction"
|
|
).fetchall()
|
|
}
|
|
assert versions == {"v1", "v2"}
|
|
targets = {
|
|
row[0]
|
|
for row in ledger.execute(
|
|
"select target_id from lead_evidence.extraction"
|
|
).fetchall()
|
|
}
|
|
assert targets == {"Q.ACR"}
|
|
finally:
|
|
ledger.close()
|
|
|
|
|
|
def test_ledger_export_command_is_registered():
|
|
result = runner.invoke(app, ["lead", "ledger", "export", "--help"])
|
|
assert result.exit_code == 0
|
|
|
|
|
|
def test_html_extraction_note_stores_the_target_and_a_claim(tmp_path):
|
|
zpath = str(tmp_path / "z.sqlite")
|
|
con = create_db(zpath)
|
|
con.close()
|
|
with Db(zpath) as zdb:
|
|
item_id = zdb.create_item(TYPE_MAP["journalArticle"], key="ABCD2345")
|
|
zdb.set_fields(item_id, {"title": "All-cause readmission"})
|
|
zdb.tag_item(item_id, "stage:4-included")
|
|
apply_extraction(
|
|
zdb,
|
|
item_id,
|
|
"lead-precursors",
|
|
{
|
|
"outcome_target_id": "Q.ACR",
|
|
"effect_measure": "OR",
|
|
"effect_value": "1.2",
|
|
"effect_ci": "1.1, 1.4",
|
|
"factor_definition": "frailty",
|
|
"design": "cohort",
|
|
},
|
|
)
|
|
zdb.commit()
|
|
ledger = open_ledger(str(tmp_path / "ledger.duckdb"))
|
|
try:
|
|
export_project(
|
|
zdb,
|
|
ledger,
|
|
"lead-precursors",
|
|
extraction_version="v1",
|
|
actor="tester",
|
|
)
|
|
target = ledger.execute(
|
|
"select target_id from lead_evidence.extraction"
|
|
).fetchone()[0]
|
|
assert target == "Q.ACR"
|
|
row = ledger.execute(
|
|
"""
|
|
select effect_value, effect_ci_low, effect_ci_high, subject
|
|
from lead_evidence.claim
|
|
"""
|
|
).fetchone()
|
|
assert row == (1.2, 1.1, 1.4, "frailty")
|
|
finally:
|
|
ledger.close()
|
|
|
|
|
|
def test_html_target_label_is_not_stored_as_a_tag(tmp_path):
|
|
zpath = str(tmp_path / "z.sqlite")
|
|
con = create_db(zpath)
|
|
con.close()
|
|
with Db(zpath) as zdb:
|
|
item_id = zdb.create_item(TYPE_MAP["journalArticle"], key="ABCD2345")
|
|
zdb.tag_item(item_id, "project:lead-precursors")
|
|
zdb.tag_item(item_id, "stage:4-included")
|
|
zdb.add_note(item_id, "<li><strong>outcome_target_id:</strong> Q.ACR</li>")
|
|
zdb.commit()
|
|
ledger = open_ledger(str(tmp_path / "ledger.duckdb"))
|
|
try:
|
|
export_project(
|
|
zdb,
|
|
ledger,
|
|
"lead-precursors",
|
|
extraction_version="v1",
|
|
actor="tester",
|
|
)
|
|
target = ledger.execute(
|
|
"select target_id from lead_evidence.extraction"
|
|
).fetchone()[0]
|
|
assert target == "Q.ACR"
|
|
finally:
|
|
ledger.close()
|
|
|
|
|
|
def test_unknown_target_in_a_note_is_not_stored(tmp_path):
|
|
zpath = str(tmp_path / "z.sqlite")
|
|
con = create_db(zpath)
|
|
con.close()
|
|
with Db(zpath) as zdb:
|
|
item_id = zdb.create_item(TYPE_MAP["journalArticle"], key="ABCD2345")
|
|
zdb.tag_item(item_id, "project:lead-limits")
|
|
zdb.tag_item(item_id, "stage:4-included")
|
|
zdb.add_note(item_id, "target_id: nope")
|
|
zdb.commit()
|
|
ledger = open_ledger(str(tmp_path / "ledger.duckdb"))
|
|
try:
|
|
with pytest.raises(KeyError):
|
|
export_project(
|
|
zdb,
|
|
ledger,
|
|
"lead-limits",
|
|
extraction_version="v1",
|
|
actor="tester",
|
|
)
|
|
count = ledger.execute(
|
|
"select count(*) from lead_evidence.extraction"
|
|
).fetchone()[0]
|
|
assert count == 0
|
|
finally:
|
|
ledger.close()
|
|
|
|
|
|
def test_extraction_without_a_target_stores_the_row_and_skips_the_claim(tmp_path):
|
|
zpath = str(tmp_path / "z.sqlite")
|
|
con = create_db(zpath)
|
|
con.close()
|
|
with Db(zpath) as zdb:
|
|
item_id = zdb.create_item(TYPE_MAP["journalArticle"], key="ABCD2345")
|
|
zdb.tag_item(item_id, "project:lead-limits")
|
|
zdb.tag_item(item_id, "stage:4-included")
|
|
zdb.add_note(item_id, "Included with no target.")
|
|
zdb.commit()
|
|
ledger = open_ledger(str(tmp_path / "ledger.duckdb"))
|
|
try:
|
|
counts = export_project(
|
|
zdb,
|
|
ledger,
|
|
"lead-limits",
|
|
extraction_version="v1",
|
|
actor="tester",
|
|
)
|
|
assert counts["extraction"] == 1
|
|
target = ledger.execute(
|
|
"select target_id from lead_evidence.extraction"
|
|
).fetchone()[0]
|
|
assert target is None
|
|
claims = ledger.execute(
|
|
"select count(*) from lead_evidence.claim"
|
|
).fetchone()[0]
|
|
assert claims == 0
|
|
finally:
|
|
ledger.close()
|