diff --git a/tests/test_textbooks_manifest.py b/tests/test_textbooks_manifest.py new file mode 100644 index 0000000..82d5009 --- /dev/null +++ b/tests/test_textbooks_manifest.py @@ -0,0 +1,463 @@ +"""Tests for ``bench.scripts.textbooks_manifest`` — manifest helper +that drives every per-textbook Make target (`make textbook ID=…`, +`make crawl-textbooks`, `make textbooks-tex`). + +Bugs here break the ingest pipeline silently. The script is the +license-discipline gate: ``_validate()`` is the fail-closed +chokepoint that refuses to emit URLs for non-allow-listed licenses. + +Coverage: + - iter_entries (skip _meta + blank lines) + - _validate (license allow-list enforcement, placeholder allowance, + missing-required-fields rejection) + - cmd_urls / cmd_summary / cmd_crawl_targets / cmd_tex_targets / + cmd_ids / cmd_lookup + - CLI dispatch via main() +""" + +from __future__ import annotations + +import io +import json +import sys +from contextlib import redirect_stderr, redirect_stdout + +import pytest + +from bench.scripts.textbooks_manifest import ( + _ALLOWED_LICENSES, + _REQUIRED_LICENSE_KEYS, + _validate, + cmd_crawl_targets, + cmd_ids, + cmd_lookup, + cmd_summary, + cmd_tex_targets, + cmd_urls, + iter_entries, + main, +) + + +# --- iter_entries --------------------------------------------------- + + +def test_iter_entries_skips_meta_block(): + """First-line _meta block must be skipped.""" + data = "\n".join([ + json.dumps({"_meta": {"version": "v1"}}), + json.dumps({"id": "real-entry", "license": "PD", + "license_url": "https://example.com"}), + ]) + entries = list(iter_entries(io.StringIO(data))) + assert len(entries) == 1 + assert entries[0]["id"] == "real-entry" + + +def test_iter_entries_skips_blank_lines(): + data = "\n".join([ + "", + json.dumps({"id": "a", "license": "PD", + "license_url": "https://example.com"}), + "", + json.dumps({"id": "b", "license": "MIT", + "license_url": "https://example.com"}), + ]) + entries = list(iter_entries(io.StringIO(data))) + assert {e["id"] for e in entries} == {"a", "b"} + + +def test_iter_entries_empty_stream_yields_nothing(): + assert list(iter_entries(io.StringIO(""))) == [] + + +# --- _validate ------------------------------------------------------ + + +@pytest.mark.parametrize("license_token", sorted(_ALLOWED_LICENSES)) +def test_validate_passes_for_each_allow_listed_license(license_token): + """Every license in the allow-list passes validation when the + entry has emit URLs.""" + entry = { + "id": "test-id", + "license": license_token, + "license_url": "https://example.com", + "urls": ["https://example.com/foo.html"], + } + _validate(entry) # must not raise + + +def test_validate_rejects_missing_license_field(): + entry = {"id": "test", "license_url": "https://example.com"} + with pytest.raises(ValueError, match="license"): + _validate(entry) + + +def test_validate_rejects_missing_license_url_field(): + entry = {"id": "test", "license": "PD"} + with pytest.raises(ValueError, match="license_url"): + _validate(entry) + + +def test_validate_rejects_empty_license_field(): + entry = {"id": "test", "license": "", "license_url": "x"} + with pytest.raises(ValueError, match="license"): + _validate(entry) + + +def test_validate_rejects_disallowed_license_with_urls(): + """Proprietary or CC-BY-NC license + emit URLs → ValueError.""" + entry = { + "id": "proprietary-test", + "license": "PROPRIETARY", + "license_url": "https://example.com", + "urls": ["https://example.com/foo"], + } + with pytest.raises(ValueError, match="redistribution allow-list"): + _validate(entry) + + +def test_validate_allows_placeholder_with_disallowed_license(): + """Placeholder rows (no urls + no crawl_url) bypass the + allow-list — kept for citation traceability per the docstring.""" + entry = { + "id": "yellow-light-placeholder", + "license": "PROPRIETARY", + "license_url": "https://example.com", + # No urls, no crawl_url + } + _validate(entry) # must not raise + + +def test_validate_allows_disallowed_license_with_empty_urls(): + """Empty `urls` list (not just missing) → still placeholder.""" + entry = { + "id": "explicit-empty", + "license": "CC-BY-NC-SA", + "license_url": "https://example.com", + "urls": [], + } + _validate(entry) # must not raise + + +def test_validate_rejects_disallowed_license_with_crawl_url(): + """`crawl_url` triggers the allow-list check just like `urls`.""" + entry = { + "id": "crawl-test", + "license": "PROPRIETARY", + "license_url": "https://example.com", + "crawl_url": "https://example.com/crawl", + } + with pytest.raises(ValueError, match="redistribution allow-list"): + _validate(entry) + + +def test_required_license_keys_constant(): + """Sanity: the required-keys constant matches what the + docstring promises.""" + assert _REQUIRED_LICENSE_KEYS == ("license", "license_url") + + +def test_allowed_licenses_excludes_proprietary_and_nc_nd(): + """Allow-list must not contain proprietary / NC / ND tokens.""" + for forbidden in ["PROPRIETARY", "CC-BY-NC", "CC-BY-NC-SA", + "CC-BY-ND", "CC-BY-NC-ND"]: + assert forbidden not in _ALLOWED_LICENSES + + +# --- cmd_urls ------------------------------------------------------- + + +def _entries_to_jsonl(entries: list[dict]) -> str: + return "\n".join(json.dumps(e) for e in entries) + "\n" + + +def test_cmd_urls_emits_each_url_once(): + """Same URL across two entries → emitted once (set dedup).""" + data = _entries_to_jsonl([ + {"id": "a", "license": "PD", "license_url": "https://l/", + "urls": ["https://example.com/page1", "https://example.com/page2"]}, + {"id": "b", "license": "PD", "license_url": "https://l/", + "urls": ["https://example.com/page1"]}, # dup + ]) + buf = io.StringIO() + with redirect_stdout(buf): + cmd_urls(io.StringIO(data)) + out = buf.getvalue().strip().split("\n") + assert len(out) == 2 # dup collapsed + assert "https://example.com/page1" in out + assert "https://example.com/page2" in out + + +def test_cmd_urls_skips_placeholder_entries(): + """Entries without `urls` (placeholders) emit nothing.""" + data = _entries_to_jsonl([ + {"id": "placeholder", "license": "PD", "license_url": "x"}, + {"id": "real", "license": "PD", "license_url": "x", + "urls": ["https://example.com/r"]}, + ]) + buf = io.StringIO() + with redirect_stdout(buf): + cmd_urls(io.StringIO(data)) + assert buf.getvalue().strip() == "https://example.com/r" + + +def test_cmd_urls_raises_on_invalid_entry(): + """Disallowed-license-with-urls → cmd_urls re-raises.""" + data = _entries_to_jsonl([ + {"id": "bad", "license": "PROPRIETARY", "license_url": "x", + "urls": ["https://example.com/x"]}, + ]) + with pytest.raises(ValueError): + cmd_urls(io.StringIO(data)) + + +# --- cmd_summary ---------------------------------------------------- + + +def test_cmd_summary_groups_by_license_and_domain(): + data = _entries_to_jsonl([ + {"id": "a", "license": "PD", "license_url": "x", + "domain": "logic", "urls": ["https://l/a"]}, + {"id": "b", "license": "MIT", "license_url": "x", + "domain": "logic", "urls": ["https://l/b"]}, + {"id": "c", "license": "PD", "license_url": "x", + "domain": "geometry", "urls": ["https://g/c"]}, + ]) + buf = io.StringIO() + with redirect_stdout(buf): + cmd_summary(io.StringIO(data)) + out = buf.getvalue() + assert "by license" in out + assert "PD" in out + assert "MIT" in out + assert "by domain" in out + assert "logic" in out + assert "geometry" in out + assert "entries: 3" in out + + +# --- cmd_crawl_targets --------------------------------------------- + + +def test_cmd_crawl_targets_emits_tab_separated(): + data = _entries_to_jsonl([ + {"id": "depth-2", "license": "PD", "license_url": "x", + "crawl_url": "https://example.com/seed", + "crawl_depth": 3, "crawl_max": 100}, + {"id": "skip-no-crawl", "license": "PD", "license_url": "x"}, + ]) + buf = io.StringIO() + with redirect_stdout(buf): + cmd_crawl_targets(io.StringIO(data)) + out_lines = buf.getvalue().strip().split("\n") + assert len(out_lines) == 1 # only the entry with crawl_url + fields = out_lines[0].split("\t") + assert fields[0] == "https://example.com/seed" + assert fields[1] == "3" + assert fields[2] == "100" + assert fields[3] == "depth-2" + + +def test_cmd_crawl_targets_uses_default_depth_max(): + """Missing crawl_depth / crawl_max → defaults 2 / 80.""" + data = _entries_to_jsonl([ + {"id": "defaults", "license": "PD", "license_url": "x", + "crawl_url": "https://example.com/seed"}, + ]) + buf = io.StringIO() + with redirect_stdout(buf): + cmd_crawl_targets(io.StringIO(data)) + fields = buf.getvalue().strip().split("\t") + assert fields[1] == "2" # default crawl_depth + assert fields[2] == "80" # default crawl_max + + +# --- cmd_tex_targets ----------------------------------------------- + + +def test_cmd_tex_targets_emits_tex_url_id_pairs(): + data = _entries_to_jsonl([ + {"id": "hilbert-tex", "license": "PD", "license_url": "x", + "tex_url": "https://gutenberg.org/files/17384/17384-t.tex"}, + {"id": "no-tex", "license": "PD", "license_url": "x", + "urls": ["https://example.com/h"]}, + ]) + buf = io.StringIO() + with redirect_stdout(buf): + cmd_tex_targets(io.StringIO(data)) + out_lines = buf.getvalue().strip().split("\n") + assert len(out_lines) == 1 + fields = out_lines[0].split("\t") + assert "hilbert-tex" == fields[1] + assert "17384-t.tex" in fields[0] + + +# --- cmd_ids -------------------------------------------------------- + + +def test_cmd_ids_emits_ingestable_only(): + """Entries with neither urls nor crawl_url are skipped.""" + data = _entries_to_jsonl([ + {"id": "with-urls", "license": "PD", "license_url": "x", + "urls": ["https://l/a"]}, + {"id": "with-crawl", "license": "PD", "license_url": "x", + "crawl_url": "https://l/c"}, + {"id": "placeholder-no-emit", "license": "PD", + "license_url": "x"}, + {"id": "license-fail", "license": "PROPRIETARY", + "license_url": "x", "urls": ["https://l/p"]}, + ]) + buf = io.StringIO() + with redirect_stdout(buf): + cmd_ids(io.StringIO(data)) + out = buf.getvalue().strip().split("\n") + # license-fail and placeholder excluded. + assert "with-urls" in out + assert "with-crawl" in out + assert "placeholder-no-emit" not in out + assert "license-fail" not in out + + +# --- cmd_lookup ---------------------------------------------------- + + +def test_cmd_lookup_emits_seven_field_tsv(monkeypatch): + """lookup emits 7 tab-separated fields incl. author.""" + monkeypatch.setattr(sys, "argv", ["prog", "lookup", "test-id"]) + data = _entries_to_jsonl([ + {"id": "test-id", "license": "PD", "license_url": "x", + "domain": "logic", "author": "Test Author", + "crawl_url": "https://l/seed", + "crawl_depth": 3, "crawl_max": 50}, + ]) + buf = io.StringIO() + with redirect_stdout(buf): + rc = cmd_lookup(io.StringIO(data)) + assert rc == 0 + fields = buf.getvalue().strip().split("\t") + assert len(fields) == 7 + assert fields[0] == "test-id" + assert fields[1] == "https://l/seed" + assert fields[2] == "3" + assert fields[3] == "50" + assert fields[4] == "PD" + assert fields[5] == "logic" + assert fields[6] == "Test Author" + + +def test_cmd_lookup_missing_id_arg_returns_2(monkeypatch): + monkeypatch.setattr(sys, "argv", ["prog", "lookup"]) # no id + err = io.StringIO() + with redirect_stderr(err): + rc = cmd_lookup(io.StringIO("")) + assert rc == 2 + assert "usage" in err.getvalue().lower() + + +def test_cmd_lookup_unknown_id_returns_4(monkeypatch): + monkeypatch.setattr(sys, "argv", ["prog", "lookup", "nonexistent"]) + data = _entries_to_jsonl([ + {"id": "real", "license": "PD", "license_url": "x"}, + ]) + err = io.StringIO() + with redirect_stderr(err): + rc = cmd_lookup(io.StringIO(data)) + assert rc == 4 + assert "not found" in err.getvalue() + + +def test_cmd_lookup_license_fail_returns_3(monkeypatch): + """Lookup of an entry with a disallowed license + emit URLs → + return code 3 (license-fail placeholder).""" + monkeypatch.setattr(sys, "argv", ["prog", "lookup", "fail-id"]) + data = _entries_to_jsonl([ + {"id": "fail-id", "license": "PROPRIETARY", "license_url": "x", + "urls": ["https://l/p"]}, + ]) + err = io.StringIO() + with redirect_stderr(err): + rc = cmd_lookup(io.StringIO(data)) + assert rc == 3 + assert "license-fail" in err.getvalue() + + +def test_cmd_lookup_uses_default_depth_max(monkeypatch): + monkeypatch.setattr(sys, "argv", ["prog", "lookup", "defaults-id"]) + data = _entries_to_jsonl([ + {"id": "defaults-id", "license": "PD", "license_url": "x", + "domain": "math", "crawl_url": "https://l/s"}, + ]) + buf = io.StringIO() + with redirect_stdout(buf): + rc = cmd_lookup(io.StringIO(data)) + assert rc == 0 + fields = buf.getvalue().strip().split("\t") + assert fields[2] == "2" # default crawl_depth + assert fields[3] == "80" # default crawl_max + + +def test_cmd_lookup_blank_author_emits_empty_field(monkeypatch): + """Entry without `author` → 7th field empty (not absent). + The output ends with a trailing tab + empty author field; + Makefile's `cut -f7` correctly returns empty string.""" + monkeypatch.setattr(sys, "argv", ["prog", "lookup", "no-author"]) + data = _entries_to_jsonl([ + {"id": "no-author", "license": "PD", "license_url": "x", + "domain": "math", "crawl_url": "https://l/s"}, + ]) + buf = io.StringIO() + with redirect_stdout(buf): + cmd_lookup(io.StringIO(data)) + # rstrip newlines but preserve trailing tabs / empty fields. + line = buf.getvalue().rstrip("\n") + fields = line.split("\t") + assert len(fields) == 7 + assert fields[6] == "" + + +# --- main dispatch ------------------------------------------------- + + +def test_main_unknown_subcommand_returns_2(monkeypatch): + err = io.StringIO() + with redirect_stderr(err): + rc = main(["prog", "bogus-cmd"]) + assert rc == 2 + assert "usage" in err.getvalue() + + +def test_main_no_subcommand_returns_2(): + err = io.StringIO() + with redirect_stderr(err): + rc = main(["prog"]) + assert rc == 2 + + +def test_main_dispatches_to_subcommand(monkeypatch): + """main(['prog', 'urls']) routes to cmd_urls.""" + data = json.dumps({"id": "x", "license": "PD", + "license_url": "x", "urls": ["https://l/a"]}) + monkeypatch.setattr(sys, "stdin", io.StringIO(data)) + buf = io.StringIO() + with redirect_stdout(buf): + rc = main(["prog", "urls"]) + assert rc == 0 + assert "https://l/a" in buf.getvalue() + + +# --- live manifest --------------------------------------------------- + + +def test_live_manifest_validates_clean(): + """The shipped bench/fixtures/textbooks/manifest-v1.jsonl + must satisfy `_validate()` for every entry. If a future PR adds + an entry with a typo in the license token, this test fires.""" + from pathlib import Path + + manifest = Path("/home/fox/git/arborist/bench/fixtures/" + "textbooks/manifest-v1.jsonl") + assert manifest.exists() + with manifest.open() as f: + for entry in iter_entries(f): + _validate(entry)