fix: CMYK image drop, empty-doc crash, page_index shadowing, sqlite hardening, flaky tests

Addresses items 9-13 and a/b/c/f from the max-effort review of PR #272.

- pdf.py: image colorspace check was `pix.n > 4`, which treats CMYK-without-
  alpha (n==4, same as RGBA) as not needing RGB conversion; pix.save() as .png
  then raises "unsupported colorspace", silently dropped by the surrounding
  except. Fixed to `pix.n - pix.alpha >= 4` (correctly converts CMYK, leaves
  RGBA untouched).

- pipeline.py: detect_strategy([]) (an empty/whitespace-only source file)
  returned "content_based", routing into the PDF-oriented TOC-detection
  pipeline -- wasting a real LLM call before raising IndexingError. Empty
  node lists now route to level_based, whose build_tree_from_levels([])
  returns an empty structure instantly with zero LLM calls.

- page_index.py (shim): pageindex/__init__.py binds the canonical `page_index`
  function as the package attribute, but this file is ALSO a real submodule
  of the same name -- importing it anywhere (import machinery, unconditional)
  overwrites that attribute with the module object, breaking
  `from pageindex import page_index; page_index(x)` for the rest of the
  process. Made the shim module itself callable (delegates to the real
  function via a ModuleType subclass), so whichever object ends up in that
  slot is callable regardless of import order.

- storage/sqlite.py: create_collection let a raw sqlite3.IntegrityError escape
  on a duplicate name (new CollectionAlreadyExistsError); the collections
  table's CHECK constraint only validated the name's first character (GLOB
  '*' is a wildcard, not a regex quantifier over the preceding class) --
  fixed to validate the whole string, and SQLiteStorage now also validates in
  Python (it's a public StorageEngine usable directly, bypassing
  LocalBackend's own check).

- tests/test_review_fixes_2.py: two tests used a ContentNode with no `level`
  set, so build_index took the content_based path and made real (retried,
  slow, and -- with a valid key -- billable) LLM calls instead of testing the
  text-stripping logic they claimed to. Mocked out _content_based_pipeline.

- retrieve.py: _parse_pages/_get_pdf_page_content were independent copies of
  the canonical parse_pages/get_pdf_page_content that had already drifted
  (missing the p>=1 filter and 1000-page DoS cap) -- delegate to canonical
  now, so the legacy pageindex.get_page_content path can't silently regress
  again.

- parser/markdown.py: a leading UTF-8 BOM broke first-header detection
  (not whitespace, .strip() doesn't remove it) -- decode utf-8-sig. Only
  backtick fences were recognized as code blocks, so a '#'-prefixed line
  inside a ~~~-fenced block (valid CommonMark) was misparsed as a heading --
  recognize both fence styles.

- run_pageindex.py: --if-thinning wasn't migrated to the bare-flag +
  legacy-yes/no convention the other four --if-add-* flags got; bare usage
  raised an argparse error and it never went through the shared coercion.

- types.py: DocumentDetail's `structure` field was inside the class's
  total=False body, so TypedDict rules made it optional even though every
  backend always populates it. Split into a required base class.

Adds regression tests for all of the above. Full suite: 244 passed, 2 skipped
(one pre-existing, unrelated flaky cloud-streaming test).

Claude-Session: https://claude.ai/code/session_01Kx5DgKbhK1N8autqXH8SmS
This commit is contained in:
mountain 2026-07-09 11:58:59 +08:00
parent b9d021916f
commit 4e6a13576d
18 changed files with 368 additions and 36 deletions

View file

@ -19,6 +19,60 @@ def test_delete_collection(storage):
storage.delete_collection("papers")
assert "papers" not in storage.list_collections()
def test_create_duplicate_collection_raises_pageindex_error(storage):
"""A raw sqlite3.IntegrityError leaking out breaks `except PageIndexError`
catch-alls; must be translated to a proper SDK exception."""
from pageindex.errors import CollectionAlreadyExistsError, PageIndexError
storage.create_collection("papers")
with pytest.raises(CollectionAlreadyExistsError):
storage.create_collection("papers")
# also catchable via the SDK's generic base class
storage.create_collection("other")
with pytest.raises(PageIndexError):
storage.create_collection("other")
@pytest.mark.parametrize("bad_name", [
"a/../../etc/passwd", "/etc/passwd", "a$(whoami)", ".hidden",
"a b", "válid", "", "a" * 129,
])
def test_create_collection_rejects_invalid_names_at_the_python_layer(storage, bad_name):
"""SQLiteStorage must validate collection names itself — it's a public
StorageEngine that can be used directly, bypassing LocalBackend's own
regex check entirely."""
from pageindex.errors import PageIndexError
with pytest.raises(PageIndexError):
storage.create_collection(bad_name)
def test_sql_check_constraint_also_rejects_invalid_names_directly(storage):
"""Defense-in-depth: even bypassing SQLiteStorage's own Python validation
and inserting via raw SQL, the schema's CHECK constraint must reject a
name that isn't ENTIRELY [a-zA-Z0-9_-] — not just its first character
(GLOB '*' is a wildcard, not a regex quantifier over the preceding class,
so 'name GLOB [a-zA-Z0-9_-]*' alone only constrains the first character)."""
import sqlite3
conn = storage._get_conn()
with pytest.raises(sqlite3.IntegrityError):
conn.execute("INSERT INTO collections (name) VALUES (?)", ("a/../../etc/passwd",))
def test_malicious_collection_name_rejected_through_local_backend_too(tmp_path):
"""End-to-end via the normal LocalBackend entry point: a path-traversal-
shaped collection name must never reach add_document's
files_dir / collection path construction. Three independent layers now
reject it (LocalBackend's own regex, SQLiteStorage's regex, and the SQL
CHECK constraint) this pins the outermost one."""
from pageindex.backend.local import LocalBackend
from pageindex.errors import PageIndexError
storage = SQLiteStorage(str(tmp_path / "t.db"))
backend = LocalBackend(storage=storage, files_dir=str(tmp_path / "files"), model="gpt-4o")
with pytest.raises(PageIndexError):
backend.create_collection("a/../../escape_me")
assert not (tmp_path / "escape_me").exists()
def test_save_and_get_document(storage):
storage.create_collection("papers")
doc = {