137 lines
4.0 KiB
Python
137 lines
4.0 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import date
|
|
from unittest.mock import MagicMock
|
|
|
|
import httpx
|
|
|
|
from bib.dockets import Docket
|
|
from bib.regulations_gov import Comment, WalkResult, discover_docket, walk_docket
|
|
from bib.store import Store
|
|
|
|
D = "CMS-2026-2377"
|
|
|
|
|
|
def _c(n: int, lm: str) -> Comment:
|
|
return Comment(
|
|
id=f"{D}-{n}",
|
|
title="",
|
|
posted_date=lm[:10],
|
|
received_date=lm[:10],
|
|
docket_id=D,
|
|
comment_on_id="x",
|
|
raw={"attributes": {"lastModifiedDate": lm}},
|
|
)
|
|
|
|
|
|
def _store() -> Store:
|
|
return Store(":memory:", storage_dir="/tmp/nope")
|
|
|
|
|
|
def _docket(**kw) -> Docket:
|
|
base = dict(
|
|
id=D,
|
|
rule_cms_id="CMS-1848-P",
|
|
fr_object_id="obj",
|
|
comment_end_date="2026-09-14",
|
|
)
|
|
base.update(kw)
|
|
return Docket(**base)
|
|
|
|
|
|
def test_discover_docket_picks_commentable_doc():
|
|
api = MagicMock()
|
|
api.find_documents_in_docket.return_value = [
|
|
{"id": "X-1", "attributes": {"objectId": "o1"}}, # no comment window
|
|
{
|
|
"id": "X-2",
|
|
"attributes": {"objectId": "o2", "commentEndDate": "2026-09-14T03:59:59Z"},
|
|
},
|
|
]
|
|
d = discover_docket(api, D, rule_cms_id="CMS-1848-P")
|
|
assert d == Docket(
|
|
id=D,
|
|
rule_cms_id="CMS-1848-P",
|
|
fr_document_id="X-2",
|
|
fr_object_id="o2",
|
|
comment_end_date="2026-09-14",
|
|
)
|
|
api.find_documents_in_docket.assert_called_once_with(D)
|
|
|
|
|
|
def test_walk_creates_counts_and_advances_watermark():
|
|
s = _store()
|
|
s.docket_upsert(_docket())
|
|
api = MagicMock()
|
|
api.iter_comments.return_value = [
|
|
_c(1, "2026-09-01T00:00:00Z"),
|
|
_c(2, "2026-09-02T00:00:00Z"),
|
|
]
|
|
r = walk_docket(s, api, s.docket_get(D), today=date(2026, 9, 8))
|
|
assert r == WalkResult(
|
|
created=2,
|
|
updated=0,
|
|
unchanged=0,
|
|
watermark="2026-09-02T00:00:00Z",
|
|
clean=True,
|
|
sealed=False,
|
|
)
|
|
d = s.docket_get(D)
|
|
assert d.pull_watermark == "2026-09-02T00:00:00Z"
|
|
assert d.last_pull_new == 2 and d.last_pull_at
|
|
api.iter_comments.assert_called_once()
|
|
assert api.iter_comments.call_args.kwargs["since"] == ""
|
|
|
|
|
|
def test_walk_passes_watermark_and_counts_unchanged():
|
|
s = _store()
|
|
s.docket_upsert(_docket())
|
|
api = MagicMock()
|
|
api.iter_comments.return_value = [_c(1, "2026-09-01T00:00:00Z")]
|
|
walk_docket(s, api, s.docket_get(D), today=date(2026, 9, 8))
|
|
api.iter_comments.return_value = [
|
|
_c(1, "2026-09-01T00:00:00Z")
|
|
] # boundary re-yield
|
|
r = walk_docket(s, api, s.docket_get(D), today=date(2026, 9, 8))
|
|
assert api.iter_comments.call_args.kwargs["since"] == "2026-09-01T00:00:00Z"
|
|
assert (r.created, r.unchanged) == (0, 1)
|
|
assert s.docket_get(D).last_pull_new == 0
|
|
|
|
|
|
def test_unclean_walk_keeps_old_watermark():
|
|
s = _store()
|
|
s.docket_upsert(_docket(pull_watermark="2026-08-01T00:00:00Z"))
|
|
api = MagicMock()
|
|
|
|
def _iter(_obj, *, since="", on_error=None):
|
|
yield _c(9, "2026-09-05T00:00:00Z")
|
|
on_error(
|
|
httpx.HTTPStatusError("boom", request=MagicMock(), response=MagicMock())
|
|
)
|
|
|
|
api.iter_comments.side_effect = _iter
|
|
r = walk_docket(s, api, s.docket_get(D), today=date(2026, 9, 8))
|
|
assert r.clean is False and r.created == 1
|
|
assert s.docket_get(D).pull_watermark == "2026-08-01T00:00:00Z"
|
|
|
|
|
|
def test_force_walks_from_scratch_and_never_seals():
|
|
s = _store()
|
|
s.docket_upsert(_docket(pull_watermark="2026-08-01T00:00:00Z"))
|
|
api = MagicMock()
|
|
api.iter_comments.return_value = []
|
|
r = walk_docket(s, api, s.docket_get(D), force=True, today=date(2027, 1, 1))
|
|
assert api.iter_comments.call_args.kwargs["since"] == ""
|
|
assert r.sealed is False and not s.docket_get(D).sealed
|
|
|
|
|
|
def test_auto_seal_after_quiet_empty_pull():
|
|
s = _store()
|
|
s.docket_upsert(_docket())
|
|
api = MagicMock()
|
|
api.iter_comments.return_value = []
|
|
r = walk_docket(s, api, s.docket_get(D), today=date(2026, 10, 20), quiet_days=30)
|
|
assert r.sealed is True
|
|
d = s.docket_get(D)
|
|
assert d.sealed and d.seal_reason == "auto"
|