feat: mdgrep 0.1.0 – boolean full-text search for Markdown

- 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>
This commit is contained in:
dschlueter 2026-10-12 19:29:44 +02:00
commit 9c7b7e81f9
6 changed files with 2642 additions and 0 deletions

60
tests/test_cli.py Normal file
View file

@ -0,0 +1,60 @@
import json
import pytest
from mdgrep.cli import main
@pytest.fixture
def docs(tmp_path):
(tmp_path / "a.md").write_text("# A\n\nRust is fast.\n\nGo is simple.\n", encoding="utf-8")
(tmp_path / "b.md").write_text("- rust crypto wallet\n- python typing\n", encoding="utf-8")
return tmp_path
def run(capsys, *args):
code = main([*args, "--color", "never"])
out, err = capsys.readouterr()
return code, out, err
def test_match_output_and_exit_code(capsys, docs):
code, out, _ = run(capsys, "rust -crypto", str(docs))
assert code == 0
assert f"{docs / 'a.md'}:3" in out
assert "Rust is fast." in out
assert "crypto" not in out
def test_no_match_exit_1(capsys, docs):
code, out, _ = run(capsys, "haskell", str(docs))
assert code == 1 and out == ""
def test_syntax_error_exit_2_with_caret(capsys, docs):
code, _, err = run(capsys, "(rust", str(docs))
assert code == 2
assert "Missing closing parenthesis" in err and "^" in err
def test_count_and_files(capsys, docs):
assert run(capsys, "rust OR python", str(docs))[0] == 0
_, out, _ = run(capsys, "rust OR python", str(docs), "--split", "list", "-c")
assert out.strip() == "3"
_, out, _ = run(capsys, "rust", str(docs), "-l")
assert out.split() == [str(docs / "a.md"), str(docs / "b.md")]
def test_json_output(capsys, docs):
code, out, _ = run(capsys, "/fa.t/", str(docs), "--json")
data = json.loads(out)
assert code == 0 and data["total_blocks"] == 1
block = data["results"][0]["block"]
assert {"t": "fast", "hit": True} in block["segments"]
assert "markdown" not in block["meta"]
def test_color_highlight(capsys, docs):
main(["rust", str(docs / "a.md"), "--color", "always"])
out = capsys.readouterr().out
assert "\033[1;31mRust\033[0m" in out

116
tests/test_engine.py Normal file
View file

@ -0,0 +1,116 @@
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

119
tests/test_markdown.py Normal file
View file

@ -0,0 +1,119 @@
import pytest
from mdgrep.markdown import load_markdown, markdown_to_text, slugify, split_markdown
DOC = """# Report
## Edition
1. **[Rust 2.0 released](https://example.com/rust)** (100 points by alice)
The Rust team announced a *major* release with `async` traits.
2. **[Go generics](https://example.com/go)** (50 points by bob)
Generics in Go are now __fast__.
## Insights
1. **Memory safety wins**
- **Trend:** Rust adoption grows.
- **Why:** fewer bugs.
2. **Typed languages**
lazy continuation line
A closing paragraph
spanning two lines.
---
*Footer text*
```
# not a heading
- not an item
```
"""
def kinds(blocks):
return [(b.meta["kind"], b.meta["section"], b.line) for b in blocks]
def test_split_list():
blocks = split_markdown(DOC, source="doc.md", split="list")
assert kinds(blocks) == [
("item", "Edition", 4),
("item", "Edition", 8),
("item", "Insights", 14),
("item", "Insights", 19),
("paragraph", "Insights", 22),
("paragraph", "Insights", 27),
("paragraph", "Insights", 29),
]
first = blocks[0]
assert first.text == ("Rust 2.0 released (100 points by alice)\n\n"
"The Rust team announced a major release with async traits.")
assert first.title == ""
assert first.id == "doc.md:4" and first.source == "doc.md"
assert first.meta["markdown"].startswith("1. **[Rust 2.0 released]")
assert "Trend: Rust adoption grows." in blocks[2].text
assert "lazy continuation line" in blocks[3].text
assert "not a heading" in blocks[6].text
def test_split_heading():
blocks = split_markdown(DOC, split="heading")
assert [(b.title, b.anchor) for b in blocks] == [
("Report", "report"), ("Edition", "edition"), ("Insights", "insights")]
assert "Go generics" in blocks[1].text
assert not blocks[1].text.startswith("Edition")
def test_split_paragraph():
blocks = split_markdown(DOC, split="paragraph")
texts = [b.text.split("\n")[0] for b in blocks]
assert "Rust 2.0 released (100 points by alice)" in texts
assert "The Rust team announced a major release with async traits." in texts
assert "A closing paragraph" in texts
fence = [b for b in blocks if "not a heading" in b.text]
assert len(fence) == 1 and "not an item" in fence[0].text
def test_split_file():
blocks = split_markdown(DOC, source="x.md", split="file")
assert len(blocks) == 1 and blocks[0].title == "Report" and blocks[0].line == 1
def test_invalid_split_mode():
with pytest.raises(ValueError):
split_markdown("x", split="bogus")
def test_markdown_to_text():
md = ("- **Bold** and *italic* and __b2__ and `code` and ~~gone~~\n"
" [link](https://x.y/a_(b)) ![alt](img.png) <https://auto.link>\n"
"> quote with snake_case_name")
assert markdown_to_text(md) == (
"Bold and italic and b2 and code and gone\n"
"link alt https://auto.link\n"
"quote with snake_case_name")
def test_slugify():
assert slugify("AI/ML Insights & Trends") == "aiml-insights--trends"
assert slugify("Über **uns**") == "über-uns"
def test_load_markdown_directory_and_meta(tmp_path):
(tmp_path / "sub").mkdir()
(tmp_path / "a.md").write_text("first para\n\nsecond para\n", encoding="utf-8")
(tmp_path / "sub" / "b.md").write_text("- item one\n- item two\n", encoding="utf-8")
(tmp_path / "ignore.txt").write_text("nope", encoding="utf-8")
blocks = load_markdown(tmp_path, split="list", meta=lambda p: {"file": p.name})
assert [(b.meta["file"], b.text) for b in blocks] == [
("a.md", "first para"), ("a.md", "second para"),
("b.md", "item one"), ("b.md", "item two")]
rel = load_markdown(tmp_path, relative_to=tmp_path)
assert [b.id for b in rel] == ["a.md:1", "a.md:3", "sub/b.md:1"]

187
tests/test_query.py Normal file
View file

@ -0,0 +1,187 @@
import time
import pytest
from mdgrep import QuerySyntaxError, QueryTimeout, compile_query
def m(query, text):
return compile_query(query).matches(text)
# ── Terms ────────────────────────────────────────────────────────────────
def test_substring_case_insensitive():
assert m("rust", "The Rustacean book")
assert m("RUST", "trusted")
assert not m("rust", "golang")
def test_umlauts_case_insensitive():
assert m("übersetzung", "Die ÜBERSETZUNG ist fertig")
assert m("ÄRGER", "kein ärger")
def test_phrase_matches_any_whitespace():
assert m('"open source"', "Open\n Source software")
assert not m('"open source"', "open the source")
def test_phrase_escaped_quote():
assert m(r'"say \"hi\""', 'they say "hi" loudly')
def test_regex():
assert m(r"/llms?\b/", "Large LLMs today")
assert m(r"/gpt-\d+/", "GPT-5 release")
assert not m(r"/^foo$/", "foobar")
def test_regex_escaped_slash():
assert m(r"/tcp\/ip/", "TCP/IP stack")
def test_words_with_special_characters():
assert m("AT&T", "at&t merger")
assert m("TCP/IP", "the tcp/ip suite")
assert m("e-mail", "send an E-Mail")
def test_lowercase_keywords_are_terms():
assert m("and", "this AND that")
assert not m("and", "nothing here")
# ── Boolean logic ────────────────────────────────────────────────────────
@pytest.mark.parametrize("query", ["a b", "a AND b", "a UND b", "a & b", "a && b"])
def test_and_variants(query):
assert compile_query(query).matches("a b")
assert not compile_query(query).matches("a only")
@pytest.mark.parametrize("query", ["a OR b", "a ODER b", "a | b", "a || b"])
def test_or_variants(query):
assert compile_query(query).matches("b only")
assert not compile_query(query).matches("nothing")
@pytest.mark.parametrize("query", ["NOT x", "NICHT x", "-x", "!x"])
def test_not_variants(query):
assert compile_query(query).matches("abc")
assert not compile_query(query).matches("xyz")
def test_precedence_and_binds_tighter_than_or():
q = compile_query("alpha OR beta AND gamma") # alpha OR (beta AND gamma)
assert q.matches("alpha")
assert not q.matches("beta")
assert q.matches("beta gamma")
def test_parentheses_override_precedence():
q = compile_query("(alpha OR beta) AND gamma")
assert not q.matches("alpha")
assert q.matches("alpha gamma")
assert q.matches("beta gamma")
def test_not_binds_tighter_than_and():
q = compile_query("NOT alpha beta") # (NOT alpha) AND beta
assert q.matches("beta")
assert not q.matches("alpha beta")
def test_nested_groups_and_negation():
q = compile_query("(rust OR go) AND NOT (crypto OR /blockchain/)")
assert q.matches("Rust compiler")
assert not q.matches("Go crypto wallet")
assert not q.matches("rust blockchain")
def test_double_negation():
assert m("NOT NOT foo", "foo")
# ── Spans (highlighting) ─────────────────────────────────────────────────
def test_spans_merged_and_sorted():
q = compile_query("open OR source")
assert q.spans("open source open") == [(0, 4), (5, 11), (12, 16)]
def test_spans_exclude_negated_terms():
q = compile_query("rust -crypto")
assert q.spans("rust and crypto") == [(0, 4)]
def test_spans_double_negation_counts_as_positive():
q = compile_query("NOT NOT foo")
assert q.spans("a foo") == [(2, 5)]
def test_spans_ignore_empty_regex_matches():
q = compile_query("/x*/")
assert q.spans("abc") == []
assert q.spans("axxb") == [(1, 3)]
def test_spans_overlapping_merged():
q = compile_query("abc OR bcd")
assert q.spans("abcd") == [(0, 4)]
# ── Errors ───────────────────────────────────────────────────────────────
@pytest.mark.parametrize("query,code,position", [
("", "empty_query", 0),
(" ", "empty_query", 0),
("(a OR b", "unclosed_paren", 0),
("a OR b)", "unexpected_paren", 6),
("a OR", "missing_operand", 4),
("OR a", "missing_operand", 0),
("a AND OR b", "missing_operand", 6),
("()", "empty_group", 0),
('"open', "unterminated_phrase", 0),
("x /abc", "unterminated_regex", 2),
("x /a(b/", "invalid_regex", 2),
("NOT", "missing_operand", 3),
])
def test_syntax_errors(query, code, position):
with pytest.raises(QuerySyntaxError) as exc:
compile_query(query)
assert exc.value.code == code
assert exc.value.position == position
assert exc.value.to_dict()["code"] == code
def test_limits():
with pytest.raises(QuerySyntaxError) as exc:
compile_query("a" * 20, max_length=10)
assert exc.value.code == "too_long"
with pytest.raises(QuerySyntaxError) as exc:
compile_query("a b c d", max_terms=3)
assert exc.value.code == "too_many_terms"
# ── Timeouts (ReDoS protection) ──────────────────────────────────────────
def test_catastrophic_regex_times_out():
q = compile_query("/(a|aa)+$/")
text = "a" * 36 + "!"
start = time.monotonic()
with pytest.raises(QueryTimeout):
q.matches(text, timeout=0.2)
assert time.monotonic() - start < 2
def test_spans_timeout():
q = compile_query("/(a|aa)+$/")
with pytest.raises(QueryTimeout):
q.spans("a" * 36 + "!", timeout=0.2)
def test_expired_deadline_raises():
q = compile_query("foo")
with pytest.raises(QueryTimeout):
q.matches("foo", deadline=time.monotonic() - 1)

59
tests/test_snippets.py Normal file
View file

@ -0,0 +1,59 @@
from mdgrep.snippets import ELLIPSIS, segments
def join(segs):
return "".join(s["t"] for s in segs)
def test_short_text_full_with_hits():
segs = segments("hello world", [(6, 11)])
assert segs == [{"t": "hello ", "hit": False}, {"t": "world", "hit": True}]
def test_no_spans_returns_whole_text():
assert segments("abc", []) == [{"t": "abc", "hit": False}]
def test_long_text_windowed_around_hit():
text = " ".join(f"w{i}" for i in range(400))
hit_start = text.index("w200")
segs = segments(text, [(hit_start, hit_start + 4)], max_chars=100, context=30)
assert segs[0]["t"] == ELLIPSIS and segs[-1]["t"] == ELLIPSIS
assert any(s["hit"] and s["t"] == "w200" for s in segs)
assert len(join(segs)) < 150
def test_windows_snap_to_word_boundaries():
text = "alpha beta gamma delta epsilon zeta eta theta iota kappa lambda"
start = text.index("epsilon")
segs = segments(text, [(start, start + 7)], max_chars=20, context=8)
inner = join(s for s in segs if s["t"] != ELLIPSIS)
assert not inner.startswith("lta") and inner.split()[0] in ("gamma", "delta")
def test_multiple_hits_separate_windows():
text = "x " * 300 + "foo " + "y " * 300 + "bar " + "z " * 300
f, b = text.index("foo"), text.index("bar")
segs = segments(text, [(f, f + 3), (b, b + 3)], max_chars=200, context=20)
hits = [s["t"] for s in segs if s["hit"]]
assert hits == ["foo", "bar"]
assert sum(1 for s in segs if s["t"] == ELLIPSIS) == 3
def test_budget_limits_windows_but_keeps_first_hit():
text = ("filler " * 50 + "hit ") * 10
spans = []
pos = 0
while (i := text.find("hit", pos)) != -1:
spans.append((i, i + 3))
pos = i + 3
segs = segments(text, spans, max_chars=60, context=20)
assert sum(1 for s in segs if s["hit"]) >= 1
assert len(join(segs)) < 200
def test_no_hits_long_text_shows_beginning():
text = "start " + "word " * 200
segs = segments(text, [], max_chars=50)
assert segs[0]["t"].startswith("start")
assert segs[-1]["t"] == ELLIPSIS

80
tests/test_web.py Normal file
View file

@ -0,0 +1,80 @@
import pytest
from mdgrep import Block, SearchEngine
from mdgrep.web import create_app
BLOCKS = [
Block("1", "rust news", title="Rust", meta={"lang": "en", "url": "u1", "markdown": "**rust**"}),
Block("2", "rust nachrichten", title="Rust", meta={"lang": "de", "url": "u1", "markdown": "x"}),
Block("3", "go news", title="Go", meta={"lang": "en", "url": "u2", "markdown": "y"}),
Block("4", "rust again", title="Rust", meta={"lang": "en", "url": "u1", "markdown": "z"}),
]
@pytest.fixture
def client():
app = create_app(SearchEngine(BLOCKS), filters=["lang"],
group_by=lambda b: (b.meta["lang"], b.meta["url"]))
return app.test_client()
def test_search_ok(client):
data = client.get("/search?q=rust").get_json()
assert data["total_blocks"] == 3 and data["total_groups"] == 2
first = data["results"][0]
assert first["block"]["id"] == "1" and [b["id"] for b in first["also"]] == ["4"]
assert "markdown" not in first["block"]["meta"]
assert "markdown" not in first["also"][0]["meta"]
def test_filters(client):
assert client.get("/search?q=rust&lang=de").get_json()["total_blocks"] == 1
assert client.get("/search?q=rust&lang=all").get_json()["total_blocks"] == 3
assert client.get("/search?q=rust&lang=de,en").get_json()["total_blocks"] == 3
def test_paging_and_limit_clamping(client):
data = client.get("/search?q=news OR rust&limit=1").get_json()
assert len(data["results"]) == 1 and data["next_offset"] == 1
data = client.get("/search?q=rust&limit=9999&offset=-5").get_json()
assert data["limit"] == 100 and data["offset"] == 0
def test_empty_query(client):
data = client.get("/search?q=").get_json()
assert data["results"] == [] and data["total_groups"] == 0
def test_syntax_error_400(client):
resp = client.get("/search?q=(rust")
assert resp.status_code == 400
assert resp.get_json()["error"] == {"code": "unclosed_paren",
"message": "Missing closing parenthesis", "position": 0}
def test_too_long_query(client):
resp = client.get("/search?q=" + "a" * 400)
assert resp.status_code == 400 and resp.get_json()["error"]["code"] == "too_long"
def test_unicode_not_escaped(client):
app = create_app(SearchEngine([Block("x", "Größe")]))
body = app.test_client().get("/search?q=größe").get_data(as_text=True)
assert "Größe" in body
def test_widget_assets_served(client):
assert client.get("/mdgrep-static/mdgrep-widget.js").status_code == 200
assert client.get("/mdgrep-static/mdgrep-widget.css").status_code == 200
def test_static_dir_and_prefix(tmp_path):
(tmp_path / "index.html").write_text("<h1>home</h1>")
(tmp_path / "sub").mkdir()
(tmp_path / "sub" / "index.html").write_text("sub page")
app = create_app(SearchEngine(BLOCKS), url_prefix="/api", static_dir=tmp_path)
c = app.test_client()
assert c.get("/").get_data(as_text=True) == "<h1>home</h1>"
assert c.get("/sub/").get_data(as_text=True) == "sub page"
assert c.get("/api/search?q=go").get_json()["total_blocks"] == 1
assert c.get("/../etc/passwd").status_code == 404