543 lines
18 KiB
Python
543 lines
18 KiB
Python
"""Exercising tests for cli/bib.py — covers function bodies, not just --help."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from typer.testing import CliRunner
|
|
|
|
from cli.bib import app
|
|
|
|
runner = CliRunner()
|
|
|
|
|
|
class TestSyncEdge:
|
|
@patch("bib.meta.collect_column_comments", return_value={})
|
|
@patch("bib.connect", return_value=MagicMock())
|
|
def test_sync_empty_comments(self, mc_connect, mc_cc):
|
|
result = runner.invoke(app, ["sync"])
|
|
assert result.exit_code == 0
|
|
assert "No col: tags" in result.output
|
|
|
|
|
|
class TestQueryEdge:
|
|
@patch("bib.connect")
|
|
def test_query_many_items(self, mc):
|
|
store = MagicMock()
|
|
items = []
|
|
for i in range(25):
|
|
item = MagicMock()
|
|
item.key = f"K{i}"
|
|
item.title = f"Title {i}"
|
|
item.tags = ["t1", "t2"]
|
|
items.append(item)
|
|
store.list_items.return_value = items
|
|
mc.return_value = store
|
|
result = runner.invoke(app, ["query", "test"])
|
|
assert result.exit_code == 0
|
|
assert "and 5 more" in result.output
|
|
|
|
@patch("bib.connect")
|
|
def test_query_no_tags(self, mc):
|
|
store = MagicMock()
|
|
item = MagicMock()
|
|
item.key = "K1"
|
|
item.title = "Title 1"
|
|
item.tags = []
|
|
store.list_items.return_value = [item]
|
|
mc.return_value = store
|
|
result = runner.invoke(app, ["query", "test"])
|
|
assert result.exit_code == 0
|
|
|
|
|
|
class TestHttpClient:
|
|
def test_http_client(self):
|
|
from cli.bib import _http_client
|
|
|
|
with patch("httpx.Client") as mc:
|
|
mc.return_value = MagicMock()
|
|
_http_client()
|
|
mc.assert_called_once()
|
|
|
|
|
|
class TestDiscoverPfsRulesExercise:
|
|
@patch("bib.connect")
|
|
@patch("bib.federalregister.pfs_rules")
|
|
@patch("bib.translate.federal_register")
|
|
def test_ingest_path(self, mc_translate, mc_pfs, mc_connect):
|
|
doc = MagicMock()
|
|
doc.publication_date = "2023-01-01"
|
|
doc.type = "Proposed Rule"
|
|
doc.document_number = "2023-12345"
|
|
doc.dockets = ["CMS-1676-P"]
|
|
doc.html_url = "https://example.com/doc"
|
|
mc_pfs.return_value = [doc]
|
|
|
|
rule = MagicMock()
|
|
rule.add_tag = MagicMock()
|
|
mc_translate.return_value = rule
|
|
|
|
store = MagicMock()
|
|
mc_connect.return_value = store
|
|
|
|
result = runner.invoke(app, ["discover-pfs-rules"])
|
|
assert result.exit_code == 0
|
|
store.upsert.assert_called_once()
|
|
|
|
@patch("bib.federalregister.pfs_rules")
|
|
@patch("bib.translate.federal_register")
|
|
def test_ingest_skip_on_error(self, mc_translate, mc_pfs):
|
|
doc = MagicMock()
|
|
doc.publication_date = "2023-01-01"
|
|
doc.type = "Proposed Rule"
|
|
doc.document_number = "2023-12345"
|
|
doc.dockets = []
|
|
doc.html_url = None
|
|
mc_pfs.return_value = [doc]
|
|
mc_translate.side_effect = ValueError("bad doc")
|
|
|
|
with patch("bib.connect", return_value=MagicMock()):
|
|
result = runner.invoke(app, ["discover-pfs-rules"])
|
|
assert result.exit_code == 0
|
|
assert "skipped" in result.output
|
|
|
|
|
|
class TestFetchDocketComments:
|
|
@patch("bib.connect")
|
|
@patch("bib.regulations_gov.Client")
|
|
@patch("bib.regulations_gov.discover_docket")
|
|
@patch("bib.regulations_gov.walk_docket")
|
|
def test_basic(self, mc_walk, mc_disc, mc_client_cls, mc_connect):
|
|
from bib.dockets import Docket
|
|
from bib.regulations_gov import WalkResult
|
|
from bib.store import Store
|
|
|
|
store = Store(":memory:", storage_dir="/tmp/nope")
|
|
mc_connect.return_value = store
|
|
|
|
api = MagicMock()
|
|
api.__enter__ = MagicMock(return_value=api)
|
|
api.__exit__ = MagicMock(return_value=False)
|
|
mc_client_cls.return_value = api
|
|
|
|
mc_disc.return_value = Docket(
|
|
id="CMS-1676-P", fr_object_id="09000001", comment_end_date="2023-12-31"
|
|
)
|
|
mc_walk.return_value = WalkResult(created=1)
|
|
|
|
result = runner.invoke(app, ["fetch-docket-comments", "CMS-1676-P", "-n", "5"])
|
|
assert result.exit_code == 0
|
|
mc_disc.assert_called_once()
|
|
mc_walk.assert_called_once()
|
|
|
|
@patch("bib.connect")
|
|
@patch("bib.regulations_gov.Client")
|
|
@patch("bib.regulations_gov.discover_docket")
|
|
@patch("bib.regulations_gov.walk_docket")
|
|
def test_with_attachments(self, mc_walk, mc_disc, mc_client_cls, mc_connect):
|
|
from bib.dockets import Docket
|
|
from bib.regulations_gov import WalkResult
|
|
from bib.store import Store
|
|
|
|
store = Store(":memory:", storage_dir="/tmp/nope")
|
|
mc_connect.return_value = store
|
|
|
|
api = MagicMock()
|
|
api.__enter__ = MagicMock(return_value=api)
|
|
api.__exit__ = MagicMock(return_value=False)
|
|
mc_client_cls.return_value = api
|
|
|
|
mc_disc.return_value = Docket(
|
|
id="CMS-1676-P", fr_object_id="09000001", comment_end_date="2023-12-31"
|
|
)
|
|
mc_walk.return_value = WalkResult(created=1)
|
|
|
|
result = runner.invoke(
|
|
app,
|
|
["fetch-docket-comments", "CMS-1676-P", "--attachments", "-n", "5"],
|
|
)
|
|
assert result.exit_code == 0
|
|
assert mc_walk.call_args.kwargs["attachments"] is True
|
|
|
|
@patch("bib.connect")
|
|
@patch("bib.regulations_gov.Client")
|
|
@patch("bib.regulations_gov.discover_docket")
|
|
@patch("bib.regulations_gov.walk_docket")
|
|
def test_skip_no_object_id(self, mc_walk, mc_disc, mc_client_cls, mc_connect):
|
|
from bib.dockets import Docket
|
|
from bib.store import Store
|
|
|
|
store = Store(":memory:", storage_dir="/tmp/nope")
|
|
mc_connect.return_value = store
|
|
|
|
api = MagicMock()
|
|
api.__enter__ = MagicMock(return_value=api)
|
|
api.__exit__ = MagicMock(return_value=False)
|
|
mc_client_cls.return_value = api
|
|
|
|
mc_disc.return_value = Docket(id="CMS-1676-P")
|
|
|
|
result = runner.invoke(app, ["fetch-docket-comments", "CMS-1676-P"])
|
|
assert result.exit_code == 0
|
|
assert "no commentable FR document" in result.output
|
|
mc_walk.assert_not_called()
|
|
|
|
|
|
class TestFetchPfsComments:
|
|
@patch("bib.connect")
|
|
@patch("bib.federalregister.pfs_rules")
|
|
@patch("bib.federalregister.split_docket_ids", return_value=["CMS-1676-P"])
|
|
@patch("bib.translate.federal_register")
|
|
@patch("bib.regulations_gov.Client")
|
|
@patch("bib.regulations_gov.discover_docket")
|
|
@patch("bib.regulations_gov.walk_docket")
|
|
def test_full_loop(
|
|
self,
|
|
mc_walk,
|
|
mc_disc,
|
|
mc_client_cls,
|
|
mc_translate,
|
|
mc_split,
|
|
mc_pfs,
|
|
mc_connect,
|
|
):
|
|
from bib.dockets import Docket
|
|
from bib.item import Rule
|
|
from bib.regulations_gov import WalkResult
|
|
from bib.store import Store
|
|
|
|
store = Store(":memory:", storage_dir="/tmp/nope")
|
|
mc_connect.return_value = store
|
|
|
|
doc = MagicMock()
|
|
doc.type = "Proposed Rule"
|
|
doc.publication_date = "2023-01-01"
|
|
doc.dockets = ["CMS-1676-P"]
|
|
doc.html_url = "https://example.com"
|
|
doc.document_number = "2023-12345"
|
|
mc_pfs.return_value = [doc]
|
|
|
|
rule = Rule(title="CY2024 PFS NPRM", url="https://example.com")
|
|
mc_translate.return_value = rule
|
|
|
|
api = MagicMock()
|
|
api.__enter__ = MagicMock(return_value=api)
|
|
api.__exit__ = MagicMock(return_value=False)
|
|
mc_client_cls.return_value = api
|
|
api.resolve_docket.return_value = "CMS-2023-0001"
|
|
|
|
mc_disc.return_value = Docket(
|
|
id="CMS-2023-0001",
|
|
rule_cms_id="CMS-1676-P",
|
|
fr_object_id="09000001",
|
|
comment_end_date="2023-12-31",
|
|
)
|
|
mc_walk.return_value = WalkResult(created=1)
|
|
|
|
result = runner.invoke(app, ["fetch-pfs-comments", "--per-docket-limit", "1"])
|
|
assert result.exit_code == 0
|
|
mc_disc.assert_called_once()
|
|
mc_walk.assert_called_once()
|
|
|
|
@patch("bib.connect")
|
|
@patch("bib.federalregister.pfs_rules", return_value=[])
|
|
def test_no_rules(self, mc_pfs, mc_connect):
|
|
mc_connect.return_value = MagicMock()
|
|
api = MagicMock()
|
|
api.__enter__ = MagicMock(return_value=api)
|
|
api.__exit__ = MagicMock(return_value=False)
|
|
with patch("bib.regulations_gov.Client", return_value=api):
|
|
result = runner.invoke(app, ["fetch-pfs-comments"])
|
|
assert result.exit_code == 0
|
|
|
|
@patch("bib.connect")
|
|
@patch("bib.federalregister.pfs_rules")
|
|
@patch("bib.federalregister.split_docket_ids", return_value=["CMS-1676-P"])
|
|
@patch("bib.translate.federal_register", side_effect=ValueError("bad"))
|
|
@patch("bib.regulations_gov.Client")
|
|
def test_skip_bad_rule(
|
|
self, mc_client_cls, mc_translate, mc_split, mc_pfs, mc_connect
|
|
):
|
|
from bib.store import Store
|
|
|
|
store = Store(":memory:", storage_dir="/tmp/nope")
|
|
mc_connect.return_value = store
|
|
|
|
doc = MagicMock()
|
|
doc.type = "Proposed Rule"
|
|
doc.publication_date = "2023-01-01"
|
|
doc.dockets = ["CMS-1676-P"]
|
|
doc.html_url = "https://example.com"
|
|
mc_pfs.return_value = [doc]
|
|
|
|
api = MagicMock()
|
|
api.__enter__ = MagicMock(return_value=api)
|
|
api.__exit__ = MagicMock(return_value=False)
|
|
mc_client_cls.return_value = api
|
|
|
|
result = runner.invoke(app, ["fetch-pfs-comments"])
|
|
assert result.exit_code == 0
|
|
mc_translate.assert_called_once()
|
|
assert "skipped rule meta" in result.output
|
|
|
|
@patch("bib.connect")
|
|
@patch("bib.federalregister.pfs_rules")
|
|
@patch("bib.federalregister.split_docket_ids", return_value=["CMS-1676-P"])
|
|
@patch("bib.translate.federal_register")
|
|
@patch("bib.regulations_gov.Client")
|
|
def test_no_docket(self, mc_client_cls, mc_translate, mc_split, mc_pfs, mc_connect):
|
|
from bib.store import Store
|
|
|
|
store = Store(":memory:", storage_dir="/tmp/nope")
|
|
mc_connect.return_value = store
|
|
|
|
doc = MagicMock()
|
|
doc.type = "Proposed Rule"
|
|
doc.publication_date = "2023-01-01"
|
|
doc.dockets = ["CMS-1676-P"]
|
|
doc.html_url = "https://example.com"
|
|
mc_pfs.return_value = [doc]
|
|
|
|
rule = MagicMock()
|
|
mc_translate.return_value = rule
|
|
|
|
api = MagicMock()
|
|
api.__enter__ = MagicMock(return_value=api)
|
|
api.__exit__ = MagicMock(return_value=False)
|
|
mc_client_cls.return_value = api
|
|
api.resolve_docket.return_value = None
|
|
|
|
result = runner.invoke(app, ["fetch-pfs-comments"])
|
|
assert result.exit_code == 0
|
|
assert "skip" in result.output
|
|
|
|
|
|
class TestIngestMail:
|
|
@patch("bib.connect")
|
|
@patch("bib.email_ingest.ingest", return_value={"new": 3, "skipped": 1})
|
|
def test_basic(self, mc_ingest, mc_connect, tmp_path):
|
|
store = MagicMock()
|
|
mc_connect.return_value = store
|
|
|
|
creds = tmp_path / "credentials.json"
|
|
creds.write_text('{"cmsupdates": "pw123"}')
|
|
|
|
with (
|
|
patch.object(Path, "exists", return_value=True),
|
|
patch.object(Path, "read_text", return_value='{"cmsupdates": "pw123"}'),
|
|
):
|
|
result = runner.invoke(
|
|
app,
|
|
["ingest-mail", "--user", "cmsupdates@mail.fhirworx.io"],
|
|
)
|
|
assert result.exit_code == 0
|
|
|
|
def test_no_creds_file(self):
|
|
with patch.object(Path, "exists", return_value=False):
|
|
result = runner.invoke(app, ["ingest-mail"])
|
|
assert result.exit_code != 0
|
|
|
|
|
|
class TestBackfillComments:
|
|
@patch("bib.connect")
|
|
@patch("bib.regulations_gov.Client")
|
|
@patch("bib.regulations_gov.backfill_details", return_value={"enriched": 10})
|
|
def test_basic(self, mc_backfill, mc_client_cls, mc_connect):
|
|
store = MagicMock()
|
|
mc_connect.return_value = store
|
|
|
|
api = MagicMock()
|
|
api.__enter__ = MagicMock(return_value=api)
|
|
api.__exit__ = MagicMock(return_value=False)
|
|
mc_client_cls.return_value = api
|
|
|
|
result = runner.invoke(app, ["backfill-comments", "-n", "10"])
|
|
assert result.exit_code == 0
|
|
assert "enriched" in result.output
|
|
|
|
@patch("bib.connect")
|
|
@patch("bib.regulations_gov.Client")
|
|
@patch("bib.regulations_gov.backfill_details", return_value={"enriched": 1})
|
|
def test_docket_forwarded(self, mc_backfill, mc_client_cls, mc_connect):
|
|
store = MagicMock()
|
|
mc_connect.return_value = store
|
|
|
|
api = MagicMock()
|
|
api.__enter__ = MagicMock(return_value=api)
|
|
api.__exit__ = MagicMock(return_value=False)
|
|
mc_client_cls.return_value = api
|
|
|
|
result = runner.invoke(app, ["backfill-comments", "--docket", "CMS-2026-2377"])
|
|
assert result.exit_code == 0
|
|
assert mc_backfill.call_args.kwargs["docket"] == "CMS-2026-2377"
|
|
|
|
|
|
class TestIngestIom:
|
|
@patch("bib.connect")
|
|
@patch("cli.bib._http_client")
|
|
@patch("bib.iom.ingest_all", return_value={"100-04": 15, "100-02": 10})
|
|
def test_basic(self, mc_ingest, mc_http, mc_connect):
|
|
store = MagicMock()
|
|
mc_connect.return_value = store
|
|
|
|
client = MagicMock()
|
|
client.__enter__ = MagicMock(return_value=client)
|
|
client.__exit__ = MagicMock(return_value=False)
|
|
mc_http.return_value = client
|
|
|
|
result = runner.invoke(app, ["ingest-iom"])
|
|
assert result.exit_code == 0
|
|
assert "chapters" in result.output
|
|
|
|
|
|
class TestAttachIom:
|
|
@patch("bib.connect")
|
|
@patch("cli.bib._http_client")
|
|
@patch("bib.iom.download_attachments", return_value={"100-04": 5})
|
|
def test_basic(self, mc_dl, mc_http, mc_connect):
|
|
store = MagicMock()
|
|
mc_connect.return_value = store
|
|
|
|
client = MagicMock()
|
|
client.__enter__ = MagicMock(return_value=client)
|
|
client.__exit__ = MagicMock(return_value=False)
|
|
mc_http.return_value = client
|
|
|
|
result = runner.invoke(app, ["attach-iom"])
|
|
assert result.exit_code == 0
|
|
|
|
|
|
class TestWatchIom:
|
|
@patch("bib.connect")
|
|
@patch("cli.bib._http_client")
|
|
@patch("bib.iom.check_future_updates", return_value=(True, "abcdef1234567890"))
|
|
def test_changed(self, mc_check, mc_http, mc_connect):
|
|
store = MagicMock()
|
|
mc_connect.return_value = store
|
|
|
|
client = MagicMock()
|
|
client.__enter__ = MagicMock(return_value=client)
|
|
client.__exit__ = MagicMock(return_value=False)
|
|
mc_http.return_value = client
|
|
|
|
result = runner.invoke(app, ["watch-iom"])
|
|
assert result.exit_code == 0
|
|
assert "CHANGED" in result.output
|
|
|
|
|
|
class TestSyncZotero:
|
|
@patch("bib.connect")
|
|
@patch(
|
|
"bib.sync.push_to_zotero",
|
|
return_value={"created": 5, "skipped": 2, "attachments": 3},
|
|
)
|
|
@patch("subprocess.run")
|
|
def test_basic(self, mc_sub, mc_push, mc_connect):
|
|
store = MagicMock()
|
|
store.list_items.return_value = []
|
|
mc_connect.return_value = store
|
|
|
|
result = runner.invoke(app, ["sync-zotero"])
|
|
assert result.exit_code == 0
|
|
|
|
@patch("bib.connect")
|
|
@patch(
|
|
"bib.sync.push_to_zotero",
|
|
return_value={"created": 0, "skipped": 0, "attachments": 0},
|
|
)
|
|
@patch("subprocess.run")
|
|
def test_no_hold(self, mc_sub, mc_push, mc_connect):
|
|
store = MagicMock()
|
|
store.list_items.return_value = []
|
|
mc_connect.return_value = store
|
|
|
|
result = runner.invoke(app, ["sync-zotero", "--no-hold"])
|
|
assert result.exit_code == 0
|
|
mc_sub.assert_not_called()
|
|
|
|
|
|
class TestIngestOig:
|
|
@patch("bib.connect")
|
|
@patch("cli.bib._http_client")
|
|
@patch("bib.oig.ingest_all", return_value={"cpg": 10, "alerts": 5})
|
|
def test_basic(self, mc_ingest, mc_http, mc_connect):
|
|
store = MagicMock()
|
|
mc_connect.return_value = store
|
|
|
|
client = MagicMock()
|
|
client.__enter__ = MagicMock(return_value=client)
|
|
client.__exit__ = MagicMock(return_value=False)
|
|
mc_http.return_value = client
|
|
|
|
result = runner.invoke(app, ["ingest-oig"])
|
|
assert result.exit_code == 0
|
|
|
|
|
|
class TestAttachOig:
|
|
@patch("bib.connect")
|
|
@patch("cli.bib._http_client")
|
|
@patch("bib.oig.download_attachments", return_value=3)
|
|
def test_basic(self, mc_dl, mc_http, mc_connect):
|
|
store = MagicMock()
|
|
mc_connect.return_value = store
|
|
|
|
client = MagicMock()
|
|
client.__enter__ = MagicMock(return_value=client)
|
|
client.__exit__ = MagicMock(return_value=False)
|
|
mc_http.return_value = client
|
|
|
|
result = runner.invoke(app, ["attach-oig"])
|
|
assert result.exit_code == 0
|
|
|
|
|
|
class TestRefreshOig:
|
|
@patch("bib.connect")
|
|
@patch("cli.bib._http_client")
|
|
@patch("bib.oig.ingest_all", return_value={"cpg": 10, "alerts": 5})
|
|
@patch("bib.oig.download_attachments", return_value=3)
|
|
@patch(
|
|
"bib.sync.push_to_zotero",
|
|
return_value={"created": 5, "skipped": 2, "attachments": 3},
|
|
)
|
|
@patch("subprocess.run")
|
|
def test_basic(self, mc_sub, mc_push, mc_dl, mc_ingest, mc_http, mc_connect):
|
|
store = MagicMock()
|
|
store.list_items.return_value = []
|
|
mc_connect.return_value = store
|
|
|
|
client = MagicMock()
|
|
client.__enter__ = MagicMock(return_value=client)
|
|
client.__exit__ = MagicMock(return_value=False)
|
|
mc_http.return_value = client
|
|
|
|
result = runner.invoke(app, ["refresh-oig"])
|
|
assert result.exit_code == 0
|
|
|
|
|
|
class TestRefreshIom:
|
|
@patch("bib.connect")
|
|
@patch("cli.bib._http_client")
|
|
@patch("bib.iom.ingest_all", return_value={"100-04": 15})
|
|
@patch("bib.iom.check_future_updates", return_value=(False, "abc123"))
|
|
@patch("bib.iom.download_attachments", return_value={"100-04": 2})
|
|
@patch(
|
|
"bib.sync.push_to_zotero",
|
|
return_value={"created": 5, "skipped": 2, "attachments": 3},
|
|
)
|
|
@patch("subprocess.run")
|
|
def test_basic(
|
|
self, mc_sub, mc_push, mc_dl, mc_check, mc_ingest, mc_http, mc_connect
|
|
):
|
|
store = MagicMock()
|
|
store.list_items.return_value = []
|
|
mc_connect.return_value = store
|
|
|
|
client = MagicMock()
|
|
client.__enter__ = MagicMock(return_value=client)
|
|
client.__exit__ = MagicMock(return_value=False)
|
|
mc_http.return_value = client
|
|
|
|
result = runner.invoke(app, ["refresh-iom"])
|
|
assert result.exit_code == 0
|