torben
bb097f614b
Files changed: - AGENTS.md - CHANGES.md - VERSION - instructions/wiki-query/SKILL.md - tools/CONTRACT.md - tools/chemenu/api.py - tools/chemenu/commands/search.py - tools/chemenu/mcp/server.py - tools/chemenu/search/service.py - tools/chemenu/search/types.py - tools/chemenu/tests/test_api.py - tools/chemenu/tests/test_mcp_server.py - tools/chemenu/tests/test_search.py
445 lines
17 KiB
Python
445 lines
17 KiB
Python
import subprocess
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from chemenu.commands.search import (
|
|
load_pages_by_path,
|
|
render_table,
|
|
run_search,
|
|
sort_hits,
|
|
unreadable_pages,
|
|
)
|
|
from chemenu.search import filters, ripgrep
|
|
from chemenu.search.base import page_key
|
|
from chemenu.search.filters import PredicateError, parse_predicate
|
|
from chemenu.search.fuse import reciprocal_rank_fusion
|
|
from chemenu.search.registry import UnknownBackend, resolve
|
|
from chemenu.search.ripgrep import RipgrepBackend, build_argv
|
|
from chemenu.search.types import (
|
|
DEFAULT_LIMIT,
|
|
Match,
|
|
Predicate,
|
|
SearchHit,
|
|
SearchQuery,
|
|
SearchResult,
|
|
)
|
|
from chemenu.frontmatter_io import write_page
|
|
|
|
|
|
@pytest.fixture
|
|
def pages(kb_dir: Path, tmp_path: Path):
|
|
return load_pages_by_path(kb_dir, tmp_path)
|
|
|
|
|
|
@pytest.fixture
|
|
def backend(kb_dir: Path, tmp_path: Path):
|
|
return RipgrepBackend(search_root=kb_dir, repo_root=tmp_path)
|
|
|
|
|
|
@pytest.fixture
|
|
def search_result(kb_dir: Path, pages):
|
|
"""Run a query against the fixture kb and return the whole `SearchResult`.
|
|
|
|
For the tests that care about the limit and the total; most only want the
|
|
hits, and use `search` below.
|
|
"""
|
|
|
|
def _search(query: SearchQuery, backends=()):
|
|
return run_search(query, pages, list(backends), kb_dir)
|
|
|
|
return _search
|
|
|
|
|
|
@pytest.fixture
|
|
def search(search_result):
|
|
"""The hits alone, for tests whose subject is ranking or filtering."""
|
|
|
|
def _search(query: SearchQuery, backends=()):
|
|
return search_result(query, backends).hits
|
|
|
|
return _search
|
|
|
|
|
|
def _titles(hits):
|
|
return [hit.title for hit in hits]
|
|
|
|
|
|
def _q(*raw, **kwargs):
|
|
return SearchQuery(predicates=tuple(parse_predicate(r) for r in raw), **kwargs)
|
|
|
|
|
|
# --- predicate parsing ------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"raw,expected",
|
|
[
|
|
("entity_type=system", Predicate("entity_type", "=", "system")),
|
|
("summary~storage", Predicate("summary", "~", "storage")),
|
|
("weight>=0.8", Predicate("weight", ">=", "0.8")),
|
|
("weight<=0.8", Predicate("weight", "<=", "0.8")),
|
|
("weight>0.8", Predicate("weight", ">", "0.8")),
|
|
("modified<2026-08-01", Predicate("modified", "<", "2026-08-01")),
|
|
("summary:*", Predicate("summary", "exists", None)),
|
|
("!summary", Predicate("summary", "absent", None)),
|
|
],
|
|
)
|
|
def test_parse_predicate_forms(raw, expected):
|
|
assert parse_predicate(raw) == expected
|
|
|
|
|
|
def test_parse_predicate_prefers_longest_operator():
|
|
"""`>=` must be tried before `>`, or the value keeps a stray `=`."""
|
|
assert parse_predicate("weight>=0.8").value == "0.8"
|
|
|
|
|
|
@pytest.mark.parametrize("raw", ["", "nonsense", "=value", "field=", "!", ":*"])
|
|
def test_parse_predicate_rejects_malformed(raw):
|
|
with pytest.raises(PredicateError):
|
|
parse_predicate(raw)
|
|
|
|
|
|
# --- predicate evaluation ---------------------------------------------------
|
|
|
|
|
|
def test_exact_match_on_frontmatter_field(search):
|
|
assert _titles(search(_q("entity_type=system"))) == ["aurora", "Borealis"]
|
|
|
|
|
|
def test_membership_on_list_field(search):
|
|
assert _titles(search(_q("tags=server"))) == ["aurora"]
|
|
|
|
|
|
def test_substring_match_is_case_insensitive(search):
|
|
assert _titles(search(_q("summary~ZFS STORAGE"))) == ["aurora"]
|
|
|
|
|
|
def test_numeric_comparison(kb_dir, tmp_path):
|
|
"""`_compare` tries a float parse before falling back to lexicographic -
|
|
exercised here on an arbitrary numeric field, since no frontmatter field
|
|
in the schema is numeric any more (Gitea #60)."""
|
|
write_page(
|
|
kb_dir / "entities/tools/heavy.md",
|
|
{"type": "types/entity.md", "entity_type": "tool", "tags": [], "created": "2026-07-25",
|
|
"modified": "2026-07-25", "related": [], "sources": [], "weight": 0.9},
|
|
"\n# heavy\n",
|
|
)
|
|
write_page(
|
|
kb_dir / "entities/tools/light.md",
|
|
{"type": "types/entity.md", "entity_type": "tool", "tags": [], "created": "2026-07-25",
|
|
"modified": "2026-07-25", "related": [], "sources": [], "weight": 0.5},
|
|
"\n# light\n",
|
|
)
|
|
pages = load_pages_by_path(kb_dir, tmp_path)
|
|
matched = filters.apply_predicates(pages, (parse_predicate("weight>=0.9"),))
|
|
assert {page.title for page in matched.values()} == {"heavy"}
|
|
|
|
|
|
def test_date_comparison_handles_yaml_date_objects(search):
|
|
"""PyYAML parses `modified: 2026-08-02` into a date, not a string - the
|
|
comparison has to normalise it or it never matches."""
|
|
assert _titles(search(_q("modified>=2026-08-01"))) == ["Borealis"]
|
|
|
|
|
|
def test_exists_and_absent_are_complementary(search, pages):
|
|
present = set(_titles(search(_q("summary:*"))))
|
|
absent = set(_titles(search(_q("!summary"))))
|
|
assert present == {"aurora"}
|
|
assert not present & absent
|
|
assert present | absent == {page.title for page in pages.values()}
|
|
|
|
|
|
def test_multiple_predicates_are_anded(search):
|
|
assert _titles(search(_q("entity_type=system", "tags=server"))) == ["aurora"]
|
|
|
|
|
|
def test_virtual_fields_resolve_without_frontmatter(search):
|
|
assert _titles(search(_q("kind=concept"))) == ["Modbus"]
|
|
assert _titles(search(_q("collection=sources"))) == ["Source - Aurora"]
|
|
assert _titles(search(_q("subtype=tool"))) == ["gdeploy"]
|
|
|
|
|
|
def test_unknown_field_fails_loudly_instead_of_returning_nothing(search):
|
|
"""A typo must not read as 'the wiki has no such pages'."""
|
|
with pytest.raises(PredicateError) as exc:
|
|
search(_q("entitiy_type=system"))
|
|
assert "entitiy_type" in str(exc.value)
|
|
assert "entity_type" in str(exc.value) # the real field is offered
|
|
|
|
|
|
# --- ripgrep backend --------------------------------------------------------
|
|
|
|
|
|
def test_build_argv_never_uses_a_shell_and_defaults_to_fixed_strings():
|
|
argv = build_argv(SearchQuery(text="a; rm -rf /"), Path("/kb"))
|
|
assert argv[0] == "rg"
|
|
assert "--fixed-strings" in argv
|
|
# The query is one argv element, so shell metacharacters stay literal.
|
|
assert "a; rm -rf /" in argv
|
|
# `--` guards a query that starts with a dash.
|
|
assert argv.index("--") < argv.index("a; rm -rf /")
|
|
|
|
|
|
def test_build_argv_regex_is_opt_in():
|
|
assert "--fixed-strings" not in build_argv(SearchQuery(text="a.*b", regex=True), Path("/kb"))
|
|
|
|
|
|
def test_text_search_finds_body_matches(search, backend):
|
|
hits = search(SearchQuery(text="Industrial protocol"), [backend])
|
|
assert _titles(hits) == ["Modbus"]
|
|
assert hits[0].matches and hits[0].matches[0].line > 0
|
|
|
|
|
|
def test_exact_title_match_outranks_a_page_that_merely_mentions_it(search, backend):
|
|
hits = search(SearchQuery(text="aurora"), [backend])
|
|
assert hits[0].title == "aurora"
|
|
assert hits[0].score > hits[1].score
|
|
|
|
|
|
def test_hits_carry_frontmatter_so_the_page_need_not_be_opened(search, backend):
|
|
hit = search(SearchQuery(text="DocStore"), [backend])[0]
|
|
assert hit.kind == "entity"
|
|
assert hit.subtype == "system"
|
|
assert hit.collection == "entities"
|
|
assert "ZFS" in hit.summary
|
|
|
|
|
|
def test_generated_and_contract_files_never_surface(search, backend):
|
|
"""index.md mentions every page, so an unfiltered grep would rank it first."""
|
|
hits = search(SearchQuery(text="aurora", limit=0), [backend])
|
|
assert hits
|
|
assert all(not hit.path.endswith("index.md") for hit in hits)
|
|
assert all("COLLECTION.md" not in hit.path for hit in hits)
|
|
|
|
|
|
def test_text_and_predicates_combine(search, backend):
|
|
hits = search(
|
|
SearchQuery(text="aurora", predicates=(parse_predicate("kind=source"),), limit=0),
|
|
[backend],
|
|
)
|
|
assert _titles(hits) == ["Source - Aurora"]
|
|
|
|
|
|
def test_no_matches_is_an_empty_result_not_an_error(search, backend):
|
|
assert search(SearchQuery(text="zzzz-no-such-term"), [backend]) == []
|
|
|
|
|
|
def test_limit_and_sort(search):
|
|
hits = search(_q("kind=entity", sort="-modified"))
|
|
assert [hit.title for hit in hits] == ["Borealis", "aurora", "gdeploy"]
|
|
assert len(search(_q("kind=entity", limit=2))) == 2
|
|
|
|
|
|
def test_the_total_survives_the_limit_so_no_second_search_is_needed(search_result):
|
|
"""The caller cannot recover the total from a truncated list, and asking
|
|
again with `--limit 0` is a second full search to learn a number the first
|
|
one already had."""
|
|
capped = search_result(_q("kind=entity", limit=2))
|
|
|
|
assert len(capped.hits) == 2
|
|
assert capped.total == 3
|
|
assert capped.truncated is True
|
|
|
|
whole = search_result(_q("kind=entity"))
|
|
assert (whole.total, whole.truncated) == (3, False)
|
|
|
|
|
|
def test_sort_puts_missing_values_last():
|
|
hits = [
|
|
SearchHit(title="b", path="b", modified=None),
|
|
SearchHit(title="a", path="a", modified="2026-01-01"),
|
|
]
|
|
assert [hit.title for hit in sort_hits(hits, "modified")] == ["a", "b"]
|
|
|
|
|
|
# --- fusion and registry ----------------------------------------------------
|
|
|
|
|
|
def test_rrf_rewards_agreement_between_backends():
|
|
a = [SearchHit(title="x", path="x", backend="a"), SearchHit(title="y", path="y", backend="a")]
|
|
b = [SearchHit(title="z", path="z", backend="b"), SearchHit(title="x", path="x", backend="b")]
|
|
fused = reciprocal_rank_fusion([a, b])
|
|
assert fused[0].path == "x"
|
|
assert fused[0].backend == "a+b"
|
|
|
|
|
|
def test_resolve_defaults_to_rg_and_rejects_unknown():
|
|
assert [b.name for b in resolve(None)] == ["rg"]
|
|
assert [b.name for b in resolve("rg,rg")] == ["rg", "rg"]
|
|
with pytest.raises(UnknownBackend):
|
|
resolve("qmd")
|
|
|
|
|
|
# --- output -----------------------------------------------------------------
|
|
|
|
|
|
def _result(hits, total=None, limit=DEFAULT_LIMIT):
|
|
return SearchResult(hits=hits, total=total if total is not None else len(hits), limit=limit)
|
|
|
|
|
|
def test_render_table_is_compact_and_reports_the_count():
|
|
hit = SearchHit(title="aurora", path="kb/x.md", kind="entity", subtype="system",
|
|
summary="Server hosting DocStore", score=8.0, matches=[Match(3, "DocStore")])
|
|
out = render_table(_result([hit]), show_matches=False)
|
|
assert "aurora" in out and "entity/system" in out
|
|
assert "kb/x.md:3" not in out
|
|
assert "1 result(s)." in out
|
|
assert "kb/x.md:3" in render_table(_result([hit]), show_matches=True)
|
|
|
|
|
|
def test_render_table_says_so_when_nothing_matched():
|
|
assert render_table(_result([]), show_matches=False) == "No matches."
|
|
|
|
|
|
def test_every_hit_carries_the_path_so_the_page_can_be_opened():
|
|
"""The reason this format changed: the instructions tell an agent to read
|
|
the pages the search points at, and the table pointed nowhere. A session
|
|
that needed filenames ran `grep -rl` over kb/ for them - a second search
|
|
that can find no page this one missed."""
|
|
hit = SearchHit(title="aurora", path="kb/entities/systems/aurora.md", kind="entity")
|
|
assert "kb/entities/systems/aurora.md" in render_table(_result([hit]), show_matches=False)
|
|
|
|
|
|
def test_title_and_path_are_never_truncated_only_the_summary_is():
|
|
"""A clipped title is not an identifier. It is what `xref add`, `cite add`
|
|
and `touch` take as an argument, and the old fixed 34-char column cut four
|
|
of five hits in the report that prompted this."""
|
|
title = "Source - Pelletofenkondensator und Verkabelung Recherche"
|
|
path = f"kb/sources/llm-sessions/{title}.md"
|
|
hit = SearchHit(title=title, path=path, kind="source", summary="x" * 400)
|
|
line = render_table(_result([hit]), show_matches=False).splitlines()[0]
|
|
|
|
assert title in line
|
|
assert path in line
|
|
assert "…" in line, "the summary is still the one lossy field"
|
|
|
|
|
|
def test_a_hit_line_parses_into_its_five_fields_even_with_prose_pipes():
|
|
"""The separator has to survive a summary that contains one. It does,
|
|
because the summary is last: a `|` there cannot shift a column. A `|` in a
|
|
title is impossible - the wikilink syntax reserves it."""
|
|
hit = SearchHit(title="aurora", path="kb/x.md", kind="entity", subtype="system",
|
|
summary="Runs `a | b` nightly", score=8.0)
|
|
line = render_table(_result([hit]), show_matches=False).splitlines()[0]
|
|
score, kind, title, path, summary = line.split(" | ", 4)
|
|
|
|
assert (score, kind, title, path) == ("8.0", "entity/system", "aurora", "kb/x.md")
|
|
assert summary == "Runs `a | b` nightly"
|
|
|
|
|
|
def test_a_capped_result_says_what_it_capped():
|
|
"""`20 result(s).` on a query matching 182 pages is a completeness claim
|
|
the output was in no position to make, and the only way to find the real
|
|
number was a second, unlimited search."""
|
|
hits = [SearchHit(title=f"p{i}", path=f"kb/p{i}.md") for i in range(20)]
|
|
out = render_table(_result(hits, total=182, limit=20), show_matches=False)
|
|
|
|
assert "20 of 182 result(s)" in out
|
|
assert "--limit" in out
|
|
|
|
|
|
def test_an_uncapped_result_claims_nothing_about_a_limit():
|
|
hits = [SearchHit(title=f"p{i}", path=f"kb/p{i}.md") for i in range(3)]
|
|
assert "3 result(s)." in render_table(_result(hits), show_matches=False)
|
|
assert " of " not in render_table(_result(hits), show_matches=False)
|
|
|
|
|
|
def test_limit_zero_never_counts_as_truncated_however_large_the_corpus():
|
|
assert _result([SearchHit(title="a", path="a")] * 182, total=182, limit=0).truncated is False
|
|
|
|
|
|
def test_hit_serialises_for_json():
|
|
hit = SearchHit(title="a", path="kb/a.md", score=1.23456, matches=[Match(1, "x")])
|
|
assert hit.as_dict()["score"] == 1.235
|
|
assert hit.as_dict()["matches"] == [{"line": 1, "text": "x"}]
|
|
|
|
|
|
def test_known_fields_includes_virtual_and_real(pages):
|
|
fields = filters.known_fields(pages)
|
|
assert {"title", "kind", "subtype", "collection"} <= fields
|
|
assert {"entity_type", "sources", "tags"} <= fields
|
|
|
|
|
|
# --- Read-path limits (Gitea #33) -------------------------------------------
|
|
|
|
|
|
def test_a_user_regex_never_reaches_pythons_backtracking_engine():
|
|
"""Regression for the ReDoS. `(\\w+\\s?)+$` against 114 characters of
|
|
ordinary page text does not terminate in eight seconds under `re`; the
|
|
ranking helper must not evaluate it as a pattern at all.
|
|
|
|
Asserted by time *and* by outcome: a bound alone would pass if the branch
|
|
came back with a cheaper engine, and the outcome alone would pass while the
|
|
call still hung on a different pattern."""
|
|
haystack = (
|
|
"Longhorn is the distributed block storage layer for Kubernetes that "
|
|
"this cluster runs, replicated across three nodes and backed up nightly"
|
|
)
|
|
query = SearchQuery(text=r"(\w+\s?)+$", regex=True)
|
|
started = time.perf_counter()
|
|
assert ripgrep._contains(haystack, query) is False
|
|
assert time.perf_counter() - started < 0.5
|
|
|
|
|
|
def test_a_mostly_literal_regex_still_earns_its_title_boost():
|
|
"""What the deleted branch cost, and what it did not: the common case of a
|
|
pattern that happens to be plain text keeps ranking as before."""
|
|
assert ripgrep._contains("Longhorn", SearchQuery(text="longhorn", regex=True)) is True
|
|
|
|
|
|
def test_ripgrep_is_called_with_a_timeout(monkeypatch, kb_dir, tmp_path):
|
|
seen = {}
|
|
|
|
def fake_run(argv, **kwargs):
|
|
seen.update(kwargs)
|
|
return subprocess.CompletedProcess(argv, 1, "", "")
|
|
|
|
monkeypatch.setattr(ripgrep.subprocess, "run", fake_run)
|
|
RipgrepBackend(kb_dir, tmp_path).search(SearchQuery(text="x"), {})
|
|
assert seen["timeout"] == ripgrep.RIPGREP_TIMEOUT_SECONDS
|
|
|
|
|
|
def test_a_hanging_ripgrep_is_reported_as_a_failure_not_a_hang(monkeypatch, kb_dir, tmp_path):
|
|
def fake_run(argv, **kwargs):
|
|
raise subprocess.TimeoutExpired(argv, kwargs["timeout"])
|
|
|
|
monkeypatch.setattr(ripgrep.subprocess, "run", fake_run)
|
|
with pytest.raises(ripgrep.RipgrepFailed) as excinfo:
|
|
RipgrepBackend(kb_dir, tmp_path).search(SearchQuery(text="x"), {})
|
|
assert "did not finish" in str(excinfo.value)
|
|
|
|
|
|
def test_a_page_with_broken_frontmatter_is_reported_not_lost(kb_dir, tmp_path):
|
|
"""It matches no positive predicate - including the sweeps meant to find
|
|
pages in exactly that state (`!sources`, `provenance=general`) - so
|
|
silence reads as 'did not match'. The page has to be nameable."""
|
|
broken = kb_dir / "entities" / "Broken.md"
|
|
broken.write_text("---\ntype: [unclosed\n---\n\n# Broken\n", encoding="utf-8")
|
|
pages = load_pages_by_path(kb_dir, tmp_path)
|
|
key = page_key(broken, tmp_path)
|
|
|
|
assert pages[key].frontmatter == {}
|
|
assert filters.apply_predicates(pages, (parse_predicate("entity_type=system"),)) .get(key) is None
|
|
|
|
reported = unreadable_pages(pages)
|
|
assert [entry["path"] for entry in reported] == [key]
|
|
assert "invalid YAML" in reported[0]["reason"]
|
|
|
|
|
|
def test_an_empty_frontmatter_block_is_not_reported_as_unreadable(kb_dir, tmp_path):
|
|
"""A page may legitimately carry an empty block - there are no fields to
|
|
lose, so there is nothing the caller was not told about. `lint` still has
|
|
an opinion about it; search does not."""
|
|
(kb_dir / "entities" / "Bare.md").write_text("---\n\n---\n\n# Bare\n", encoding="utf-8")
|
|
assert unreadable_pages(load_pages_by_path(kb_dir, tmp_path)) == []
|
|
|
|
|
|
def test_a_page_with_no_frontmatter_at_all_is_reported(kb_dir, tmp_path):
|
|
"""Unlike an empty block, this page has no `type:` either - it cannot match
|
|
a predicate, and nothing else would say so."""
|
|
(kb_dir / "entities" / "Naked.md").write_text("# Naked\n\nProse only.\n", encoding="utf-8")
|
|
reported = unreadable_pages(load_pages_by_path(kb_dir, tmp_path))
|
|
assert [entry["path"] for entry in reported] == ["kb/entities/Naked.md"]
|