tgit/tests/logs/test_grep.py

import time

import pytest

from app.logs.grep import (
    MAX_PATTERN,
    GrepError,
    GrepErrorCode,
    GrepQuery,
    bre_to_ere,
    parse_grep,
    run,
    to_command,
)

LOG = "\n".join(
    [
        "08:00 start",
        "08:01 connect db",
        "08:02 ERROR connection refused",
        "08:03 retry",
        "08:04 ok",
        "08:05 idle",
        "08:06 idle",
        "08:07 error: timeout",
        "08:08 done",
    ]
)


@pytest.mark.parametrize(
    ("expr", "expected"),
    [
        ("error", GrepQuery("error")),
        ("grep -i error", GrepQuery("error", ignore_case=True)),
        ("-iv 'two words'", GrepQuery("two words", ignore_case=True, invert=True)),
        ("-E -c 'a|b'", GrepQuery("a|b", extended=True, count=True)),
        ("-C 2 x", GrepQuery("x", context=2)),
        ("-C2 x", GrepQuery("x", context=2)),
        ("-iC3 x", GrepQuery("x", ignore_case=True, context=3)),
        ("-- -v", GrepQuery("-v")),
        ('"привет мир"', GrepQuery("привет мир")),
    ],
)
def test_parse(expr: str, expected: GrepQuery) -> None:
    assert parse_grep(expr) == expected


@pytest.mark.parametrize(
    ("expr", "code"),
    [
        ("", GrepErrorCode.EMPTY),
        ("-i", GrepErrorCode.EMPTY),
        ("'unclosed", GrepErrorCode.QUOTES),
        ("-r error", GrepErrorCode.FLAG),
        ("--include=*.py x", GrepErrorCode.FLAG),
        ("-C x y", GrepErrorCode.CONTEXT),
        ("-C 99 y", GrepErrorCode.CONTEXT),
        ("-C", GrepErrorCode.CONTEXT),
        ("error /etc/passwd", GrepErrorCode.EXTRA),
        ("x" * (MAX_PATTERN + 1), GrepErrorCode.TOO_LONG),
        ("-E '(unclosed'", GrepErrorCode.REGEX),
    ],
)
def test_parse_errors(expr: str, code: GrepErrorCode) -> None:
    with pytest.raises(GrepError) as e:
        parse_grep(expr)
    assert e.value.code is code


def test_shell_is_never_involved() -> None:
    # metacharacters are just pattern text
    q = parse_grep("'$(rm -rf /); `id`'")
    assert q.pattern == "$(rm -rf /); `id`"
    assert to_command(q) == "grep '$(rm -rf /); `id`'"


def test_basic_vs_extended_regex() -> None:
    assert bre_to_ere("a+b") == r"a\+b"
    assert bre_to_ere(r"a\+b") == "a+b"
    assert bre_to_ere(r"\(x\|y\)") == "(x|y)"
    assert bre_to_ere("[+?]x") == "[+?]x"
    assert bre_to_ere("[]a]+") == r"[]a]\+"
    assert run(parse_grep("1+1"), "1+1\n11").lines == ["1+1"]
    assert run(parse_grep("-E 1+1"), "1+1\n11").lines == ["11"]


def test_ignore_case_and_invert() -> None:
    assert run(parse_grep("-i error"), LOG).lines == ["08:02 ERROR connection refused", "08:07 error: timeout"]
    assert run(parse_grep("-v idle"), LOG).matched == 7


def test_count() -> None:
    result = run(parse_grep("-ci error"), LOG)
    assert result.matched == 2
    assert result.lines == []


def test_context_merges_and_separates_groups() -> None:
    result = run(parse_grep("-i -C 1 error"), LOG)
    assert result.lines == [
        "08:01 connect db",
        "08:02 ERROR connection refused",
        "08:03 retry",
        "--",
        "08:06 idle",
        "08:07 error: timeout",
        "08:08 done",
    ]
    overlapping = run(parse_grep("-C 2 idle"), LOG)
    assert overlapping.lines == LOG.splitlines()[3:]  # 08:03…08:08, no duplicates, no separator


def test_output_is_capped() -> None:
    text = "\n".join(f"line {i}" for i in range(2000))
    result = run(parse_grep("line"), text, max_lines=500)
    assert len(result.lines) == 500
    assert result.truncated
    assert result.matched == 2000  # counting goes on


@pytest.mark.parametrize("pattern", ["(a+)+$", "(a|aa)+$"])  # regex optimizes the first; the second really blows up
def test_redos_does_not_hang(pattern: str) -> None:
    evil = parse_grep(f"-E '{pattern}'")
    text = "\n".join(["a" * 40 + "!"] * 200 + ["fine"])
    started = time.monotonic()

    result = run(evil, text, budget=1.0)

    assert time.monotonic() - started < 2.0
    assert result.matched == 0


def test_big_input() -> None:
    text = "\n".join(f"{i} request ok" if i % 1000 else f"{i} ERROR" for i in range(1, 100_001))
    result = run(parse_grep("ERROR"), text)
    assert result.matched == 100
    assert not result.partial


def test_to_command() -> None:
    assert to_command(parse_grep("-i -C 2 -E 'conn(ect)?'")) == "grep -iE -C 2 'conn(ect)?'"
    assert to_command(GrepQuery("-x")) == "grep -- -x"
    assert parse_grep(to_command(parse_grep("-vc 'a b'"))) == parse_grep("-vc 'a b'")


def test_catastrophic_lines_are_skipped_then_budget_stops() -> None:
    text = "\n".join(["a" * 40 + "!"] * 100)
    result = run(parse_grep("-E '(a|aa)+$'"), text, budget=0.3)
    assert result.skipped >= 1
    assert result.partial