118 lines
3.8 KiB
Python
118 lines
3.8 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
|
|
from rex.comments.combine import is_current, source_paths
|
|
from rex.comments.walker import walk_and_extract
|
|
|
|
|
|
def _dir(tmp_path: Path, docket="CMS-2024-0001", cid="CMS-2024-0001-0001") -> Path:
|
|
d = tmp_path / docket / cid
|
|
d.mkdir(parents=True)
|
|
return d
|
|
|
|
|
|
def test_source_paths_excludes_outputs(tmp_path: Path):
|
|
d = _dir(tmp_path)
|
|
(d / "attachment_1.pdf").write_bytes(b"x")
|
|
(d / "attachment_1.pdf.md").write_text("sibling")
|
|
(d / "combined.md").write_text("c")
|
|
(d / "combined.md.tmp").write_text("t")
|
|
(d / ".hidden").write_text("h")
|
|
assert [p.name for p in source_paths(d)] == ["attachment_1.pdf"]
|
|
|
|
|
|
def test_is_current_false_without_combined(tmp_path: Path):
|
|
d = _dir(tmp_path)
|
|
(d / "attachment_1.pdf").write_bytes(b"x")
|
|
assert is_current(d) is False
|
|
|
|
|
|
def test_is_current_true_when_combined_newer(tmp_path: Path):
|
|
d = _dir(tmp_path)
|
|
(d / "attachment_1.pdf").write_bytes(b"x")
|
|
os.utime(d / "attachment_1.pdf", ns=(1_000, 1_000))
|
|
(d / "combined.md").write_text("c")
|
|
assert is_current(d) is True
|
|
|
|
|
|
def test_is_current_false_when_source_newer(tmp_path: Path):
|
|
d = _dir(tmp_path)
|
|
(d / "combined.md").write_text("c")
|
|
os.utime(d / "combined.md", ns=(1_000, 1_000))
|
|
(d / "attachment_2.pdf").write_bytes(b"new")
|
|
assert is_current(d) is False
|
|
|
|
|
|
def test_walker_reextracts_stale_dir(tmp_path: Path, monkeypatch):
|
|
d = _dir(tmp_path)
|
|
(d / "combined.md").write_text("---\ncomment_id: x\n---\n\nold\n")
|
|
os.utime(d / "combined.md", ns=(1_000, 1_000))
|
|
(d / "attachment_1.pdf").write_bytes(b"%PDF") # newer than combined.md
|
|
called = []
|
|
monkeypatch.setattr(
|
|
"rex.comments.walker.extract_comment",
|
|
lambda cdir, **kw: called.append(cdir) or (cdir / "combined.md"),
|
|
)
|
|
stats = walk_and_extract(tmp_path, inline_body_lookup=lambda _c: "", workers=1)
|
|
assert stats["written"] == 1 and called == [d]
|
|
|
|
|
|
def test_walker_skips_current_dir_without_reading(tmp_path: Path, monkeypatch):
|
|
d = _dir(tmp_path)
|
|
(d / "attachment_1.pdf").write_bytes(b"%PDF")
|
|
os.utime(d / "attachment_1.pdf", ns=(1_000, 1_000))
|
|
(d / "combined.md").write_text("---\ncomment_id: x\n---\n\nbody\n")
|
|
reads = []
|
|
real_read_text = Path.read_text
|
|
|
|
def spy(self, *a, **kw):
|
|
if self.name == "combined.md":
|
|
reads.append(self)
|
|
return real_read_text(self, *a, **kw)
|
|
|
|
monkeypatch.setattr(Path, "read_text", spy)
|
|
seen = []
|
|
stats = walk_and_extract(
|
|
tmp_path,
|
|
inline_body_lookup=lambda _c: "",
|
|
workers=1,
|
|
on_extracted=lambda cid, _d: seen.append(cid),
|
|
)
|
|
assert stats == {"written": 0, "skipped": 1, "failed": 0}
|
|
assert reads == [] # skipped dirs are not read
|
|
assert seen == [] # and the callback does not fire without --reattach
|
|
|
|
|
|
def test_walker_reattach_fires_callback_for_skipped(tmp_path: Path):
|
|
d = _dir(tmp_path)
|
|
(d / "attachment_1.pdf").write_bytes(b"%PDF")
|
|
os.utime(d / "attachment_1.pdf", ns=(1_000, 1_000))
|
|
(d / "combined.md").write_text(
|
|
"---\ncomment_id: x\ndocket_id: y\n---\n\n## attachment_1.pdf\n\nbody\n"
|
|
)
|
|
seen = []
|
|
walk_and_extract(
|
|
tmp_path,
|
|
inline_body_lookup=lambda _c: "",
|
|
workers=1,
|
|
reattach=True,
|
|
on_extracted=lambda cid, _d: seen.append(cid),
|
|
)
|
|
assert seen == [d.name]
|
|
assert (d / "attachment_1.pdf.md").is_file() # siblings derived on reattach
|
|
|
|
|
|
def test_walker_skip_dockets(tmp_path: Path):
|
|
d = _dir(tmp_path)
|
|
(d / "attachment_1.pdf").write_bytes(b"%PDF")
|
|
stats = walk_and_extract(
|
|
tmp_path,
|
|
inline_body_lookup=lambda _c: "",
|
|
workers=1,
|
|
skip_dockets={"CMS-2024-0001"},
|
|
)
|
|
assert stats == {"written": 0, "skipped": 0, "failed": 0, "skipped_sealed": 1}
|
|
assert not (d / "combined.md").exists()
|