- query language: substrings, phrases, /regex/, AND/OR/NOT (+ UND/ODER/NICHT), parentheses; case-insensitive; ReDoS protection via regex timeouts; syntax errors with stable code and position - SearchEngine over in-memory blocks: filters, predicate, grouping, sorting, paging, time budget, automatic reload, highlighted snippets - Markdown loader with split modes file/heading/list/paragraph - grep-like CLI (mdgrep), JSON output, grep exit codes - Flask blueprint/app with JSON API and widget assets, dev server - framework-free JS widget: live search, URL state, filters, paging, syntax help, en/de labels, CSS custom properties for theming Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
116 lines
4.1 KiB
Python
116 lines
4.1 KiB
Python
import pytest
|
|
|
|
from mdgrep import Block, QuerySyntaxError, SearchEngine
|
|
|
|
|
|
def blk(i, text, **meta):
|
|
return Block(id=str(i), text=text, title=f"T{i} {text.split()[0]}", meta=meta)
|
|
|
|
|
|
BLOCKS = [
|
|
blk(1, "rust compiler news", lang="en", date="2026-01-01", url="u1"),
|
|
blk(2, "rust compiler news again", lang="en", date="2026-02-01", url="u1"),
|
|
blk(3, "Rust Übersetzer", lang="de", date="2026-01-15", url="u1"),
|
|
blk(4, "go generics", lang="en", date="2026-03-01", url="u2"),
|
|
blk(5, "python typing and rust bindings", lang="en", date="2025-12-01", url="u3"),
|
|
]
|
|
|
|
|
|
def ids(result):
|
|
return [g.block.id for g in result.groups]
|
|
|
|
|
|
def test_basic_search_original_order():
|
|
res = SearchEngine(BLOCKS).search("rust")
|
|
assert ids(res) == ["1", "2", "3", "5"]
|
|
assert res.total_blocks == 4 and res.total_groups == 4
|
|
|
|
|
|
def test_filters_single_and_multiple_values():
|
|
eng = SearchEngine(BLOCKS)
|
|
assert ids(eng.search("rust", filters={"lang": "de"})) == ["3"]
|
|
assert ids(eng.search("rust", filters={"lang": ["de", "en"]})) == ["1", "2", "3", "5"]
|
|
assert ids(eng.search("rust", filters={"lang": ""})) == ["1", "2", "3", "5"]
|
|
|
|
|
|
def test_predicate():
|
|
res = SearchEngine(BLOCKS).search("rust", predicate=lambda b: b.id != "1")
|
|
assert ids(res) == ["2", "3", "5"]
|
|
|
|
|
|
def test_sort_newest_first():
|
|
res = SearchEngine(BLOCKS).search("rust", sort_key=lambda b: b.meta["date"], reverse=True)
|
|
assert ids(res) == ["2", "3", "1", "5"]
|
|
|
|
|
|
def test_grouping_uses_first_block_after_sorting():
|
|
res = SearchEngine(BLOCKS).search(
|
|
"rust", sort_key=lambda b: b.meta["date"], reverse=True,
|
|
group_by=lambda b: (b.meta["lang"], b.meta["url"]))
|
|
assert ids(res) == ["2", "3", "5"]
|
|
assert [b.id for b in res.groups[0].also] == ["1"]
|
|
assert res.total_groups == 3 and res.total_blocks == 4
|
|
|
|
|
|
def test_title_is_searched():
|
|
eng = SearchEngine([Block("t", "body text", title="Special Title")])
|
|
res = eng.search("special")
|
|
assert ids(res) == ["t"]
|
|
assert res.groups[0].title_segments == [{"t": "Special", "hit": True}, {"t": " Title", "hit": False}]
|
|
assert res.groups[0].hits == 0 # hits counts snippet matches
|
|
|
|
|
|
def test_paging():
|
|
eng = SearchEngine(BLOCKS)
|
|
first = eng.search("rust", limit=2)
|
|
assert ids(first) == ["1", "2"] and first.next_offset == 2
|
|
second = eng.search("rust", limit=2, offset=first.next_offset)
|
|
assert ids(second) == ["3", "5"] and second.next_offset is None
|
|
|
|
|
|
def test_segments_and_title_segments():
|
|
res = SearchEngine(BLOCKS).search("compiler")
|
|
d = res.to_dict()["results"][0]["block"]
|
|
assert {"t": "compiler", "hit": True} in d["segments"]
|
|
assert d["hits"] == 1
|
|
assert d["title_segments"] == [{"t": "T1 rust", "hit": False}]
|
|
|
|
|
|
def test_to_dict_contract():
|
|
d = SearchEngine(BLOCKS).search("go", limit=10).to_dict()
|
|
assert set(d) == {"query", "total_groups", "total_blocks", "offset", "limit",
|
|
"next_offset", "truncated", "elapsed_ms", "results"}
|
|
r = d["results"][0]
|
|
assert set(r) == {"block", "also"}
|
|
assert r["block"]["meta"] == {"lang": "en", "date": "2026-03-01", "url": "u2"}
|
|
|
|
|
|
def test_syntax_error_propagates():
|
|
with pytest.raises(QuerySyntaxError):
|
|
SearchEngine(BLOCKS).search("(rust")
|
|
|
|
|
|
def test_time_budget_truncates():
|
|
blocks = [Block(id=str(i), text="a" * 30 + "!") for i in range(50)]
|
|
res = SearchEngine(blocks).search("/(a|aa)+$/", time_budget=0.3)
|
|
assert res.truncated
|
|
assert res.elapsed_ms < 3000
|
|
|
|
|
|
def test_reload_on_token_change():
|
|
state = {"token": 1, "blocks": [Block("a", "first")]}
|
|
eng = SearchEngine(lambda: list(state["blocks"]), reload_check=lambda: state["token"],
|
|
reload_interval=0)
|
|
assert ids(eng.search("first")) == ["a"]
|
|
state["blocks"] = [Block("b", "second")]
|
|
assert ids(eng.search("second")) == [] # token unchanged → no reload
|
|
state["token"] = 2
|
|
assert ids(eng.search("second")) == ["b"]
|
|
|
|
|
|
def test_reload_interval_respected():
|
|
calls = []
|
|
eng = SearchEngine(lambda: [], reload_check=lambda: calls.append(1) or 1, reload_interval=3600)
|
|
eng.search("x")
|
|
eng.search("x")
|
|
assert len(calls) == 1 # only the initial check
|