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 Match, Predicate, SearchHit, SearchQuery 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(kb_dir: Path, pages): """Run a query against the fixture kb rather than the real one.""" def _search(query: SearchQuery, backends=()): return run_search(query, pages, list(backends), kb_dir) 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_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 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([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([hit], show_matches=True) def test_render_table_says_so_when_nothing_matched(): assert render_table([], show_matches=False) == "No matches." 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"]