b5e07e67b2
The PFS register-and-walk path now serves the annual IPPS/LTCH and OPPS/ASC rules and DMEPOS payment rules: title-filtered FR discovery (discover-*-rules) and docket registration + comment walks (fetch-*-comments). Correction documents (C1-...) parse, a null FR abstract no longer becomes 'None', and the store waits out the mail poller's write bursts (busy_timeout) instead of failing SQLITE_BUSY.
210 lines
7.0 KiB
Python
210 lines
7.0 KiB
Python
"""IPPS and OPPS comment discovery uses the PFS register-and-walk path."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from typer.testing import CliRunner
|
|
|
|
from bib.dockets import Docket
|
|
from bib.item import Rule
|
|
from bib.regulations_gov import WalkResult
|
|
from bib.store import Store
|
|
from cli.bib import app
|
|
|
|
runner = CliRunner()
|
|
|
|
|
|
def _api():
|
|
api = MagicMock()
|
|
api.__enter__ = MagicMock(return_value=api)
|
|
api.__exit__ = MagicMock(return_value=False)
|
|
api.find_documents_in_docket.return_value = []
|
|
api.iter_comments.return_value = []
|
|
return api
|
|
|
|
|
|
def _doc(number: str, cms_id: str):
|
|
doc = MagicMock()
|
|
doc.type = "Proposed Rule"
|
|
doc.publication_date = "2018-05-07"
|
|
doc.dockets = [cms_id]
|
|
doc.html_url = f"https://example.com/{number}"
|
|
doc.document_number = number
|
|
return doc
|
|
|
|
|
|
@patch("bib.federalregister.ipps_rules", return_value=[])
|
|
@patch("bib.connect")
|
|
def test_fetch_ipps_comments_empty(mc_connect, _rules):
|
|
mc_connect.return_value = MagicMock()
|
|
api = _api()
|
|
with patch("bib.regulations_gov.Client", return_value=api):
|
|
result = runner.invoke(app, ["fetch-ipps-comments"])
|
|
assert result.exit_code == 0, result.output
|
|
assert "IPPS rules" in result.output
|
|
|
|
|
|
@patch("bib.federalregister.opps_rules", return_value=[])
|
|
@patch("bib.connect")
|
|
def test_fetch_opps_comments_empty(mc_connect, _rules):
|
|
mc_connect.return_value = MagicMock()
|
|
api = _api()
|
|
with patch("bib.regulations_gov.Client", return_value=api):
|
|
result = runner.invoke(app, ["fetch-opps-comments"])
|
|
assert result.exit_code == 0, result.output
|
|
assert "OPPS rules" in result.output
|
|
|
|
|
|
@patch("bib.federalregister.ipps_rules", return_value=[])
|
|
def test_discover_ipps_dry_run(_rules):
|
|
result = runner.invoke(app, ["discover-ipps-rules", "--dry-run"])
|
|
assert result.exit_code == 0
|
|
|
|
|
|
@patch("bib.federalregister.opps_rules", return_value=[])
|
|
def test_discover_opps_dry_run(_rules):
|
|
result = runner.invoke(app, ["discover-opps-rules", "--dry-run"])
|
|
assert result.exit_code == 0
|
|
|
|
|
|
@patch("bib.regulations_gov.walk_docket", return_value=WalkResult())
|
|
@patch("bib.regulations_gov.discover_docket")
|
|
@patch("bib.regulations_gov.Client")
|
|
@patch("bib.translate.federal_register")
|
|
@patch("bib.federalregister.ipps_rules")
|
|
@patch("bib.connect")
|
|
def test_correction_reuses_the_proposed_rule_docket(
|
|
mc_connect, mc_rules, mc_fr, mc_client, mc_disc, mc_walk
|
|
):
|
|
store = Store(":memory:", storage_dir="/tmp/nope")
|
|
mc_connect.return_value = store
|
|
mc_rules.return_value = [
|
|
_doc("2018-1", "CMS-1694-P"),
|
|
_doc("2018-2", "CMS-1694-CN"),
|
|
]
|
|
mc_fr.return_value = Rule(title="FY2019 IPPS", url="https://example.com/2018-1")
|
|
api = _api()
|
|
api.resolve_docket.return_value = "CMS-2018-0040"
|
|
mc_client.return_value = api
|
|
|
|
def _discover(_api, docket_id, rule_cms_id="", title_contains=""):
|
|
assert title_contains == "inpatient prospective payment"
|
|
return Docket(
|
|
id=docket_id,
|
|
rule_cms_id=rule_cms_id,
|
|
fr_object_id="obj",
|
|
comment_end_date="2018-06-25",
|
|
)
|
|
|
|
mc_disc.side_effect = _discover
|
|
|
|
result = runner.invoke(app, ["fetch-ipps-comments"])
|
|
assert result.exit_code == 0, result.output
|
|
mc_disc.assert_called_once()
|
|
saved = store.docket_get("CMS-2018-0040")
|
|
assert saved is not None
|
|
assert saved.rule_cms_id == "CMS-1694-P"
|
|
assert mc_walk.call_count == 2
|
|
|
|
|
|
@patch("bib.regulations_gov.walk_docket", return_value=WalkResult())
|
|
@patch("bib.regulations_gov.discover_docket")
|
|
@patch("bib.regulations_gov.Client")
|
|
@patch("bib.translate.federal_register")
|
|
@patch("bib.federalregister.opps_rules")
|
|
@patch("bib.connect")
|
|
def test_reused_sealed_docket_is_not_walked(
|
|
mc_connect, mc_rules, mc_fr, mc_client, mc_disc, mc_walk
|
|
):
|
|
store = Store(":memory:", storage_dir="/tmp/nope")
|
|
store.docket_upsert(
|
|
Docket(
|
|
id="CMS-2018-0042",
|
|
rule_cms_id="CMS-1695-P",
|
|
fr_object_id="obj",
|
|
sealed_at="2020-01-01T00:00:00Z",
|
|
seal_reason="auto",
|
|
)
|
|
)
|
|
mc_connect.return_value = store
|
|
mc_rules.return_value = [_doc("2018-3", "CMS-1695-CN")]
|
|
mc_fr.return_value = Rule(title="CY2019 OPPS", url="https://example.com/2018-3")
|
|
api = _api()
|
|
api.resolve_docket.return_value = "CMS-2018-0042"
|
|
mc_client.return_value = api
|
|
|
|
result = runner.invoke(app, ["fetch-opps-comments"])
|
|
assert result.exit_code == 0, result.output
|
|
assert "sealed" in result.output
|
|
mc_disc.assert_not_called()
|
|
mc_walk.assert_not_called()
|
|
|
|
|
|
@patch("bib.federalregister.dmepos_rules", return_value=[])
|
|
@patch("bib.connect")
|
|
def test_fetch_dmepos_comments_empty(mc_connect, _rules):
|
|
mc_connect.return_value = MagicMock()
|
|
api = _api()
|
|
with patch("bib.regulations_gov.Client", return_value=api):
|
|
result = runner.invoke(app, ["fetch-dmepos-comments"])
|
|
assert result.exit_code == 0, result.output
|
|
assert "DMEPOS rules" in result.output
|
|
|
|
|
|
@patch("bib.federalregister.dmepos_rules", return_value=[])
|
|
def test_discover_dmepos_dry_run(_rules):
|
|
result = runner.invoke(app, ["discover-dmepos-rules", "--dry-run"])
|
|
assert result.exit_code == 0
|
|
|
|
|
|
@patch("bib.regulations_gov.walk_docket", return_value=WalkResult())
|
|
@patch("bib.regulations_gov.pull_titled_siblings", return_value=0)
|
|
@patch("bib.regulations_gov.discover_docket")
|
|
@patch("bib.regulations_gov.Client")
|
|
@patch("bib.translate.federal_register")
|
|
@patch("bib.federalregister.dmepos_rules")
|
|
@patch("bib.connect")
|
|
def test_dmepos_resolve_omits_the_title_needle(
|
|
mc_connect, mc_rules, mc_fr, mc_client, mc_disc, _siblings, _walk
|
|
):
|
|
store = Store(":memory:", storage_dir="/tmp/nope")
|
|
mc_connect.return_value = store
|
|
acronym = _doc("2018-14986", "CMS-1691-P")
|
|
acronym.title = (
|
|
"Durable Medical Equipment, Prosthetics, Orthotics and "
|
|
"Supplies (DMEPOS) Competitive Bidding Program"
|
|
)
|
|
spelled = _doc("2023-14044", "CMS-1780-P")
|
|
spelled.title = (
|
|
"Certain Requirements for Durable Medical Equipment "
|
|
"Prosthetics and Orthotics Supplies"
|
|
)
|
|
mc_rules.return_value = [acronym, spelled]
|
|
mc_fr.return_value = Rule(title="DMEPOS", url="https://example.com/dmepos")
|
|
api = _api()
|
|
needles: list[tuple[str, str]] = []
|
|
|
|
def _resolve(cms_id, title_contains=""):
|
|
needles.append((cms_id, title_contains))
|
|
return "CMS-2018-0099" if cms_id == "CMS-1691-P" else "CMS-2023-0100"
|
|
|
|
api.resolve_docket.side_effect = _resolve
|
|
mc_client.return_value = api
|
|
|
|
def _discover(_api, docket_id, rule_cms_id="", title_contains=""):
|
|
return Docket(
|
|
id=docket_id,
|
|
rule_cms_id=rule_cms_id,
|
|
fr_object_id="obj",
|
|
comment_end_date="2018-09-10",
|
|
)
|
|
|
|
mc_disc.side_effect = _discover
|
|
|
|
result = runner.invoke(app, ["fetch-dmepos-comments"])
|
|
assert result.exit_code == 0, result.output
|
|
assert needles == [("CMS-1691-P", ""), ("CMS-1780-P", "")]
|
|
for call in api.resolve_docket.call_args_list:
|
|
assert "title_contains" not in call.kwargs
|