stack/tests/prisma/test_ingest_exercise.py

"""Exercise prisma.ingest — apply_screen/eligibility/extraction decisions."""

from __future__ import annotations

from prisma.ingest import (
    _html_escape,
    _replace_stage,
    _slug,
    apply_eligibility_decision,
    apply_extraction,
    apply_screen_decision,
)
from zot.db import Db
from zot.schema import create_db


def _setup_db(tmp_path):
    path = str(tmp_path / "z.sqlite")
    con = create_db(path)
    con.close()
    return path


def _insert_item(db, title="Test"):
    db.con.execute(
        "INSERT INTO items (itemTypeID, libraryID, key, dateAdded, dateModified, clientDateModified) "
        "VALUES (2, 1, 'TESTKEY1', '', '', '')"
    )
    return db.con.execute("SELECT last_insert_rowid()").fetchone()[0]


class TestHtmlEscape:
    def test_escapes(self):
        assert _html_escape("<b>test</b>") == "&lt;b&gt;test&lt;/b&gt;"
        assert _html_escape("a & b") == "a &amp; b"
        assert _html_escape("") == ""
        assert _html_escape(None) == ""


class TestSlug:
    def test_basic(self):
        assert _slug("Hello World!") == "hello-world"
        assert _slug("  multiple   spaces  ") == "multiple-spaces"
        assert _slug("") == ""


class TestReplaceStage:
    def test_replaces(self, tmp_path):
        path = _setup_db(tmp_path)
        with Db(path) as db:
            iid = _insert_item(db)
            db.sync_tags(iid, ["stage:1-identified", "other:tag"])
            _replace_stage(db, iid, "stage:2-title-abstract")
            tags = [
                r[0]
                for r in db.con.execute(
                    "SELECT t.name FROM itemTags it JOIN tags t ON it.tagID=t.tagID WHERE it.itemID=?",
                    (iid,),
                ).fetchall()
            ]
            assert "stage:2-title-abstract" in tags
            assert "stage:1-identified" not in tags
            assert "other:tag" in tags


class TestApplyScreenDecision:
    def test_include(self, tmp_path):
        path = _setup_db(tmp_path)
        with Db(path) as db:
            iid = _insert_item(db)
            payload = {
                "decision": "include",
                "reasons": ["relevant"],
                "themes": ["Billing Compliance"],
                "rationale": "Directly relevant to the review topic.",
            }
            apply_screen_decision(db, iid, "test-proj", payload)
            tags = [
                r[0]
                for r in db.con.execute(
                    "SELECT t.name FROM itemTags it JOIN tags t ON it.tagID=t.tagID WHERE it.itemID=?",
                    (iid,),
                ).fetchall()
            ]
            assert "screen:include" in tags
            assert "screen:reason:relevant" in tags
            assert "theme:billing-compliance" in tags
            assert "stage:2-title-abstract" in tags

    def test_exclude(self, tmp_path):
        path = _setup_db(tmp_path)
        with Db(path) as db:
            iid = _insert_item(db)
            payload = {
                "decision": "exclude",
                "reasons": ["irrelevant"],
                "themes": [],
                "rationale": "Not relevant.",
            }
            apply_screen_decision(db, iid, "test-proj", payload)
            tags = [
                r[0]
                for r in db.con.execute(
                    "SELECT t.name FROM itemTags it JOIN tags t ON it.tagID=t.tagID WHERE it.itemID=?",
                    (iid,),
                ).fetchall()
            ]
            assert "screen:exclude" in tags


class TestApplyEligibilityDecision:
    def test_include_advances_to_stage4(self, tmp_path):
        path = _setup_db(tmp_path)
        with Db(path) as db:
            iid = _insert_item(db)
            payload = {
                "decision": "include",
                "reasons": [],
                "themes": [],
                "rationale": "Full text confirms relevance.",
            }
            apply_eligibility_decision(db, iid, "test-proj", payload)
            tags = [
                r[0]
                for r in db.con.execute(
                    "SELECT t.name FROM itemTags it JOIN tags t ON it.tagID=t.tagID WHERE it.itemID=?",
                    (iid,),
                ).fetchall()
            ]
            assert "stage:4-included" in tags

    def test_exclude_stays_stage3(self, tmp_path):
        path = _setup_db(tmp_path)
        with Db(path) as db:
            iid = _insert_item(db)
            payload = {
                "decision": "exclude",
                "reasons": ["no_fulltext"],
                "themes": [],
                "rationale": "Cannot access full text.",
            }
            apply_eligibility_decision(db, iid, "test-proj", payload)
            tags = [
                r[0]
                for r in db.con.execute(
                    "SELECT t.name FROM itemTags it JOIN tags t ON it.tagID=t.tagID WHERE it.itemID=?",
                    (iid,),
                ).fetchall()
            ]
            assert "stage:3-full-text" in tags


class TestApplyExtraction:
    def test_writes_note(self, tmp_path):
        path = _setup_db(tmp_path)
        with Db(path) as db:
            iid = _insert_item(db)
            payload = {
                "population": "Medicare beneficiaries",
                "intervention": "skin substitutes",
                "outcome": "healing rate",
                "extraction_notes": "Data from Table 2.",
            }
            apply_extraction(db, iid, "test-proj", payload)
            tags = [
                r[0]
                for r in db.con.execute(
                    "SELECT t.name FROM itemTags it JOIN tags t ON it.tagID=t.tagID WHERE it.itemID=?",
                    (iid,),
                ).fetchall()
            ]
            assert "prisma:extracted" in tags