wehub-resource-sync
409 lines
15 KiB
Python
409 lines
15 KiB
Python
"""Tests for parameterised parser hints (``[-R(chunk_ts=800,chunk_ol=80)]``).
|
|
|
|
Phase 1 wires only ``chunk_token_size``/``chunk_ts`` and
|
|
``chunk_overlap_token_size``/``chunk_ol`` through the existing per-document
|
|
``chunk_options`` channel. ``parser_rules=""`` is passed explicitly so the
|
|
assertions are independent of any ambient ``LIGHTRAG_PARSER``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib
|
|
import sys
|
|
|
|
import pytest
|
|
|
|
from lightrag.parser.param_schema import (
|
|
parse_chunk_params,
|
|
split_top_level,
|
|
take_paren_block,
|
|
)
|
|
from lightrag.parser.routing import (
|
|
FilenameParserHintError,
|
|
ParserRoutingConfigError,
|
|
canonicalize_parser_hinted_basename,
|
|
filename_parser_directives,
|
|
resolve_file_parser_directives,
|
|
resolve_parser_directives,
|
|
validate_parser_routing_config,
|
|
)
|
|
|
|
# Importing document_routes runs an argparse over sys.argv at import time;
|
|
# neutralise pytest's argv during that first import (subsequent imports hit the
|
|
# module cache and skip argparse). Mirrors tests/api/routes/test_document_routes_chunking.py.
|
|
_original_argv = sys.argv[:]
|
|
sys.argv = [sys.argv[0]]
|
|
importlib.import_module("lightrag.api.routers.document_routes")
|
|
sys.argv = _original_argv
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# param_schema: scanners
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_split_top_level_protects_commas_inside_parens():
|
|
# A comma inside a parameter block must not split the surrounding rules.
|
|
assert split_top_level(
|
|
"pdf:legacy-R(chunk_ts=800,chunk_ol=80);*:legacy-R", ";,"
|
|
) == ["pdf:legacy-R(chunk_ts=800,chunk_ol=80)", "*:legacy-R"]
|
|
# Legacy comma rule separator still works at depth 0.
|
|
assert split_top_level("a:native-F,b:legacy-R", ";,") == [
|
|
"a:native-F",
|
|
"b:legacy-R",
|
|
]
|
|
|
|
|
|
def test_take_paren_block():
|
|
assert take_paren_block("R(chunk_ts=800)", 1) == ("chunk_ts=800", 15)
|
|
assert take_paren_block("R", 1) == (None, 1) # no block
|
|
assert take_paren_block("R(unterminated", 1) == (None, 1) # unbalanced
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# param_schema: parse_chunk_params (alias / type / target / repeat)
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_parse_chunk_params_alias_normalised_to_canonical():
|
|
parsed, errors = parse_chunk_params(
|
|
"chunk_ts=800,chunk_ol=80", selector="R", label="x"
|
|
)
|
|
assert errors == []
|
|
assert parsed == {"chunk_token_size": 800, "chunk_overlap_token_size": 80}
|
|
|
|
|
|
def test_parse_chunk_params_canonical_names_accepted():
|
|
parsed, errors = parse_chunk_params(
|
|
"chunk_token_size=1500", selector="V", label="x"
|
|
)
|
|
assert errors == [] and parsed == {"chunk_token_size": 1500}
|
|
|
|
|
|
def test_parse_chunk_params_overlap_rejected_for_vector():
|
|
parsed, errors = parse_chunk_params("chunk_ol=80", selector="V", label="x")
|
|
assert parsed == {}
|
|
assert errors and "not supported for chunk strategy 'V'" in errors[0]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"block,expected",
|
|
[
|
|
("drop_rf=true", True),
|
|
("drop_references=true", True),
|
|
("drop_rf=false", False),
|
|
("drop_rf=yes", True),
|
|
("drop_rf=0", False),
|
|
],
|
|
)
|
|
def test_parse_chunk_params_drop_references_bool(block, expected):
|
|
parsed, errors = parse_chunk_params(block, selector="P", label="x")
|
|
assert errors == []
|
|
assert parsed == {"drop_references": expected}
|
|
|
|
|
|
def test_parse_chunk_params_drop_references_invalid_bool():
|
|
parsed, errors = parse_chunk_params("drop_rf=maybe", selector="P", label="x")
|
|
assert parsed == {}
|
|
assert errors and "must be a boolean" in errors[0]
|
|
|
|
|
|
@pytest.mark.parametrize("block", ["drop_rf", "drop_references"])
|
|
def test_parse_chunk_params_drop_references_bare_flag_is_true(block):
|
|
# A bare boolean flag is shorthand for ``=true``.
|
|
parsed, errors = parse_chunk_params(block, selector="P", label="x")
|
|
assert errors == []
|
|
assert parsed == {"drop_references": True}
|
|
|
|
|
|
@pytest.mark.parametrize("selector", ["F", "R", "V"])
|
|
def test_parse_chunk_params_drop_references_rejected_off_paragraph(selector):
|
|
parsed, errors = parse_chunk_params("drop_rf=true", selector=selector, label="x")
|
|
assert parsed == {}
|
|
assert errors and f"not supported for chunk strategy {selector!r}" in errors[0]
|
|
|
|
|
|
def test_filename_hint_drop_references_extracted():
|
|
d = resolve_parser_directives("notes.[-P(drop_rf=true)].md", parser_rules="")
|
|
assert d.process_options == "P" # pure selector, no params
|
|
assert d.chunk_params == {"P": {"drop_references": True}}
|
|
|
|
|
|
def test_filename_hint_drop_references_bare_flag():
|
|
# ``P(drop_rf)`` with no value is shorthand for ``drop_rf=true``.
|
|
d = resolve_parser_directives("notes.[-P(drop_rf)].md", parser_rules="")
|
|
assert d.process_options == "P"
|
|
assert d.chunk_params == {"P": {"drop_references": True}}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"block,needle",
|
|
[
|
|
("chunk_ts=abc", "must be an integer"),
|
|
("chunk_ts=0", "must be >= 1"),
|
|
("foo=1", "unknown parameter 'foo'"),
|
|
("chunk_ts=1,chunk_ts=2", "may not be repeated"),
|
|
("correct_tl", "unknown parameter 'correct_tl'"),
|
|
# A non-boolean parameter cannot be written bare (only flags can).
|
|
("chunk_ts", "must be written as 'key=value'"),
|
|
("", "empty parameter"),
|
|
# Cross-field invariant: explicit overlap >= size is rejected here, so
|
|
# both rule and filename-hint validation reject it (not just enqueue).
|
|
("chunk_ts=50,chunk_ol=100", "must be < chunk_token_size"),
|
|
("chunk_ts=100,chunk_ol=100", "must be < chunk_token_size"),
|
|
],
|
|
)
|
|
def test_parse_chunk_params_errors(block, needle):
|
|
_parsed, errors = parse_chunk_params(block, selector="F", label="x")
|
|
assert errors and any(needle in e for e in errors)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# routing: filename hint params
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_filename_hint_selector_stays_pure_and_params_extracted():
|
|
d = resolve_parser_directives(
|
|
"notes.[-R(chunk_ts=800,chunk_ol=80)].md", parser_rules=""
|
|
)
|
|
assert d.engine == "legacy"
|
|
assert d.process_options == "R" # pure selector, no params
|
|
assert d.chunk_params == {
|
|
"R": {"chunk_token_size": 800, "chunk_overlap_token_size": 80}
|
|
}
|
|
assert d.engine_params == {}
|
|
|
|
|
|
def test_back_compat_two_tuple_and_directives_unaffected_by_params():
|
|
# The legacy 2-tuple accessor must keep returning (engine, pure-selector).
|
|
assert resolve_file_parser_directives(
|
|
"notes.[-R(chunk_ts=800)].md", parser_rules=""
|
|
) == ("legacy", "R")
|
|
assert filename_parser_directives("notes.[-R(chunk_ts=800)].md") == (None, "R")
|
|
# A param-free hint is completely unchanged.
|
|
assert filename_parser_directives("paper.[native-iet].docx") == ("native", "iet")
|
|
|
|
|
|
def test_canonicalize_strips_parameterised_hint():
|
|
assert (
|
|
canonicalize_parser_hinted_basename("notes.[-R(chunk_ts=800,chunk_ol=80)].md")
|
|
== "notes.md"
|
|
)
|
|
|
|
|
|
def test_engine_params_rejected_for_engine_that_declares_none():
|
|
# Chunk params on a chunk char still work...
|
|
d = resolve_parser_directives("doc.[native-F(chunk_ts=900)].docx", parser_rules="")
|
|
assert d.chunk_params == {"F": {"chunk_token_size": 900}}
|
|
# ...but native declares no engine params, so an engine-level block on it is
|
|
# a hard error on ingestion (Phase 2: only mineru/docling accept params).
|
|
with pytest.raises(FilenameParserHintError, match="does not accept parameters"):
|
|
resolve_file_parser_directives(
|
|
"doc.[native(do_ocr=true)-F].docx", parser_rules=""
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"name",
|
|
[
|
|
"x.[-V(chunk_ol=80)].md", # overlap invalid for V
|
|
"x.[-R(chunk_ts=abc)].md", # non-integer
|
|
"x.[native-R(foo=1)].md", # unknown parameter
|
|
"x.[-R(chunk_ts=50,chunk_ol=100)].md", # explicit overlap >= size
|
|
],
|
|
)
|
|
def test_upload_validation_raises_on_bad_filename_params(name):
|
|
with pytest.raises(FilenameParserHintError):
|
|
resolve_file_parser_directives(name, parser_rules="")
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# routing: LIGHTRAG_PARSER rule params + overlay merge
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_rule_params_with_semicolon_separator_and_comma_in_parens():
|
|
d = resolve_parser_directives(
|
|
"a.md", parser_rules="md:legacy-R(chunk_ts=800,chunk_ol=80);*:legacy-R"
|
|
)
|
|
assert d.engine == "legacy"
|
|
assert d.chunk_params == {
|
|
"R": {"chunk_token_size": 800, "chunk_overlap_token_size": 80}
|
|
}
|
|
|
|
|
|
def test_rule_params_with_comma_separator_and_comma_in_parens():
|
|
# Rule splitting is parenthesis-aware: a comma inside a parameter block is
|
|
# NOT a rule separator, so ',' still separates rules even when a rule carries
|
|
# parameters (the docs recommend ';' but ',' must keep working).
|
|
assert split_top_level(
|
|
"md:legacy-R(chunk_ts=800,chunk_ol=80),*:legacy-R", ";,"
|
|
) == ["md:legacy-R(chunk_ts=800,chunk_ol=80)", "*:legacy-R"]
|
|
d = resolve_parser_directives(
|
|
"a.md", parser_rules="md:legacy-R(chunk_ts=800,chunk_ol=80),*:legacy-R"
|
|
)
|
|
assert d.engine == "legacy"
|
|
assert d.chunk_params == {
|
|
"R": {"chunk_token_size": 800, "chunk_overlap_token_size": 80}
|
|
}
|
|
|
|
|
|
def test_overlay_rule_then_filename_hint_wins_per_key():
|
|
# Design worked example: rule supplies chunk_ol on P, the filename hint
|
|
# supplies chunk_ts on P; the surviving selector is the filename's "P".
|
|
d = resolve_parser_directives(
|
|
"paper.[-P(chunk_ts=3000)].pdf", parser_rules="pdf:legacy-iteP(chunk_ol=100)"
|
|
)
|
|
assert d.process_options == "P" # filename options wholesale-override rule
|
|
assert d.chunk_params["P"] == {
|
|
"chunk_overlap_token_size": 100, # inherited from the rule
|
|
"chunk_token_size": 3000, # from the filename hint
|
|
}
|
|
|
|
|
|
def test_filename_hint_key_overrides_rule_key():
|
|
d = resolve_parser_directives(
|
|
"paper.[-R(chunk_ts=500)].md",
|
|
parser_rules="md:legacy-R(chunk_ts=900,chunk_ol=50)",
|
|
)
|
|
assert d.chunk_params["R"] == {
|
|
"chunk_token_size": 500, # filename wins over the rule's 900
|
|
"chunk_overlap_token_size": 50, # inherited from the rule
|
|
}
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# routing: startup validation
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_validate_parser_routing_config_accepts_good_params():
|
|
validate_parser_routing_config("pdf:legacy-R(chunk_ts=800,chunk_ol=80);*:legacy-R")
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"rules,needle",
|
|
[
|
|
("pdf:legacy-V(chunk_ol=80)", "not supported for chunk strategy 'V'"),
|
|
("pdf:legacy-R(chunk_ts=abc)", "must be an integer"),
|
|
("pdf:legacy-R(foo=1)", "unknown parameter 'foo'"),
|
|
# Cross-field invariant: explicit overlap >= size is rejected at startup
|
|
# so a global rule cannot make every matching upload fail at enqueue.
|
|
("pdf:legacy-R(chunk_ts=50,chunk_ol=100)", "must be < chunk_token_size"),
|
|
("pdf:legacy-R(chunk_ts=100,chunk_ol=100)", "must be < chunk_token_size"),
|
|
],
|
|
)
|
|
def test_validate_parser_routing_config_rejects_bad_params(rules, needle):
|
|
with pytest.raises(ParserRoutingConfigError, match=needle):
|
|
validate_parser_routing_config(rules)
|
|
|
|
|
|
def test_validate_parser_routing_config_allows_one_sided_overlap():
|
|
# Only overlap is explicit: the effective size depends on env / addon_params
|
|
# and is unknown at config time, so startup accepts it and defers to the
|
|
# upload-time effective-overlap check on the resolved snapshot.
|
|
validate_parser_routing_config("pdf:legacy-R(chunk_ol=100);*:legacy-R")
|
|
|
|
|
|
def test_legacy_comma_rules_without_params_still_valid():
|
|
# Pure back-compat: legacy comma-separated, param-free rules unchanged.
|
|
validate_parser_routing_config("docx:native-iteP,pdf:legacy-R,*:legacy-R")
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# upload path: filename-hint params reach the enqueued chunk_options
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def _recording_rag():
|
|
import asyncio # noqa: F401 - imported for clarity at call sites
|
|
from types import SimpleNamespace
|
|
|
|
captured: dict = {}
|
|
|
|
async def fake_enqueue(content, **kwargs):
|
|
captured.update(kwargs)
|
|
return "track-xyz" # non-None -> success path
|
|
|
|
async def fake_error(files, track_id):
|
|
captured["error_files"] = files
|
|
|
|
rag = SimpleNamespace(
|
|
addon_params={},
|
|
apipeline_enqueue_documents=fake_enqueue,
|
|
apipeline_enqueue_error_documents=fake_error,
|
|
)
|
|
return rag, captured
|
|
|
|
|
|
def test_upload_path_applies_hint_chunk_params(tmp_path):
|
|
import asyncio
|
|
|
|
from lightrag.api.routers.document_routes import pipeline_enqueue_file
|
|
|
|
rag, captured = _recording_rag()
|
|
f = tmp_path / "notes.[-R(chunk_ts=800,chunk_ol=80)].md"
|
|
f.write_text("hello world", encoding="utf-8")
|
|
|
|
ok, _track = asyncio.run(pipeline_enqueue_file(rag, f))
|
|
|
|
assert ok is True
|
|
assert "error_files" not in captured
|
|
assert captured["parse_engine"] == "legacy"
|
|
assert captured["process_options"] == "R" # pure selector forwarded
|
|
sub = captured["chunk_options"]["recursive_character"]
|
|
assert sub["chunk_token_size"] == 800
|
|
assert sub["chunk_overlap_token_size"] == 80
|
|
|
|
|
|
def test_upload_path_without_params_omits_chunk_options(tmp_path):
|
|
import asyncio
|
|
|
|
from lightrag.api.routers.document_routes import pipeline_enqueue_file
|
|
|
|
rag, captured = _recording_rag()
|
|
f = tmp_path / "plain.[native-R].md"
|
|
f.write_text("hello world", encoding="utf-8")
|
|
|
|
ok, _track = asyncio.run(pipeline_enqueue_file(rag, f))
|
|
|
|
assert ok is True
|
|
# Legacy behaviour: no per-file chunk_options when the hint has no params.
|
|
assert "chunk_options" not in captured
|
|
|
|
|
|
def test_upload_path_rejects_bad_hint_params(tmp_path):
|
|
import asyncio
|
|
|
|
from lightrag.api.routers.document_routes import pipeline_enqueue_file
|
|
|
|
rag, captured = _recording_rag()
|
|
f = tmp_path / "bad.[-V(chunk_ol=80)].md"
|
|
f.write_text("hello world", encoding="utf-8")
|
|
|
|
ok, _track = asyncio.run(pipeline_enqueue_file(rag, f))
|
|
|
|
assert ok is False
|
|
assert captured["error_files"] # an error document was enqueued
|
|
assert "chunk_options" not in captured
|
|
|
|
|
|
def test_upload_path_rejects_effective_overlap_violation(tmp_path):
|
|
import asyncio
|
|
|
|
from lightrag.api.routers.document_routes import pipeline_enqueue_file
|
|
|
|
rag, captured = _recording_rag()
|
|
# An explicit overlap (100) >= size (50) pair is rejected up front by
|
|
# filename-hint validation (uniform with rule validation), so the file
|
|
# never enqueues and the overlap message is surfaced in the error document.
|
|
f = tmp_path / "tiny.[-R(chunk_ts=50,chunk_ol=100)].md"
|
|
f.write_text("hello world", encoding="utf-8")
|
|
|
|
ok, _track = asyncio.run(pipeline_enqueue_file(rag, f))
|
|
|
|
assert ok is False
|
|
assert captured["error_files"]
|
|
assert "must be < chunk_token_size" in captured["error_files"][0]["original_error"]
|