stack/tests/pfs/test_descriptors.py

"""pfs.descriptors — where descriptor text comes from."""

from __future__ import annotations

import sqlite3

import duckdb
import pytest

from pfs.descriptors import (
    DescriptorRun,
    descriptor_runs,
    hcpcs_long_description,
    rule_year_of,
    rvu_descriptions,
)


class _Store:
    """Minimal stand-in for bib.Store: only ``_con()`` is used."""

    def __init__(self) -> None:
        self.con = sqlite3.connect(":memory:")
        self.con.row_factory = sqlite3.Row
        self.con.executescript(
            """
            CREATE TABLE items (key TEXT PRIMARY KEY, title TEXT, date_published TEXT);
            CREATE TABLE fr_anchors (item_key TEXT, p_id INTEGER, page INTEGER, ordinal INTEGER, text TEXT);
            """
        )

    def _con(self):
        return self.con

    def close(self):
        self.con.close()


@pytest.fixture
def store():
    s = _Store()
    s.con.execute(
        "INSERT INTO items VALUES ('DE2VH9PD', 'Medicare Program; CY 2015 PFS Final Rule', '2014-11-13')"
    )
    rows = [
        (
            "DE2VH9PD",
            1243,
            67716,
            "Comment: commenters noted the CPT panel created a code.",
        ),
        (
            "DE2VH9PD",
            1244,
            67716,
            "These commenters suggested that we use the new CPT code 99490 (Chronic care management services, at least 20 minutes of clinical staff time directed by a physician or other qualified health care professional, per calendar month, with the following required elements:",
        ),
        (
            "DE2VH9PD",
            1245,
            67716,
            "Multiple (two or more) chronic conditions expected to last at least 12 months, or until the death of the patient;",
        ),
        (
            "DE2VH9PD",
            1246,
            67716,
            "Chronic conditions place the patient at significant risk of death, acute exacerbation/decompensation, or functional decline;",
        ),
        (
            "DE2VH9PD",
            1247,
            67716,
            "Comprehensive care plan established, implemented, revised, or monitored).",
        ),
        (
            "DE2VH9PD",
            1248,
            67716,
            "Many of these commenters expressed a preference for the per calendar month period.",
        ),
        ("DE2VH9PD", 1249, 67716, "Response: It is our preference to use CPT codes."),
    ]
    s.con.executemany(
        "INSERT INTO fr_anchors VALUES (?,?,?,?,?)",
        [(k, p, pg, p, t) for k, p, pg, t in rows],
    )
    yield s
    s.close()


@pytest.fixture
def con():
    c = duckdb.connect(":memory:")
    c.execute("CREATE SCHEMA terminology; CREATE SCHEMA pfs")
    c.execute(
        "CREATE TABLE terminology.hcpcs_level_2 (hcpcs VARCHAR, seqnum VARCHAR, recid VARCHAR, long_description VARCHAR, short_description VARCHAR)"
    )
    c.execute(
        "INSERT INTO terminology.hcpcs_level_2 VALUES ('G0556','10','3','Advanced primary care management services … per calendar month','Adv prim care mgmt lvl 1')"
    )
    c.execute(
        "INSERT INTO terminology.hcpcs_level_2 VALUES ('G0556','5','1','Lower seqnum description for G0556','Short desc')"
    )
    c.execute(
        "CREATE TABLE pfs.rvu (hcpcs VARCHAR, mod VARCHAR, description VARCHAR, status_code VARCHAR, year INTEGER)"
    )
    c.executemany(
        "INSERT INTO pfs.rvu VALUES (?,?,?,?,?)",
        [
            ("99490", None, "Chron care mgmt srvc 20 min", "A", 2015),
            ("99490", "26", "ignored modifier row", "A", 2015),
            ("99490", "", "Chrnc care mgmt staff 1st 20", "A", 2022),
        ],
    )
    yield c
    c.close()


class TestRuleYear:
    def test_from_title(self):
        assert (
            rule_year_of("Medicare Program; CY 2015 PFS Final Rule", "2014-11-13")
            == 2015
        )
        assert (
            rule_year_of(
                "Medicare and Medicaid Programs; CY 2027 Payment Policies", "2026-07-16"
            )
            == 2027
        )

    def test_fallback_to_date_plus_one(self):
        assert rule_year_of("Revisions to Payment Policies", "2017-07-21") == 2018


class TestDescriptorRuns:
    def test_stem_and_following_elements(self, store):
        runs = descriptor_runs(store, "99490")
        assert len(runs) == 1
        run = runs[0]
        assert isinstance(run, DescriptorRun)
        assert run.item_key == "DE2VH9PD" and run.rule_year == 2015
        assert run.stem.p_id == 1244
        assert [p.p_id for p in run.elements] == [1245, 1246, 1247]
        assert (
            "per calendar month" in run.text and "Comprehensive care plan" in run.text
        )

    def test_no_stem_no_run(self, store):
        assert descriptor_runs(store, "99491") == []

    def test_max_elements_cap(self, store):
        run = descriptor_runs(store, "99490", max_elements=2)[0]
        assert [p.p_id for p in run.elements] == [1245, 1246]

    def test_enumeration_list_item_is_not_a_descriptor_stem(self, store):
        # C1 live-corpus incident: "( 9) 99439 (code for non-complex
        # chronic care management)." reads like a stem ("99439 (") but is
        # an enumeration/cross-reference line, not the opening of 99439's
        # own descriptor.
        store.con.executemany(
            "INSERT INTO fr_anchors VALUES (?,?,?,?,?)",
            [
                (
                    "DE2VH9PD",
                    1300,
                    67800,
                    1300,
                    "( 9) 99439 (code for non-complex chronic care management).",
                ),
                (
                    "DE2VH9PD",
                    1301,
                    67800,
                    1301,
                    "( 10) 99457 and 99458 (codes for remote physiologic monitoring).",
                ),
            ],
        )
        assert descriptor_runs(store, "99439") == []

    def test_lowercase_stem_still_matches(self):
        # `_stem_pattern` stays case-insensitive: some rules print the
        # descriptor lowercase.
        s = _Store()
        s.con.execute(
            "INSERT INTO items VALUES ('AAAAAAAA', 'Medicare Program; CY 2021 PFS Final Rule', '2020-11-13')"
        )
        s.con.executemany(
            "INSERT INTO fr_anchors VALUES (?,?,?,?,?)",
            [
                (
                    "AAAAAAAA",
                    1,
                    100,
                    1,
                    "CPT code 99490 (chronic care management services, at least 20 "
                    "minutes of clinical staff time directed by a physician or other "
                    "qualified health care professional, per calendar month, with the "
                    "following required elements:",
                ),
                ("AAAAAAAA", 2, 100, 2, "Consent;"),
            ],
        )
        try:
            runs = descriptor_runs(s, "99490")
            assert len(runs) == 1
        finally:
            s.close()


class TestDuckDBSources:
    def test_long_description(self, con):
        assert hcpcs_long_description(con, "G0556").startswith("Advanced primary care")
        assert hcpcs_long_description(con, "99490") == ""

    def test_long_description_deterministic_seqnum(self, con):
        # With multiple rows for same hcpcs, ORDER BY seqnum DESC picks the highest
        result = hcpcs_long_description(con, "G0556")
        assert (
            result == "Advanced primary care management services … per calendar month"
        )

    def test_rvu_descriptions_base_rows_only(self, con):
        assert rvu_descriptions(con, "99490") == [
            (2015, "A", "Chron care mgmt srvc 20 min"),
            (2022, "A", "Chrnc care mgmt staff 1st 20"),
        ]


class TestDescriptorSpan:
    """#700: a long background paragraph quoting a descriptor inline must
    contribute only the quoted descriptor, not the whole paragraph."""

    def test_trims_to_the_codes_parenthetical(self):
        from pfs.descriptors import descriptor_span

        text = (
            "In the CY 2017 PFS final rule we created HCPCS code G0502 (Initial "
            "psychiatric collaborative care management, first 70 minutes in the "
            "first calendar month of behavioral health care manager activities "
            "(in consultation with a psychiatric consultant)). "
            + "Background prose about chronic care management services, comprehensive "
            "care plans and multiple chronic conditions expected to last 12 months. "
            * 60
        )
        span = descriptor_span(text, "G0502")
        assert span.startswith("G0502 (Initial psychiatric")
        assert span.endswith("consultant))")
        assert "chronic care management" not in span
        assert len(span) < 300

    def test_no_close_paren_cuts_at_a_sentence_inside_the_cap(self):
        from pfs.descriptors import descriptor_span

        text = "HCPCS code G0502 (Initial psychiatric care. " + "More words. " * 400
        span = descriptor_span(text, "G0502", max_chars=200)
        assert len(span) <= 200
        assert span.endswith(".")

    def test_no_stem_returns_the_text(self):
        from pfs.descriptors import descriptor_span

        assert descriptor_span("nothing about the code here", "G0502") == (
            "nothing about the code here"
        )

    def test_descriptor_runs_stem_is_the_span(self, store):
        from pfs.descriptors import descriptor_runs

        con = store._con()
        con.execute(
            "INSERT INTO fr_anchors VALUES (?,?,?,?,?)",
            (
                "LONGPARA",
                9,
                1,
                1,
                "As background, HCPCS code G0502 (Initial psychiatric collaborative "
                "care management, first 70 minutes) was created in CY 2017. "
                + "Chronic care management prose. "
                * 100,
            ),
        )
        con.execute(
            "INSERT INTO items (key, title, date_published) VALUES (?,?,?)",
            ("LONGPARA", "Medicare Program; CY 2027 PFS Proposed Rule", "2026-07-16"),
        )
        con.commit()
        run = [r for r in descriptor_runs(store, "G0502") if r.item_key == "LONGPARA"][
            0
        ]
        assert run.stem.text.startswith("G0502 (Initial psychiatric")
        assert "Chronic care management prose" not in run.stem.text