chemenu/tools/chemenu/tests/test_search.py

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


@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")),
        ("confidence>=0.8", Predicate("confidence", ">=", "0.8")),
        ("confidence<=0.8", Predicate("confidence", "<=", "0.8")),
        ("confidence>0.8", Predicate("confidence", ">", "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("confidence>=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(search):
    assert _titles(search(_q("confidence>=0.9"))) == ["aurora", "Borealis"]


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 hit.confidence == 0.9
    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="-confidence"))
    assert [hit.confidence for hit in hits] == [0.9, 0.9, 0.8]
    assert len(search(_q("kind=entity", limit=2))) == 2


def test_sort_puts_missing_values_last():
    hits = [
        SearchHit(title="b", path="b", confidence=None),
        SearchHit(title="a", path="a", confidence=0.5),
    ]
    assert [hit.title for hit in sort_hits(hits, "confidence")] == ["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", "confidence", "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 low-confidence sweep
    meant to find pages in exactly that state - 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("confidence<0.6"),)) .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"]