Error

.""" result = run_script("jats_to_text.py", fixture("jats_europepmc.xml")) self.assertEqual(result.returncode, 0, result.stderr) self.assertTrue(json.loads(result.stdout)["full_text_available"]) def test_section_filter_matches_case_insensitively(self) -> None: result = run_script("jats_to_text.py", fixture("jats_with_body.xml"), "--sections", "methods") self.assertEqual(result.returncode, 0, result.stderr) payload = json.loads(result.stdout) self.assertEqual(payload["section_count"], 1) def test_unmatched_section_filter_lists_what_is_available(self) -> None: result = run_script("jats_to_text.py", fixture("jats_with_body.xml"), "--sections", "nope") self.assertEqual(result.returncode, 1) self.assertIn("available:", result.stderr) def test_non_jats_xml_is_rejected_with_a_useful_message(self) -> None: result = run_script("jats_to_text.py", fixture("not_jats.xml")) self.assertEqual(result.returncode, 1) self.assertIn("no

element", result.stderr) def test_xref_text_is_dropped_but_the_sentence_survives(self) -> None: """Citation markers must not land mid-sentence, and the tail must stay.""" body = jats_to_text.parse( "Results" "

Growth increased12 after treatment.

" "" ) section = jats_to_text.collect_sections(body)[0] self.assertEqual(section["text"], "Results Growth increased after treatment.") class ArxivTests(unittest.TestCase): def test_entry_is_parsed_with_version_split(self) -> None: result = run_script("arxiv_atom.py", fixture("arxiv_feed.xml")) self.assertEqual(result.returncode, 0, result.stderr) payload = json.loads(result.stdout) entry = payload["entries"][0] self.assertEqual(entry["arxiv_id"], "1706.03762") self.assertEqual(entry["arxiv_id_versioned"], "1706.03762v7") self.assertEqual(entry["version"], "7") def test_pdf_link_is_selected_by_rel_not_position(self) -> None: """The feed's own precedes the entries and must not be picked up.""" result = run_script("arxiv_atom.py", fixture("arxiv_feed.xml")) entry = json.loads(result.stdout)["entries"][0] self.assertEqual(entry["pdf_url"], "https://arxiv.org/pdf/1706.03762v7") self.assertEqual(entry["abstract_url"], "https://arxiv.org/abs/1706.03762v7") self.assertNotIn("api/query", entry["pdf_url"]) def test_hard_wrapped_title_and_abstract_are_collapsed(self) -> None: result = run_script("arxiv_atom.py", fixture("arxiv_feed.xml")) entry = json.loads(result.stdout)["entries"][0] self.assertEqual(entry["title"], "Attention Is All You Need") self.assertNotIn("\n", entry["abstract"]) def test_absent_arxiv_doi_is_none_not_invented(self) -> None: result = run_script("arxiv_atom.py", fixture("arxiv_feed.xml")) entry = json.loads(result.stdout)["entries"][0] self.assertIsNone(entry["doi"]) def test_error_feed_exits_3(self) -> None: """arXiv sends this with HTTP 200 and totalResults 1.""" result = run_script("arxiv_atom.py", fixture("arxiv_error.xml")) self.assertEqual(result.returncode, 3, result.stdout) self.assertIn("start must be an integer", result.stderr) def test_rate_limited_plain_text_body_exits_5(self) -> None: """arXiv answers a throttled caller with 14 bytes of plain text, not XML.""" result = run_script("arxiv_atom.py", fixture("arxiv_rate_limited.txt")) self.assertEqual(result.returncode, 5, result.stdout) self.assertIn("throttling", result.stderr) self.assertIn("three seconds", result.stderr) def test_non_xml_body_reports_what_it_actually_got(self) -> None: result = run_script("arxiv_atom.py", fixture("openalex_work.json")) self.assertEqual(result.returncode, 1) self.assertIn("first 100 bytes", result.stderr) def test_empty_feed_is_reported_as_a_genuine_no_match(self) -> None: result = run_script("arxiv_atom.py", fixture("arxiv_empty.xml")) self.assertEqual(result.returncode, 0, result.stderr) payload = json.loads(result.stdout) self.assertEqual(payload["total_results"], 0) self.assertEqual(payload["returned"], 0) self.assertIn("not found in arXiv", payload["note"]) def test_query_as_executed_is_reported(self) -> None: """An unknown field prefix is silently rewritten to `all:` upstream.""" result = run_script("arxiv_atom.py", fixture("arxiv_rewritten_query.xml")) payload = json.loads(result.stdout) self.assertIn("all:badfield:xyz", payload["query_as_executed"]) def test_ids_only_strips_versions(self) -> None: result = run_script("arxiv_atom.py", fixture("arxiv_feed.xml"), "--ids-only") self.assertEqual(result.stdout.strip(), "1706.03762") def test_split_version_leaves_unversioned_ids_alone(self) -> None: self.assertEqual(arxiv_atom.split_version("2103.15348"), ("2103.15348", None)) self.assertEqual(arxiv_atom.split_version("2103.15348v12"), ("2103.15348", "12")) # Old-style IDs contain letters and a slash but no version. self.assertEqual(arxiv_atom.split_version("hep-th/9901001"), ("hep-th/9901001", None)) def test_id_from_url_is_scheme_agnostic(self) -> None: self.assertEqual(arxiv_atom.id_from_url("http://arxiv.org/abs/1706.03762v7"), "1706.03762v7") self.assertEqual(arxiv_atom.id_from_url("https://arxiv.org/abs/1706.03762v7"), "1706.03762v7") class OpenAlexAbstractTests(unittest.TestCase): def test_abstract_is_reconstructed_in_position_order(self) -> None: text, anomalies = openalex_abstract.reconstruct( {"Despite": [0], "growing": [1], "interest": [2], "in": [3], "OA": [4]} ) self.assertEqual(text, "Despite growing interest in OA") self.assertEqual(anomalies, []) def test_duplicate_positions_keep_both_tokens(self) -> None: """The naive {position: word} inversion silently drops one of these.""" text, anomalies = openalex_abstract.reconstruct({"alpha": [0], "beta": [0], "gamma": [1]}) self.assertIn("alpha", text) self.assertIn("beta", text) self.assertIn("gamma", text) self.assertTrue(any("more than one token" in note for note in anomalies)) def test_gaps_in_the_index_are_reported(self) -> None: _, anomalies = openalex_abstract.reconstruct({"first": [0], "last": [5]}) self.assertTrue(any("absent from the index" in note for note in anomalies)) def test_non_integer_positions_are_reported_not_crashed_on(self) -> None: text, anomalies = openalex_abstract.reconstruct({"word": ["x"], "real": [0]}) self.assertEqual(text, "real") self.assertTrue(any("non-integer position" in note for note in anomalies)) def test_empty_index_yields_empty_text(self) -> None: self.assertEqual(openalex_abstract.reconstruct({}), ("", [])) def test_single_work_payload(self) -> None: result = run_script("openalex_abstract.py", fixture("openalex_work.json")) self.assertEqual(result.returncode, 0, result.stderr) payload = json.loads(result.stdout) self.assertEqual(payload["count"], 1) self.assertEqual(payload["with_abstract"], 1) self.assertIn("Open Access", payload["works"][0]["abstract"]) def test_list_response_payload(self) -> None: result = run_script("openalex_abstract.py", fixture("openalex_list.json")) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(json.loads(result.stdout)["count"], 2) def test_missing_index_distinguishes_no_abstract_from_select_exclusion(self) -> None: result = run_script("openalex_abstract.py", fixture("openalex_no_abstract.json")) self.assertEqual(result.returncode, 0, result.stderr) work = json.loads(result.stdout)["works"][0] self.assertIsNone(work["abstract"]) self.assertIn("`select=`", work["abstract_warnings"][0]) def test_text_only_fails_loudly_when_nothing_reconstructs(self) -> None: result = run_script("openalex_abstract.py", fixture("openalex_no_abstract.json"), "--text-only") self.assertEqual(result.returncode, 1) self.assertIn("no abstracts", result.stderr) def test_works_from_rejects_a_bare_scalar(self) -> None: with self.assertRaises(_common.InputError): openalex_abstract.works_from("not a work") class PaginateAdapterTests(unittest.TestCase): """Adapter unit tests. The network walk is covered by manual verification; what matters here is the arithmetic that made the walk wrong before.""" def test_biorxiv_steps_by_the_reported_count_not_100(self) -> None: payload = { "messages": [{"status": "ok", "cursor": 0, "count": 30, "total": "360"}], "collection": [{"doi": f"10.1101/x{i}"} for i in range(30)], } page = paginate._rxiv_parse(payload, 0) self.assertEqual(page.total, 360) self.assertEqual(page.next_state, 30, "stepping by 100 would skip records 30-99") self.assertEqual(len(page.records), 30) def test_biorxiv_pubs_steps_by_100(self) -> None: payload = { "messages": [{"status": "ok", "cursor": 0, "count": 100, "total": "124"}], "collection": [{"preprint_doi": f"10.1101/x{i}"} for i in range(100)], } self.assertEqual(paginate._rxiv_parse(payload, 0).next_state, 100) def test_biorxiv_stops_when_the_next_offset_reaches_total(self) -> None: payload = { "messages": [{"status": "ok", "cursor": 330, "count": 30, "total": "360"}], "collection": [{"doi": f"10.1101/x{i}"} for i in range(30)], } self.assertIsNone(paginate._rxiv_parse(payload, 330).next_state) def test_biorxiv_reports_the_count_mismatch_it_falls_back_from(self) -> None: payload = { "messages": [{"status": "ok", "cursor": 0, "count": 30, "total": "360"}], "collection": [{"doi": "10.1101/x"}], } page = paginate._rxiv_parse(payload, 0) self.assertEqual(page.next_state, 1) self.assertTrue(any("but returned" in note for note in page.notes or [])) def test_biorxiv_no_articles_found_is_surfaced(self) -> None: """HTTP 200 with an empty collection, indistinguishable without status.""" page = paginate._rxiv_parse( {"messages": [{"status": "no articles found"}], "collection": []}, 0 ) self.assertEqual(page.records, []) self.assertIsNone(page.next_state) self.assertTrue(any("no articles found" in note for note in page.notes or [])) def test_biorxiv_endpoint_without_counts_reports_no_total(self) -> None: """DOI and N-most-recent lookups omit count/total entirely.""" page = paginate._rxiv_parse( {"messages": [{"status": "ok", "category": "all"}], "collection": [{"doi": "10.1101/x"}]}, 0, ) self.assertIsNone(page.total) self.assertEqual(page.next_state, 1) def test_biorxiv_url_selects_details_or_pubs(self) -> None: build = paginate._rxiv_url("biorxiv") self.assertIn("/details/biorxiv/2024-01-01/2024-01-03/0/json", build("2024-01-01/2024-01-03", 0, 100)) self.assertIn("/pubs/biorxiv/2024-01-01/2024-01-03/30/json", build("pubs:2024-01-01/2024-01-03", 30, 100)) def test_medrxiv_never_uses_the_api_medrxiv_host(self) -> None: url = paginate._rxiv_url("medrxiv")("2024-01-01/2024-01-03", 0, 100) self.assertIn("api.biorxiv.org", url) self.assertNotIn("api.medrxiv.org", url) def test_europepmc_stops_on_an_echoed_cursor(self) -> None: """There is no null terminator: exhaustion echoes your own cursor back.""" page = paginate._europepmc_parse( {"hitCount": 19, "nextCursorMark": "ABC", "resultList": {"result": []}}, "ABC" ) self.assertIsNone(page.next_state) self.assertEqual(page.total, 19) def test_europepmc_continues_on_a_new_cursor(self) -> None: page = paginate._europepmc_parse( {"hitCount": 19, "nextCursorMark": "DEF", "resultList": {"result": [{"id": "1"}]}}, "ABC" ) self.assertEqual(page.next_state, "DEF") def test_europepmc_errcode_in_a_200_body_raises(self) -> None: with self.assertRaises(RuntimeError) as caught: paginate._europepmc_parse( {"errCode": 404, "errMsg": "Invalid page size provided"}, "*" ) self.assertIn("Invalid page size", str(caught.exception)) def test_europepmc_reports_the_query_as_parsed(self) -> None: page = paginate._europepmc_parse( { "hitCount": 1, "nextCursorMark": "B", "request": {"queryString": 'SRC:"PPR"'}, "resultList": {"result": [{"id": "PPR1"}]}, }, "*", ) self.assertTrue(any("as parsed" in note for note in page.notes or [])) def test_openalex_stops_on_an_empty_page(self) -> None: page = paginate._openalex_parse({"meta": {"count": 5, "next_cursor": "X"}, "results": []}, "*") self.assertIsNone(page.next_state) def test_openalex_surfaces_the_reported_cost(self) -> None: page = paginate._openalex_parse( {"meta": {"count": 5, "next_cursor": "X", "cost_usd": 0.001}, "results": [{"id": "W1"}]}, "*", ) self.assertTrue(any("cost_usd" in note for note in page.notes or [])) def test_crossref_stops_on_an_empty_page(self) -> None: page = paginate._crossref_parse( {"message": {"total-results": 7, "next-cursor": "X", "items": []}}, "*" ) self.assertIsNone(page.next_state) self.assertEqual(page.total, 7) def test_every_api_declares_a_delay_and_a_query_format(self) -> None: for name, api in paginate.APIS.items(): with self.subTest(api=name): self.assertGreater(api.delay, 0, "an API with no delay would be parallelized by accident") self.assertTrue(api.note) def test_dry_run_makes_no_request(self) -> None: result = run_script( "paginate.py", "--api", "biorxiv", "--query", "2024-01-01/2024-01-03", "--dry-run" ) self.assertEqual(result.returncode, 0, result.stderr) payload = json.loads(result.stdout) self.assertTrue(payload["first_url"].startswith("https://api.biorxiv.org/details/")) def test_list_apis_needs_no_other_argument(self) -> None: result = run_script("paginate.py", "--list-apis") self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(sorted(json.loads(result.stdout)), sorted(paginate.APIS)) def test_missing_required_arguments_exit_2(self) -> None: result = run_script("paginate.py", "--api", "biorxiv") self.assertEqual(result.returncode, 2) self.assertIn("--list-apis", result.stderr) def test_nonpositive_bounds_are_rejected(self) -> None: for flag in ("--max-records", "--max-calls", "--page-size"): with self.subTest(flag=flag): result = run_script( "paginate.py", "--api", "biorxiv", "--query", "5", flag, "0", "--dry-run" ) self.assertEqual(result.returncode, 1) class StructureTests(unittest.TestCase): """Repo rules that apply to this skill specifically.""" def test_frontmatter_conforms(self) -> None: self.assertEqual(skill_contract.structure.frontmatter_problems(SKILL_ROOT), []) def test_skill_md_is_within_the_line_budget(self) -> None: self.assertEqual(skill_contract.structure.length_problems(SKILL_ROOT), []) def test_no_stray_tests_under_the_skill(self) -> None: self.assertEqual(skill_contract.structure.stray_test_problems(SKILL_ROOT), []) def test_no_bytecode_shipped(self) -> None: """Running the scripts by hand during development leaves __pycache__ behind.""" self.assertEqual(skill_contract.structure.bytecode_problems(SKILL_ROOT), []) def test_referenced_paths_exist(self) -> None: self.assertEqual(skill_contract.structure.link_problems(SKILL_ROOT), []) def test_scripts_compile(self) -> None: self.assertEqual(skill_contract.structure.compile_problems(SKILL_ROOT), []) def test_no_dynamic_execution(self) -> None: self.assertEqual(skill_contract.structure.dynamic_execution_problems(SKILL_ROOT), []) def test_no_leftover_tool_call_markup(self) -> None: """A previous release shipped a stray at EOF.""" for path in sorted(SKILL_ROOT.rglob("*.md")): with self.subTest(path=path.name): text = path.read_text(encoding="utf-8") for marker in ("", "", ""): self.assertNotIn(marker, text) def test_every_reference_file_is_linked_from_skill_md(self) -> None: skill_md = (SKILL_ROOT / "SKILL.md").read_text(encoding="utf-8") for reference in sorted((SKILL_ROOT / "references").glob("*.md")): with self.subTest(reference=reference.name): self.assertIn(f"references/{reference.name}", skill_md) def test_every_script_is_documented_in_skill_md(self) -> None: skill_md = (SKILL_ROOT / "SKILL.md").read_text(encoding="utf-8") for script in sorted(SCRIPTS.glob("*.py")): if script.name == "_common.py": continue with self.subTest(script=script.name): self.assertIn(f"scripts/{script.name}", skill_md) def test_paginate_covers_every_api_it_advertises(self) -> None: skill_md = (SKILL_ROOT / "SKILL.md").read_text(encoding="utf-8") for name in paginate.APIS: with self.subTest(api=name): self.assertIn(name, skill_md.lower()) if __name__ == "__main__": unittest.main()